elinuxhenrik | cce95ff | 2021-09-05 17:27:02 +0200 | [diff] [blame] | 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 ( |
elinuxhenrik | ba96d84 | 2021-09-06 16:05:01 +0200 | [diff] [blame] | 24 | "fmt" |
elinuxhenrik | 382870d | 2021-09-23 11:09:09 +0200 | [diff] [blame] | 25 | "net/http" |
elinuxhenrik | 63a42ca | 2021-09-06 22:16:24 +0200 | [diff] [blame] | 26 | "sync" |
elinuxhenrik | 65a53d2 | 2021-09-29 15:41:26 +0200 | [diff] [blame] | 27 | "time" |
elinuxhenrik | cce95ff | 2021-09-05 17:27:02 +0200 | [diff] [blame] | 28 | |
elinuxhenrik | 803e81a | 2021-10-26 13:22:55 +0200 | [diff] [blame^] | 29 | "github.com/hashicorp/go-retryablehttp" |
elinuxhenrik | cce95ff | 2021-09-05 17:27:02 +0200 | [diff] [blame] | 30 | log "github.com/sirupsen/logrus" |
| 31 | "oransc.org/nonrtric/dmaapmediatorproducer/internal/config" |
elinuxhenrik | 63a42ca | 2021-09-06 22:16:24 +0200 | [diff] [blame] | 32 | "oransc.org/nonrtric/dmaapmediatorproducer/internal/jobs" |
elinuxhenrik | 65a53d2 | 2021-09-29 15:41:26 +0200 | [diff] [blame] | 33 | "oransc.org/nonrtric/dmaapmediatorproducer/internal/restclient" |
elinuxhenrik | ba96d84 | 2021-09-06 16:05:01 +0200 | [diff] [blame] | 34 | "oransc.org/nonrtric/dmaapmediatorproducer/internal/server" |
elinuxhenrik | cce95ff | 2021-09-05 17:27:02 +0200 | [diff] [blame] | 35 | ) |
| 36 | |
elinuxhenrik | 803e81a | 2021-10-26 13:22:55 +0200 | [diff] [blame^] | 37 | const timeoutDistributionClient = time.Second * 5 |
| 38 | const retryWaitMax = time.Minute |
| 39 | const retryMax = int(^uint(0) >> 1) |
elinuxhenrik | 65a53d2 | 2021-09-29 15:41:26 +0200 | [diff] [blame] | 40 | |
elinuxhenrik | cce95ff | 2021-09-05 17:27:02 +0200 | [diff] [blame] | 41 | var configuration *config.Config |
elinuxhenrik | 803e81a | 2021-10-26 13:22:55 +0200 | [diff] [blame^] | 42 | var retryClient restclient.HTTPClient |
elinuxhenrik | 65a53d2 | 2021-09-29 15:41:26 +0200 | [diff] [blame] | 43 | var jobHandler *jobs.JobHandlerImpl |
elinuxhenrik | cce95ff | 2021-09-05 17:27:02 +0200 | [diff] [blame] | 44 | |
| 45 | func init() { |
| 46 | configuration = config.New() |
elinuxhenrik | cce95ff | 2021-09-05 17:27:02 +0200 | [diff] [blame] | 47 | } |
| 48 | |
| 49 | func main() { |
elinuxhenrik | 65a53d2 | 2021-09-29 15:41:26 +0200 | [diff] [blame] | 50 | log.SetLevel(configuration.LogLevel) |
| 51 | log.Debug("Initializing DMaaP Mediator Producer") |
| 52 | if err := validateConfiguration(configuration); err != nil { |
| 53 | log.Fatalf("Stopping producer due to error: %v", err) |
| 54 | } |
| 55 | callbackAddress := fmt.Sprintf("%v:%v", configuration.InfoProducerHost, configuration.InfoProducerPort) |
| 56 | |
elinuxhenrik | 803e81a | 2021-10-26 13:22:55 +0200 | [diff] [blame^] | 57 | distributionClient := &http.Client{ |
| 58 | Timeout: timeoutDistributionClient, |
elinuxhenrik | 65a53d2 | 2021-09-29 15:41:26 +0200 | [diff] [blame] | 59 | } |
elinuxhenrik | 803e81a | 2021-10-26 13:22:55 +0200 | [diff] [blame^] | 60 | |
| 61 | rawRetryClient := retryablehttp.NewClient() |
| 62 | rawRetryClient.RetryWaitMax = retryWaitMax |
| 63 | rawRetryClient.RetryMax = retryMax |
| 64 | retryClient = rawRetryClient.StandardClient() |
| 65 | |
| 66 | jobHandler = jobs.NewJobHandlerImpl("configs/type_config.json", retryClient, distributionClient) |
elinuxhenrik | 65a53d2 | 2021-09-29 15:41:26 +0200 | [diff] [blame] | 67 | if err := registerTypesAndProducer(jobHandler, configuration.InfoCoordinatorAddress, callbackAddress); err != nil { |
| 68 | log.Fatalf("Stopping producer due to: %v", err) |
| 69 | } |
| 70 | |
elinuxhenrik | cce95ff | 2021-09-05 17:27:02 +0200 | [diff] [blame] | 71 | log.Debug("Starting DMaaP Mediator Producer") |
elinuxhenrik | 63a42ca | 2021-09-06 22:16:24 +0200 | [diff] [blame] | 72 | wg := new(sync.WaitGroup) |
elinuxhenrik | ba96d84 | 2021-09-06 16:05:01 +0200 | [diff] [blame] | 73 | |
elinuxhenrik | 382870d | 2021-09-23 11:09:09 +0200 | [diff] [blame] | 74 | // add two goroutines to `wg` WaitGroup, one for each running go routine |
| 75 | wg.Add(2) |
elinuxhenrik | 63a42ca | 2021-09-06 22:16:24 +0200 | [diff] [blame] | 76 | |
elinuxhenrik | 382870d | 2021-09-23 11:09:09 +0200 | [diff] [blame] | 77 | log.Debugf("Starting callback server at port %v", configuration.InfoProducerPort) |
elinuxhenrik | 63a42ca | 2021-09-06 22:16:24 +0200 | [diff] [blame] | 78 | go func() { |
elinuxhenrik | 65a53d2 | 2021-09-29 15:41:26 +0200 | [diff] [blame] | 79 | r := server.NewRouter(jobHandler) |
elinuxhenrik | fe61c61 | 2021-09-24 15:08:47 +0200 | [diff] [blame] | 80 | log.Warn(http.ListenAndServe(fmt.Sprintf(":%v", configuration.InfoProducerPort), r)) |
elinuxhenrik | 63a42ca | 2021-09-06 22:16:24 +0200 | [diff] [blame] | 81 | wg.Done() |
| 82 | }() |
| 83 | |
elinuxhenrik | 2803856 | 2021-09-21 15:43:11 +0200 | [diff] [blame] | 84 | go func() { |
elinuxhenrik | 65a53d2 | 2021-09-29 15:41:26 +0200 | [diff] [blame] | 85 | jobHandler.RunJobs(fmt.Sprintf("%v:%v", configuration.MRHost, configuration.MRPort)) |
elinuxhenrik | 2803856 | 2021-09-21 15:43:11 +0200 | [diff] [blame] | 86 | wg.Done() |
| 87 | }() |
| 88 | |
elinuxhenrik | 63a42ca | 2021-09-06 22:16:24 +0200 | [diff] [blame] | 89 | // wait until WaitGroup is done |
| 90 | wg.Wait() |
elinuxhenrik | ba96d84 | 2021-09-06 16:05:01 +0200 | [diff] [blame] | 91 | log.Debug("Stopping DMaaP Mediator Producer") |
elinuxhenrik | cce95ff | 2021-09-05 17:27:02 +0200 | [diff] [blame] | 92 | } |
elinuxhenrik | 65a53d2 | 2021-09-29 15:41:26 +0200 | [diff] [blame] | 93 | |
| 94 | func validateConfiguration(configuration *config.Config) error { |
| 95 | if configuration.InfoProducerHost == "" { |
| 96 | return fmt.Errorf("missing INFO_PRODUCER_HOST") |
| 97 | } |
| 98 | return nil |
| 99 | } |
| 100 | |
| 101 | func registerTypesAndProducer(jobHandler jobs.JobTypeHandler, infoCoordinatorAddress string, callbackAddress string) error { |
elinuxhenrik | 803e81a | 2021-10-26 13:22:55 +0200 | [diff] [blame^] | 102 | registrator := config.NewRegistratorImpl(infoCoordinatorAddress, retryClient) |
elinuxhenrik | 65a53d2 | 2021-09-29 15:41:26 +0200 | [diff] [blame] | 103 | if types, err := jobHandler.GetTypes(); err == nil { |
| 104 | if regErr := registrator.RegisterTypes(types); regErr != nil { |
| 105 | return fmt.Errorf("unable to register all types due to: %v", regErr) |
| 106 | } |
| 107 | } else { |
| 108 | return fmt.Errorf("unable to get types to register due to: %v", err) |
| 109 | } |
| 110 | producer := config.ProducerRegistrationInfo{ |
| 111 | InfoProducerSupervisionCallbackUrl: callbackAddress + server.StatusPath, |
| 112 | SupportedInfoTypes: jobHandler.GetSupportedTypes(), |
| 113 | InfoJobCallbackUrl: callbackAddress + server.AddJobPath, |
| 114 | } |
| 115 | if err := registrator.RegisterProducer("DMaaP_Mediator_Producer", &producer); err != nil { |
| 116 | return fmt.Errorf("unable to register producer due to: %v", err) |
| 117 | } |
| 118 | return nil |
| 119 | } |