65a84a2150b70fcfcf1daa95859daf61194e7354
[nonrtric.git] / 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 }