blob: 3068c981ae477b98112520b013d839b6ddc4acff [file] [log] [blame]
khenaidoobf6e7bb2018-08-14 22:27:29 -04001/*
2 * Copyright 2018-present Open Networking Foundation
3
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
7
8 * http://www.apache.org/licenses/LICENSE-2.0
9
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
15 */
npujar1d86a522019-11-14 17:11:16 +053016
khenaidoocfee5f42018-07-19 22:47:38 -040017package main
18
19import (
khenaidoo5c11af72018-07-20 17:21:05 -040020 "context"
21 "errors"
khenaidoocfee5f42018-07-19 22:47:38 -040022 "fmt"
23 "os"
24 "os/signal"
khenaidoocfee5f42018-07-19 22:47:38 -040025 "strconv"
khenaidoocfee5f42018-07-19 22:47:38 -040026 "syscall"
khenaidoo5c11af72018-07-20 17:21:05 -040027 "time"
npujar1d86a522019-11-14 17:11:16 +053028
29 "github.com/opencord/voltha-go/rw_core/config"
30 c "github.com/opencord/voltha-go/rw_core/core"
31 "github.com/opencord/voltha-go/rw_core/utils"
divyadesaic68c9c02020-02-26 12:48:09 +000032 conf "github.com/opencord/voltha-lib-go/v3/pkg/config"
serkant.uluderya2ae470f2020-01-21 11:13:09 -080033 "github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore"
34 "github.com/opencord/voltha-lib-go/v3/pkg/kafka"
35 "github.com/opencord/voltha-lib-go/v3/pkg/log"
36 "github.com/opencord/voltha-lib-go/v3/pkg/probe"
37 "github.com/opencord/voltha-lib-go/v3/pkg/version"
38 ic "github.com/opencord/voltha-protos/v3/go/inter_container"
khenaidoocfee5f42018-07-19 22:47:38 -040039)
40
41type rwCore struct {
khenaidoo5c11af72018-07-20 17:21:05 -040042 kvClient kvstore.Client
43 config *config.RWCoreFlags
44 halted bool
45 exitChannel chan int
khenaidoob9203542018-09-17 22:56:37 -040046 //kmp *kafka.KafkaMessagingProxy
khenaidoo43c82122018-11-22 18:38:28 -050047 kafkaClient kafka.Client
48 core *c.Core
khenaidooabad44c2018-08-03 16:58:35 -040049 //For test
khenaidoo79232702018-12-04 11:00:41 -050050 receiverChannels []<-chan *ic.InterContainerMessage
khenaidoocfee5f42018-07-19 22:47:38 -040051}
52
khenaidoob9203542018-09-17 22:56:37 -040053func init() {
npujar1d86a522019-11-14 17:11:16 +053054 _, err := log.AddPackage(log.JSON, log.DebugLevel, nil)
55 if err != nil {
56 log.Errorw("unable-to-register-package-to-the-log-map", log.Fields{"error": err})
57 }
khenaidoob9203542018-09-17 22:56:37 -040058}
59
khenaidoocfee5f42018-07-19 22:47:38 -040060func newKVClient(storeType string, address string, timeout int) (kvstore.Client, error) {
61
khenaidoo5c11af72018-07-20 17:21:05 -040062 log.Infow("kv-store-type", log.Fields{"store": storeType})
khenaidoocfee5f42018-07-19 22:47:38 -040063 switch storeType {
64 case "consul":
65 return kvstore.NewConsulClient(address, timeout)
66 case "etcd":
67 return kvstore.NewEtcdClient(address, timeout)
68 }
69 return nil, errors.New("unsupported-kv-store")
70}
71
Scott Bakeree6a0872019-10-29 15:59:52 -070072func newKafkaClient(clientType string, host string, port int, instanceID string, livenessChannelInterval time.Duration) (kafka.Client, error) {
khenaidoo43c82122018-11-22 18:38:28 -050073
74 log.Infow("kafka-client-type", log.Fields{"client": clientType})
75 switch clientType {
76 case "sarama":
77 return kafka.NewSaramaClient(
78 kafka.Host(host),
khenaidoo90847922018-12-03 14:47:51 -050079 kafka.Port(port),
khenaidooca301322019-01-09 23:06:32 -050080 kafka.ConsumerType(kafka.GroupCustomer),
khenaidoo90847922018-12-03 14:47:51 -050081 kafka.ProducerReturnOnErrors(true),
82 kafka.ProducerReturnOnSuccess(true),
83 kafka.ProducerMaxRetries(6),
khenaidooca301322019-01-09 23:06:32 -050084 kafka.NumPartitions(3),
85 kafka.ConsumerGroupName(instanceID),
86 kafka.ConsumerGroupPrefix(instanceID),
khenaidoo54e0ddf2019-02-27 16:21:33 -050087 kafka.AutoCreateTopic(true),
khenaidooca301322019-01-09 23:06:32 -050088 kafka.ProducerFlushFrequency(5),
Scott Bakeree6a0872019-10-29 15:59:52 -070089 kafka.ProducerRetryBackoff(time.Millisecond*30),
90 kafka.LivenessChannelInterval(livenessChannelInterval),
91 ), nil
khenaidoo43c82122018-11-22 18:38:28 -050092 }
93 return nil, errors.New("unsupported-client-type")
94}
95
khenaidoocfee5f42018-07-19 22:47:38 -040096func newRWCore(cf *config.RWCoreFlags) *rwCore {
97 var rwCore rwCore
98 rwCore.config = cf
99 rwCore.halted = false
100 rwCore.exitChannel = make(chan int, 1)
khenaidoo79232702018-12-04 11:00:41 -0500101 rwCore.receiverChannels = make([]<-chan *ic.InterContainerMessage, 0)
khenaidoocfee5f42018-07-19 22:47:38 -0400102 return &rwCore
103}
104
npujar1d86a522019-11-14 17:11:16 +0530105func (rw *rwCore) start(ctx context.Context, instanceID string) {
khenaidoo5c11af72018-07-20 17:21:05 -0400106 log.Info("Starting RW Core components")
khenaidoob9203542018-09-17 22:56:37 -0400107
khenaidoo90847922018-12-03 14:47:51 -0500108 // Setup KV Client
Richard Jankowskie4d77662018-10-17 13:53:21 -0400109 log.Debugw("create-kv-client", log.Fields{"kvstore": rw.config.KVStoreType})
Kent Hagerman16ce36a2019-12-17 13:40:53 -0500110 var err error
111 if rw.kvClient, err = newKVClient(
112 rw.config.KVStoreType,
113 rw.config.KVStoreHost+":"+strconv.Itoa(rw.config.KVStorePort),
114 rw.config.KVStoreTimeout); err != nil {
115 log.Fatal(err)
116 }
divyadesaic68c9c02020-02-26 12:48:09 +0000117 cm := conf.NewConfigManager(rw.kvClient, rw.config.KVStoreType, rw.config.KVStoreHost, rw.config.KVStorePort, rw.config.KVStoreTimeout)
118 go conf.ProcessLogConfigChange(cm, ctx)
Kent Hagerman16ce36a2019-12-17 13:40:53 -0500119
120 // Setup KV transaction context
121 if err := c.SetTransactionContext(instanceID,
122 rw.config.KVStoreDataPrefix+"/transactions/",
123 rw.kvClient,
124 rw.config.KVStoreTimeout); err != nil {
125 log.Fatal("creating-transaction-context-failed")
Richard Jankowskie4d77662018-10-17 13:53:21 -0400126 }
127
khenaidoo90847922018-12-03 14:47:51 -0500128 // Setup Kafka Client
Scott Bakeree6a0872019-10-29 15:59:52 -0700129 if rw.kafkaClient, err = newKafkaClient("sarama",
130 rw.config.KafkaAdapterHost,
131 rw.config.KafkaAdapterPort,
npujar1d86a522019-11-14 17:11:16 +0530132 instanceID,
Girish Kumar4d3887d2019-11-22 14:22:05 +0000133 rw.config.LiveProbeInterval/2); err != nil {
khenaidoo43c82122018-11-22 18:38:28 -0500134 log.Fatal("Unsupported-kafka-client")
135 }
136
khenaidoob9203542018-09-17 22:56:37 -0400137 // Create the core service
Thomas Lee Se5a44012019-11-07 20:32:24 +0530138 rw.core = c.NewCore(ctx, instanceID, rw.config, rw.kvClient, rw.kafkaClient)
khenaidoob9203542018-09-17 22:56:37 -0400139
140 // start the core
Thomas Lee Se5a44012019-11-07 20:32:24 +0530141 err = rw.core.Start(ctx)
142 if err != nil {
143 log.Fatalf("failed-to-start-rwcore", log.Fields{"error": err})
144 }
khenaidoocfee5f42018-07-19 22:47:38 -0400145}
146
David K. Bainbridgeb4a9ab02019-09-20 15:12:16 -0700147func (rw *rwCore) stop(ctx context.Context) {
khenaidoocfee5f42018-07-19 22:47:38 -0400148 // Stop leadership tracking
khenaidoob9203542018-09-17 22:56:37 -0400149 rw.halted = true
khenaidoocfee5f42018-07-19 22:47:38 -0400150
151 // send exit signal
khenaidoob9203542018-09-17 22:56:37 -0400152 rw.exitChannel <- 0
khenaidoocfee5f42018-07-19 22:47:38 -0400153
154 // Cleanup - applies only if we had a kvClient
khenaidoob9203542018-09-17 22:56:37 -0400155 if rw.kvClient != nil {
khenaidoocfee5f42018-07-19 22:47:38 -0400156 // Release all reservations
npujar467fe752020-01-16 20:17:45 +0530157 if err := rw.kvClient.ReleaseAllReservations(ctx); err != nil {
khenaidoo5c11af72018-07-20 17:21:05 -0400158 log.Infow("fail-to-release-all-reservations", log.Fields{"error": err})
khenaidoocfee5f42018-07-19 22:47:38 -0400159 }
160 // Close the DB connection
khenaidoob9203542018-09-17 22:56:37 -0400161 rw.kvClient.Close()
khenaidoocfee5f42018-07-19 22:47:38 -0400162 }
khenaidoo43c82122018-11-22 18:38:28 -0500163
David K. Bainbridgeb4a9ab02019-09-20 15:12:16 -0700164 rw.core.Stop(ctx)
khenaidoo43c82122018-11-22 18:38:28 -0500165
166 //if rw.kafkaClient != nil {
167 // rw.kafkaClient.Stop()
168 //}
khenaidoocfee5f42018-07-19 22:47:38 -0400169}
170
171func waitForExit() int {
172 signalChannel := make(chan os.Signal, 1)
173 signal.Notify(signalChannel,
174 syscall.SIGHUP,
175 syscall.SIGINT,
176 syscall.SIGTERM,
177 syscall.SIGQUIT)
178
179 exitChannel := make(chan int)
180
181 go func() {
182 s := <-signalChannel
183 switch s {
184 case syscall.SIGHUP,
185 syscall.SIGINT,
186 syscall.SIGTERM,
187 syscall.SIGQUIT:
khenaidoo5c11af72018-07-20 17:21:05 -0400188 log.Infow("closing-signal-received", log.Fields{"signal": s})
khenaidoocfee5f42018-07-19 22:47:38 -0400189 exitChannel <- 0
190 default:
khenaidoo5c11af72018-07-20 17:21:05 -0400191 log.Infow("unexpected-signal-received", log.Fields{"signal": s})
khenaidoocfee5f42018-07-19 22:47:38 -0400192 exitChannel <- 1
193 }
194 }()
195
196 code := <-exitChannel
197 return code
198}
199
khenaidoo5c11af72018-07-20 17:21:05 -0400200func printBanner() {
201 fmt.Println(" ")
202 fmt.Println(" ______ ______ ")
203 fmt.Println("| _ \\ \\ / / ___|___ _ __ ___ ")
204 fmt.Println("| |_) \\ \\ /\\ / / | / _ \\| '__/ _ \\ ")
205 fmt.Println("| _ < \\ V V /| |__| (_) | | | __/ ")
206 fmt.Println("|_| \\_\\ \\_/\\_/ \\____\\___/|_| \\___| ")
207 fmt.Println(" ")
208}
209
David K. Bainbridgef430cd52019-05-28 15:00:35 -0700210func printVersion() {
211 fmt.Println("VOLTHA Read-Write Core")
212 fmt.Println(version.VersionInfo.String(" "))
213}
214
khenaidoocfee5f42018-07-19 22:47:38 -0400215func main() {
216 start := time.Now()
217
218 cf := config.NewRWCoreFlags()
219 cf.ParseCommandArguments()
220
khenaidoo631fe542019-05-31 15:44:43 -0400221 // Set the instance ID as the hostname
npujar1d86a522019-11-14 17:11:16 +0530222 var instanceID string
khenaidoo631fe542019-05-31 15:44:43 -0400223 hostName := utils.GetHostName()
224 if len(hostName) > 0 {
npujar1d86a522019-11-14 17:11:16 +0530225 instanceID = hostName
khenaidoo631fe542019-05-31 15:44:43 -0400226 } else {
227 log.Fatal("HOSTNAME not set")
228 }
Rohan Agrawal7f72f0c2020-01-14 12:05:51 +0000229
Don Newton8eca4622020-02-10 16:44:48 -0500230 realMain()
Rohan Agrawal7f72f0c2020-01-14 12:05:51 +0000231
232 logLevel, err := log.StringToLogLevel(cf.LogLevel)
233 if err != nil {
234 panic(err)
235 }
236
khenaidoob9203542018-09-17 22:56:37 -0400237 //Setup default logger - applies for packages that do not have specific logger set
Rohan Agrawal7f72f0c2020-01-14 12:05:51 +0000238 if _, err := log.SetDefaultLogger(log.JSON, logLevel, log.Fields{"instanceId": instanceID}); err != nil {
khenaidoocfee5f42018-07-19 22:47:38 -0400239 log.With(log.Fields{"error": err}).Fatal("Cannot setup logging")
240 }
khenaidoob9203542018-09-17 22:56:37 -0400241
khenaidoo631fe542019-05-31 15:44:43 -0400242 // Update all loggers (provisioned via init) with a common field
npujar1d86a522019-11-14 17:11:16 +0530243 if err := log.UpdateAllLoggers(log.Fields{"instanceId": instanceID}); err != nil {
khenaidoob9203542018-09-17 22:56:37 -0400244 log.With(log.Fields{"error": err}).Fatal("Cannot setup logging")
245 }
246
khenaidoo631fe542019-05-31 15:44:43 -0400247 // Update all loggers to log level specified as input parameter
Rohan Agrawal7f72f0c2020-01-14 12:05:51 +0000248 log.SetAllLogLevel(logLevel)
khenaidoo2c6a0992019-04-29 13:46:56 -0400249
khenaidoo631fe542019-05-31 15:44:43 -0400250 //log.SetPackageLogLevel("github.com/opencord/voltha-go/rw_core/core", log.DebugLevel)
khenaidoob9203542018-09-17 22:56:37 -0400251
npujar1d86a522019-11-14 17:11:16 +0530252 defer func() {
253 err := log.CleanUp()
254 if err != nil {
255 log.Errorw("unable-to-flush-any-buffered-log-entries", log.Fields{"error": err})
256 }
257 }()
khenaidoocfee5f42018-07-19 22:47:38 -0400258
khenaidoo631fe542019-05-31 15:44:43 -0400259 // Print version / build information and exit
David K. Bainbridgef430cd52019-05-28 15:00:35 -0700260 if cf.DisplayVersionOnly {
261 printVersion()
262 return
263 }
264
khenaidoo5c11af72018-07-20 17:21:05 -0400265 // Print banner if specified
266 if cf.Banner {
267 printBanner()
268 }
269
270 log.Infow("rw-core-config", log.Fields{"config": *cf})
271
David K. Bainbridgeb4a9ab02019-09-20 15:12:16 -0700272 // Create the core
273 rw := newRWCore(cf)
274
275 // Create a context adding the status update channel
khenaidoo5c11af72018-07-20 17:21:05 -0400276 ctx, cancel := context.WithCancel(context.Background())
277 defer cancel()
khenaidoocfee5f42018-07-19 22:47:38 -0400278
David K. Bainbridgeb4a9ab02019-09-20 15:12:16 -0700279 /*
280 * Create and start the liveness and readiness container management probes. This
281 * is done in the main function so just in case the main starts multiple other
282 * objects there can be a single probe end point for the process.
283 */
284 p := &probe.Probe{}
Kent Hagermanc4618832019-10-07 12:24:36 -0400285 go p.ListenAndServe(fmt.Sprintf("%s:%d", rw.config.ProbeHost, rw.config.ProbePort))
David K. Bainbridgeb4a9ab02019-09-20 15:12:16 -0700286
287 // Add the probe to the context to pass to all the services started
288 probeCtx := context.WithValue(ctx, probe.ProbeContextKey, p)
289
290 // Start the core
npujar1d86a522019-11-14 17:11:16 +0530291 go rw.start(probeCtx, instanceID)
khenaidoocfee5f42018-07-19 22:47:38 -0400292
293 code := waitForExit()
khenaidoo5c11af72018-07-20 17:21:05 -0400294 log.Infow("received-a-closing-signal", log.Fields{"code": code})
khenaidoocfee5f42018-07-19 22:47:38 -0400295
296 // Cleanup before leaving
David K. Bainbridgeb4a9ab02019-09-20 15:12:16 -0700297 rw.stop(probeCtx)
khenaidoocfee5f42018-07-19 22:47:38 -0400298
299 elapsed := time.Since(start)
npujar1d86a522019-11-14 17:11:16 +0530300 log.Infow("rw-core-run-time", log.Fields{"core": instanceID, "time": elapsed / time.Second})
khenaidoocfee5f42018-07-19 22:47:38 -0400301}