X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=dmaap-mediator-producer%2Fmain.go;h=819ffa9d811ee143a8b4b109619ccaa799646dee;hb=9fe98aaeb6bd40cbb46eb8bb1e7f46ce3c8d3b02;hp=e8689bd49b06d17c2147e485c0a4fe53ea8da52e;hpb=11b500e83e30dbd4bd39be88db4291d643607592;p=nonrtric.git diff --git a/dmaap-mediator-producer/main.go b/dmaap-mediator-producer/main.go index e8689bd4..819ffa9d 100644 --- a/dmaap-mediator-producer/main.go +++ b/dmaap-mediator-producer/main.go @@ -26,15 +26,16 @@ import ( "net/http" "time" - "github.com/hashicorp/go-retryablehttp" log "github.com/sirupsen/logrus" "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" ) var configuration *config.Config +var registered bool func init() { configuration = config.New() @@ -43,33 +44,33 @@ func init() { func main() { log.SetLevel(configuration.LogLevel) log.Debug("Initializing DMaaP Mediator Producer") + 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) - var retryClient restclient.HTTPClient - if cert, err := createClientCertificate(); err == nil { - retryClient = createRetryClient(cert) + 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) } - jobHandler := jobs.NewJobHandlerImpl("configs/type_config.json", retryClient, &http.Client{ - Timeout: time.Second * 5, - }) - if err := registerTypesAndProducer(jobHandler, configuration.InfoCoordinatorAddress, callbackAddress, retryClient); err != nil { + 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") - go func() { - log.Debugf("Starting callback server at port %v", configuration.InfoProducerPort) - r := server.NewRouter(jobHandler) - log.Fatalf("Server stopped: %v", http.ListenAndServeTLS(fmt.Sprintf(":%v", configuration.InfoProducerPort), configuration.ProducerCertPath, configuration.ProducerKeyPath, r)) - }() - - go jobHandler.RunJobs(configuration.DMaaPMRAddress) keepProducerAlive() } @@ -81,45 +82,25 @@ func validateConfiguration(configuration *config.Config) error { 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 createClientCertificate() (*tls.Certificate, error) { - if cert, err := tls.LoadX509KeyPair(configuration.ProducerCertPath, configuration.ProducerKeyPath); err == nil { - return &cert, nil - } else { - return nil, fmt.Errorf("cannot create x509 keypair from cert file %s and key file %s due to: %v", configuration.ProducerCertPath, configuration.ProducerKeyPath, err) +func registerTypesAndProducer(jobTypesHandler 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) } -} - -func createRetryClient(cert *tls.Certificate) *http.Client { - rawRetryClient := retryablehttp.NewClient() - rawRetryClient.RetryWaitMax = time.Minute - rawRetryClient.RetryMax = int(^uint(0) >> 1) - rawRetryClient.HTTPClient.Transport = &http.Transport{ - TLSClientConfig: &tls.Config{ - Certificates: []tls.Certificate{ - *cert, - }, - InsecureSkipVerify: true, - }, + regErr := registrator.RegisterTypes(jobTypesHandler.LoadTypesFromConfiguration(configTypes)) + if regErr != nil { + return fmt.Errorf("unable to register all types due to: %v", regErr) } - return rawRetryClient.StandardClient() -} - -func registerTypesAndProducer(jobHandler jobs.JobTypeHandler, infoCoordinatorAddress string, callbackAddress string, client restclient.HTTPClient) error { - registrator := config.NewRegistratorImpl(infoCoordinatorAddress, client) - if types, err := jobHandler.GetTypes(); err == nil { - if regErr := registrator.RegisterTypes(types); regErr != nil { - return fmt.Errorf("unable to register all types due to: %v", regErr) - } - } else { - return fmt.Errorf("unable to get types to register due to: %v", err) - } producer := config.ProducerRegistrationInfo{ - InfoProducerSupervisionCallbackUrl: callbackAddress + server.StatusPath, - SupportedInfoTypes: jobHandler.GetSupportedTypes(), + InfoProducerSupervisionCallbackUrl: callbackAddress + server.HealthCheckPath, + SupportedInfoTypes: jobTypesHandler.GetSupportedTypes(), InfoJobCallbackUrl: callbackAddress + server.AddJobPath, } if err := registrator.RegisterProducer("DMaaP_Mediator_Producer", &producer); err != nil { @@ -128,6 +109,24 @@ func registerTypesAndProducer(jobHandler jobs.JobTypeHandler, infoCoordinatorAdd return nil } +func startCallbackServer(jobsManager jobs.JobsManager, callbackAddress string) { + log.Debugf("Starting callback server at port %v", configuration.InfoProducerPort) + r := server.NewRouter(jobsManager, statusHandler) + 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)) + } +} + +func statusHandler(w http.ResponseWriter, r *http.Request) { + registeredStatus := "not registered" + if registered { + registeredStatus = "registered" + } + fmt.Fprintf(w, `{"status": "%v"}`, registeredStatus) +} + func keepProducerAlive() { forever := make(chan int) <-forever