X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=dmaap-mediator-producer%2Fmain.go;h=65a84a2150b70fcfcf1daa95859daf61194e7354;hb=43b5864e8e5d8fad832c308301dd81ed02fbc2f5;hp=15207ec713af51ab0385c96f96974c2677bf5b00;hpb=fe1aa1e639246094905f6a9f09a051084bf431ed;p=nonrtric.git diff --git a/dmaap-mediator-producer/main.go b/dmaap-mediator-producer/main.go index 15207ec7..65a84a21 100644 --- a/dmaap-mediator-producer/main.go +++ b/dmaap-mediator-producer/main.go @@ -21,71 +21,147 @@ package main import ( + "crypto/tls" + "encoding/json" "fmt" "net/http" - "sync" + "time" + "github.com/gorilla/mux" log "github.com/sirupsen/logrus" + _ "oransc.org/nonrtric/dmaapmediatorproducer/api" "oransc.org/nonrtric/dmaapmediatorproducer/internal/config" "oransc.org/nonrtric/dmaapmediatorproducer/internal/jobs" + "oransc.org/nonrtric/dmaapmediatorproducer/internal/kafkaclient" + "oransc.org/nonrtric/dmaapmediatorproducer/internal/restclient" "oransc.org/nonrtric/dmaapmediatorproducer/internal/server" + + httpSwagger "github.com/swaggo/http-swagger" ) var configuration *config.Config -var callbackAddress string +var registered bool func init() { configuration = config.New() - if loglevel, err := log.ParseLevel(configuration.LogLevel); err == nil { - log.SetLevel(loglevel) - } else { - log.Warnf("Invalid log level: %v. Log level will be Info!", configuration.LogLevel) - } +} + +// @title DMaaP Mediator Producer +// @version 1.1.0 +// @license.name Apache 2.0 +// @license.url http://www.apache.org/licenses/LICENSE-2.0.html + +func main() { + log.SetLevel(configuration.LogLevel) log.Debug("Initializing DMaaP Mediator Producer") - if configuration.InfoProducerHost == "" { - log.Fatal("Missing INFO_PRODUCER_SUPERVISION_CALLBACK_HOST") + log.Debug("Using configuration: ", configuration) + if err := validateConfiguration(configuration); err != nil { + log.Fatalf("Stopping producer due to error: %v", err) } - callbackAddress = fmt.Sprintf("%v:%v", configuration.InfoProducerHost, configuration.InfoProducerPort) + callbackAddress := fmt.Sprintf("%v:%v", configuration.InfoProducerHost, configuration.InfoProducerPort) - registrator := config.NewRegistratorImpl(configuration.InfoCoordinatorAddress) - if types, err := jobs.GetTypes(); err == nil { - if regErr := registrator.RegisterTypes(types); regErr != nil { - log.Fatalf("Unable to register all types due to: %v", regErr) - } + var cert tls.Certificate + if c, err := restclient.CreateClientCertificate(configuration.ProducerCertPath, configuration.ProducerKeyPath); err == nil { + cert = c } else { - log.Fatalf("Unable to get types to register due to: %v", err) + log.Fatalf("Stopping producer due to error: %v", err) + } + + retryClient := restclient.CreateRetryClient(cert) + kafkaFactory := kafkaclient.KafkaFactoryImpl{BootstrapServer: configuration.KafkaBootstrapServers} + distributionClient := restclient.CreateClientWithoutRetry(cert, 10*time.Second) + + jobsManager := jobs.NewJobsManagerImpl(retryClient, configuration.DMaaPMRAddress, kafkaFactory, distributionClient) + go startCallbackServer(jobsManager, callbackAddress) + + if err := registerTypesAndProducer(jobsManager, configuration.InfoCoordinatorAddress, callbackAddress, retryClient); err != nil { + log.Fatalf("Stopping producer due to: %v", err) + } + registered = true + jobsManager.StartJobsForAllTypes() + + log.Debug("Starting DMaaP Mediator Producer") + + keepProducerAlive() +} + +func validateConfiguration(configuration *config.Config) error { + if configuration.InfoProducerHost == "" { + return fmt.Errorf("missing INFO_PRODUCER_HOST") } + if configuration.ProducerCertPath == "" || configuration.ProducerKeyPath == "" { + return fmt.Errorf("missing PRODUCER_CERT and/or PRODUCER_KEY") + } + if configuration.DMaaPMRAddress == "" && configuration.KafkaBootstrapServers == "" { + return fmt.Errorf("at least one of DMAAP_MR_ADDR or KAFKA_BOOTSRAP_SERVERS must be provided") + } + return nil +} +func registerTypesAndProducer(jobTypesManager jobs.JobTypesManager, infoCoordinatorAddress string, callbackAddress string, client restclient.HTTPClient) error { + registrator := config.NewRegistratorImpl(infoCoordinatorAddress, client) + configTypes, err := config.GetJobTypesFromConfiguration("configs") + if err != nil { + return fmt.Errorf("unable to register all types due to: %v", err) + } + regErr := registrator.RegisterTypes(jobTypesManager.LoadTypesFromConfiguration(configTypes)) + if regErr != nil { + return fmt.Errorf("unable to register all types due to: %v", regErr) + } + producer := config.ProducerRegistrationInfo{ - InfoProducerSupervisionCallbackUrl: callbackAddress + server.StatusPath, - SupportedInfoTypes: jobs.GetSupportedTypes(), + InfoProducerSupervisionCallbackUrl: callbackAddress + server.HealthCheckPath, + SupportedInfoTypes: jobTypesManager.GetSupportedTypes(), InfoJobCallbackUrl: callbackAddress + server.AddJobPath, } if err := registrator.RegisterProducer("DMaaP_Mediator_Producer", &producer); err != nil { - log.Fatalf("Unable to register producer due to: %v", err) + return fmt.Errorf("unable to register producer due to: %v", err) } + return nil } -func main() { - log.Debug("Starting DMaaP Mediator Producer") - wg := new(sync.WaitGroup) +func startCallbackServer(jobsManager jobs.JobsManager, callbackAddress string) { + log.Debugf("Starting callback server at port %v", configuration.InfoProducerPort) + r := server.NewRouter(jobsManager, statusHandler) + addSwaggerHandler(r) + if restclient.IsUrlSecure(callbackAddress) { + log.Fatalf("Server stopped: %v", http.ListenAndServeTLS(fmt.Sprintf(":%v", configuration.InfoProducerPort), configuration.ProducerCertPath, configuration.ProducerKeyPath, r)) + } else { + log.Fatalf("Server stopped: %v", http.ListenAndServe(fmt.Sprintf(":%v", configuration.InfoProducerPort), r)) + } +} - // add two goroutines to `wg` WaitGroup, one for each running go routine - wg.Add(2) +type ProducerStatus struct { + // The registration status of the producer in Information Coordinator Service. Either `registered` or `not registered` + RegisteredStatus string `json:"registeredStatus" swaggertype:"string" example:"registered"` +} // @name ProducerStatus - log.Debugf("Starting callback server at port %v", configuration.InfoProducerPort) - go func() { - r := server.NewRouter() - log.Warn(http.ListenAndServe(fmt.Sprintf(":%v", configuration.InfoProducerPort), r)) - wg.Done() - }() - - go func() { - jobs.RunJobs(fmt.Sprintf("%v:%v", configuration.MRHost, configuration.MRPort)) - wg.Done() - }() - - // wait until WaitGroup is done - wg.Wait() - log.Debug("Stopping DMaaP Mediator Producer") +// @Summary Get status +// @Description Get the status of the producer. Will show if the producer has registered in ICS. +// @Tags Data producer (callbacks) +// @Produce json +// @Success 200 {object} ProducerStatus +// @Router /health_check [get] +func statusHandler(w http.ResponseWriter, r *http.Request) { + status := ProducerStatus{ + RegisteredStatus: "not registered", + } + if registered { + status.RegisteredStatus = "registered" + } + json.NewEncoder(w).Encode(status) +} + +// @Summary Get Swagger Documentation +// @Description Get the Swagger API documentation for the producer. +// @Tags Admin +// @Success 200 +// @Router /swagger [get] +func addSwaggerHandler(r *mux.Router) { + r.PathPrefix("/swagger").Handler(httpSwagger.WrapHandler) +} + +func keepProducerAlive() { + forever := make(chan int) + <-forever }