X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;ds=sidebyside;f=dmaap-mediator-producer%2Fmain.go;h=65a84a2150b70fcfcf1daa95859daf61194e7354;hb=2889efe2ff8128ed574d1debd668d1ad88e4681a;hp=b357f6962418fc3cce537489122fe6946a7d3cd5;hpb=ba96d847ab72a882bfe3acd39bc70c8f544f094a;p=nonrtric.git diff --git a/dmaap-mediator-producer/main.go b/dmaap-mediator-producer/main.go index b357f696..65a84a21 100644 --- a/dmaap-mediator-producer/main.go +++ b/dmaap-mediator-producer/main.go @@ -21,63 +21,147 @@ package main import ( + "crypto/tls" + "encoding/json" "fmt" "net/http" + "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/jobtypes" + "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 supervisionCallbackAddress string -var jobInfoCallbackAddress 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.InfoProducerSupervisionCallbackHost == "" { - 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) } - supervisionCallbackAddress = fmt.Sprintf("%v:%v", configuration.InfoProducerSupervisionCallbackHost, configuration.InfoProducerSupervisionCallbackPort) + callbackAddress := fmt.Sprintf("%v:%v", configuration.InfoProducerHost, configuration.InfoProducerPort) - if configuration.InfoJobCallbackHost == "" { - log.Fatal("Missing INFO_JOB_CALLBACK_HOST") + var cert tls.Certificate + if c, err := restclient.CreateClientCertificate(configuration.ProducerCertPath, configuration.ProducerKeyPath); err == nil { + cert = c + } else { + log.Fatalf("Stopping producer due to error: %v", err) } - jobInfoCallbackAddress = fmt.Sprintf("%v:%v", configuration.InfoJobCallbackHost, configuration.InfoJobCallbackPort) - registrator := config.NewRegistratorImpl(configuration.InfoCoordinatorAddress) - if types, err := jobtypes.GetTypes(); err == nil { - if regErr := registrator.RegisterTypes(types); regErr != nil { - log.Fatalf("Unable to register all types due to: %v", regErr) - } - } else { - log.Fatalf("Unable to get types to register due to: %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: supervisionCallbackAddress, - SupportedInfoTypes: jobtypes.GetSupportedTypes(), - InfoJobCallbackUrl: jobInfoCallbackAddress, + 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") - log.Debugf("Starting status callback server at port %v", configuration.InfoProducerSupervisionCallbackPort) - http.HandleFunc("/", server.StatusHandler) +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)) + } +} - if err := http.ListenAndServe(":"+configuration.InfoProducerSupervisionCallbackPort, nil); err != nil { - log.Fatal(err) +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 + +// @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" } - log.Debug("Stopping DMaaP Mediator Producer") + 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 }