dmaap-mediator-producer/main.go

   1 // -
   2 //   ========================LICENSE_START=================================
   3 //   O-RAN-SC
   4 //   %%
   5 //   Copyright (C) 2021: Nordix Foundation
   6 //   %%
   7 //   Licensed under the Apache License, Version 2.0 (the "License");
   8 //   you may not use this file except in compliance with the License.
   9 //   You may obtain a copy of the License at
  10 //
  11 //        http://www.apache.org/licenses/LICENSE-2.0
  12 //
  13 //   Unless required by applicable law or agreed to in writing, software
  14 //   distributed under the License is distributed on an "AS IS" BASIS,
  15 //   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16 //   See the License for the specific language governing permissions and
  17 //   limitations under the License.
  18 //   ========================LICENSE_END===================================
  19 //
  20
  21 package main
  22
  23 import (
  24         "crypto/tls"
  25         "fmt"
  26         "net/http"
  27         "time"
  28
  29         log "github.com/sirupsen/logrus"
  30         "oransc.org/nonrtric/dmaapmediatorproducer/internal/config"
  31         "oransc.org/nonrtric/dmaapmediatorproducer/internal/jobs"
  32         "oransc.org/nonrtric/dmaapmediatorproducer/internal/kafkaclient"
  33         "oransc.org/nonrtric/dmaapmediatorproducer/internal/restclient"
  34         "oransc.org/nonrtric/dmaapmediatorproducer/internal/server"
  35 )
  36
  37 var configuration *config.Config
  38 var registered bool
  39
  40 func init() {
  41         configuration = config.New()
  42 }
  43
  44 func main() {
  45         log.SetLevel(configuration.LogLevel)
  46         log.Debug("Initializing DMaaP Mediator Producer")
  47         log.Debug("Using configuration: ", configuration)
  48         if err := validateConfiguration(configuration); err != nil {
  49                 log.Fatalf("Stopping producer due to error: %v", err)
  50         }
  51         callbackAddress := fmt.Sprintf("%v:%v", configuration.InfoProducerHost, configuration.InfoProducerPort)
  52
  53         var cert tls.Certificate
  54         if c, err := restclient.CreateClientCertificate(configuration.ProducerCertPath, configuration.ProducerKeyPath); err == nil {
  55                 cert = c
  56         } else {
  57                 log.Fatalf("Stopping producer due to error: %v", err)
  58         }
  59
  60         retryClient := restclient.CreateRetryClient(cert)
  61         kafkaFactory := kafkaclient.KafkaFactoryImpl{BootstrapServer: configuration.KafkaBootstrapServers}
  62         distributionClient := restclient.CreateClientWithoutRetry(cert, 10*time.Second)
  63
  64         jobsManager := jobs.NewJobsManagerImpl(retryClient, configuration.DMaaPMRAddress, kafkaFactory, distributionClient)
  65         go startCallbackServer(jobsManager, callbackAddress)
  66
  67         if err := registerTypesAndProducer(jobsManager, configuration.InfoCoordinatorAddress, callbackAddress, retryClient); err != nil {
  68                 log.Fatalf("Stopping producer due to: %v", err)
  69         }
  70         registered = true
  71         jobsManager.StartJobsForAllTypes()
  72
  73         log.Debug("Starting DMaaP Mediator Producer")
  74
  75         keepProducerAlive()
  76 }
  77
  78 func validateConfiguration(configuration *config.Config) error {
  79         if configuration.InfoProducerHost == "" {
  80                 return fmt.Errorf("missing INFO_PRODUCER_HOST")
  81         }
  82         if configuration.ProducerCertPath == "" || configuration.ProducerKeyPath == "" {
  83                 return fmt.Errorf("missing PRODUCER_CERT and/or PRODUCER_KEY")
  84         }
  85         if configuration.DMaaPMRAddress == "" && configuration.KafkaBootstrapServers == "" {
  86                 return fmt.Errorf("at least one of DMAAP_MR_ADDR or KAFKA_BOOTSRAP_SERVERS must be provided")
  87         }
  88         return nil
  89 }
  90 func registerTypesAndProducer(jobTypesHandler jobs.JobTypesManager, infoCoordinatorAddress string, callbackAddress string, client restclient.HTTPClient) error {
  91         registrator := config.NewRegistratorImpl(infoCoordinatorAddress, client)
  92         configTypes, err := config.GetJobTypesFromConfiguration("configs")
  93         if err != nil {
  94                 return fmt.Errorf("unable to register all types due to: %v", err)
  95         }
  96         regErr := registrator.RegisterTypes(jobTypesHandler.LoadTypesFromConfiguration(configTypes))
  97         if regErr != nil {
  98                 return fmt.Errorf("unable to register all types due to: %v", regErr)
  99         }
 100
 101         producer := config.ProducerRegistrationInfo{
 102                 InfoProducerSupervisionCallbackUrl: callbackAddress + server.HealthCheckPath,
 103                 SupportedInfoTypes:                 jobTypesHandler.GetSupportedTypes(),
 104                 InfoJobCallbackUrl:                 callbackAddress + server.AddJobPath,
 105         }
 106         if err := registrator.RegisterProducer("DMaaP_Mediator_Producer", &producer); err != nil {
 107                 return fmt.Errorf("unable to register producer due to: %v", err)
 108         }
 109         return nil
 110 }
 111
 112 func startCallbackServer(jobsManager jobs.JobsManager, callbackAddress string) {
 113         log.Debugf("Starting callback server at port %v", configuration.InfoProducerPort)
 114         r := server.NewRouter(jobsManager, statusHandler)
 115         if restclient.IsUrlSecure(callbackAddress) {
 116                 log.Fatalf("Server stopped: %v", http.ListenAndServeTLS(fmt.Sprintf(":%v", configuration.InfoProducerPort), configuration.ProducerCertPath, configuration.ProducerKeyPath, r))
 117         } else {
 118                 log.Fatalf("Server stopped: %v", http.ListenAndServe(fmt.Sprintf(":%v", configuration.InfoProducerPort), r))
 119         }
 120 }
 121
 122 func statusHandler(w http.ResponseWriter, r *http.Request) {
 123         registeredStatus := "not registered"
 124         if registered {
 125                 registeredStatus = "registered"
 126         }
 127         fmt.Fprintf(w, `{"status": "%v"}`, registeredStatus)
 128 }
 129
 130 func keepProducerAlive() {
 131         forever := make(chan int)
 132         <-forever
 133 }