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         "encoding/json"
  26         "fmt"
  27         "net/http"
  28         "time"
  29
  30         "github.com/gorilla/mux"
  31         log "github.com/sirupsen/logrus"
  32         _ "oransc.org/nonrtric/dmaapmediatorproducer/api"
  33         "oransc.org/nonrtric/dmaapmediatorproducer/internal/config"
  34         "oransc.org/nonrtric/dmaapmediatorproducer/internal/jobs"
  35         "oransc.org/nonrtric/dmaapmediatorproducer/internal/kafkaclient"
  36         "oransc.org/nonrtric/dmaapmediatorproducer/internal/restclient"
  37         "oransc.org/nonrtric/dmaapmediatorproducer/internal/server"
  38
  39         httpSwagger "github.com/swaggo/http-swagger"
  40 )
  41
  42 var configuration *config.Config
  43 var registered bool
  44
  45 func init() {
  46         configuration = config.New()
  47 }
  48
  49 // @title    DMaaP Mediator Producer
  50 // @version  1.1.0
  51
  52 // @license.name  Apache 2.0
  53 // @license.url   http://www.apache.org/licenses/LICENSE-2.0.html
  54
  55 func main() {
  56         log.SetLevel(configuration.LogLevel)
  57         log.Debug("Initializing DMaaP Mediator Producer")
  58         log.Debug("Using configuration: ", configuration)
  59         if err := validateConfiguration(configuration); err != nil {
  60                 log.Fatalf("Stopping producer due to error: %v", err)
  61         }
  62         callbackAddress := fmt.Sprintf("%v:%v", configuration.InfoProducerHost, configuration.InfoProducerPort)
  63
  64         var cert tls.Certificate
  65         if c, err := restclient.CreateClientCertificate(configuration.ProducerCertPath, configuration.ProducerKeyPath); err == nil {
  66                 cert = c
  67         } else {
  68                 log.Fatalf("Stopping producer due to error: %v", err)
  69         }
  70
  71         retryClient := restclient.CreateRetryClient(cert)
  72         kafkaFactory := kafkaclient.KafkaFactoryImpl{BootstrapServer: configuration.KafkaBootstrapServers}
  73         distributionClient := restclient.CreateClientWithoutRetry(cert, 10*time.Second)
  74
  75         jobsManager := jobs.NewJobsManagerImpl(retryClient, configuration.DMaaPMRAddress, kafkaFactory, distributionClient)
  76         go startCallbackServer(jobsManager, callbackAddress)
  77
  78         if err := registerTypesAndProducer(jobsManager, configuration.InfoCoordinatorAddress, callbackAddress, retryClient); err != nil {
  79                 log.Fatalf("Stopping producer due to: %v", err)
  80         }
  81         registered = true
  82         jobsManager.StartJobsForAllTypes()
  83
  84         log.Debug("Starting DMaaP Mediator Producer")
  85
  86         keepProducerAlive()
  87 }
  88
  89 func validateConfiguration(configuration *config.Config) error {
  90         if configuration.InfoProducerHost == "" {
  91                 return fmt.Errorf("missing INFO_PRODUCER_HOST")
  92         }
  93         if configuration.ProducerCertPath == "" || configuration.ProducerKeyPath == "" {
  94                 return fmt.Errorf("missing PRODUCER_CERT and/or PRODUCER_KEY")
  95         }
  96         if configuration.DMaaPMRAddress == "" && configuration.KafkaBootstrapServers == "" {
  97                 return fmt.Errorf("at least one of DMAAP_MR_ADDR or KAFKA_BOOTSRAP_SERVERS must be provided")
  98         }
  99         return nil
 100 }
 101 func registerTypesAndProducer(jobTypesManager jobs.JobTypesManager, infoCoordinatorAddress string, callbackAddress string, client restclient.HTTPClient) error {
 102         registrator := config.NewRegistratorImpl(infoCoordinatorAddress, client)
 103         configTypes, err := config.GetJobTypesFromConfiguration("configs")
 104         if err != nil {
 105                 return fmt.Errorf("unable to register all types due to: %v", err)
 106         }
 107         regErr := registrator.RegisterTypes(jobTypesManager.LoadTypesFromConfiguration(configTypes))
 108         if regErr != nil {
 109                 return fmt.Errorf("unable to register all types due to: %v", regErr)
 110         }
 111
 112         producer := config.ProducerRegistrationInfo{
 113                 InfoProducerSupervisionCallbackUrl: callbackAddress + server.HealthCheckPath,
 114                 SupportedInfoTypes:                 jobTypesManager.GetSupportedTypes(),
 115                 InfoJobCallbackUrl:                 callbackAddress + server.AddJobPath,
 116         }
 117         if err := registrator.RegisterProducer("DMaaP_Mediator_Producer", &producer); err != nil {
 118                 return fmt.Errorf("unable to register producer due to: %v", err)
 119         }
 120         return nil
 121 }
 122
 123 func startCallbackServer(jobsManager jobs.JobsManager, callbackAddress string) {
 124         log.Debugf("Starting callback server at port %v", configuration.InfoProducerPort)
 125         r := server.NewRouter(jobsManager, statusHandler)
 126         addSwaggerHandler(r)
 127         if restclient.IsUrlSecure(callbackAddress) {
 128                 log.Fatalf("Server stopped: %v", http.ListenAndServeTLS(fmt.Sprintf(":%v", configuration.InfoProducerPort), configuration.ProducerCertPath, configuration.ProducerKeyPath, r))
 129         } else {
 130                 log.Fatalf("Server stopped: %v", http.ListenAndServe(fmt.Sprintf(":%v", configuration.InfoProducerPort), r))
 131         }
 132 }
 133
 134 type ProducerStatus struct {
 135         // The registration status of the producer in Information Coordinator Service. Either `registered` or `not registered`
 136         RegisteredStatus string `json:"registeredStatus" swaggertype:"string" example:"registered"`
 137 } // @name  ProducerStatus
 138
 139 // @Summary      Get status
 140 // @Description  Get the status of the producer. Will show if the producer has registered in ICS.
 141 // @Tags         Data producer (callbacks)
 142 // @Produce      json
 143 // @Success      200  {object}  ProducerStatus
 144 // @Router       /health_check [get]
 145 func statusHandler(w http.ResponseWriter, r *http.Request) {
 146         status := ProducerStatus{
 147                 RegisteredStatus: "not registered",
 148         }
 149         if registered {
 150                 status.RegisteredStatus = "registered"
 151         }
 152         json.NewEncoder(w).Encode(status)
 153 }
 154
 155 // @Summary      Get Swagger Documentation
 156 // @Description  Get the Swagger API documentation for the producer.
 157 // @Tags         Admin
 158 // @Success      200
 159 // @Router       /swagger [get]
 160 func addSwaggerHandler(r *mux.Router) {
 161         r.PathPrefix("/swagger").Handler(httpSwagger.WrapHandler)
 162 }
 163
 164 func keepProducerAlive() {
 165         forever := make(chan int)
 166         <-forever
 167 }