blob: 76d0952616e4aa019a5224bf9102e6cc27e5b112 [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"
32 "github.com/opencord/voltha-lib-go/v2/pkg/db/kvstore"
33 "github.com/opencord/voltha-lib-go/v2/pkg/kafka"
34 "github.com/opencord/voltha-lib-go/v2/pkg/log"
35 "github.com/opencord/voltha-lib-go/v2/pkg/probe"
36 "github.com/opencord/voltha-lib-go/v2/pkg/version"
37 ic "github.com/opencord/voltha-protos/v2/go/inter_container"
khenaidoocfee5f42018-07-19 22:47:38 -040038)
39
40type rwCore struct {
khenaidoo5c11af72018-07-20 17:21:05 -040041 kvClient kvstore.Client
42 config *config.RWCoreFlags
43 halted bool
44 exitChannel chan int
khenaidoob9203542018-09-17 22:56:37 -040045 //kmp *kafka.KafkaMessagingProxy
khenaidoo43c82122018-11-22 18:38:28 -050046 kafkaClient kafka.Client
47 core *c.Core
khenaidooabad44c2018-08-03 16:58:35 -040048 //For test
khenaidoo79232702018-12-04 11:00:41 -050049 receiverChannels []<-chan *ic.InterContainerMessage
khenaidoocfee5f42018-07-19 22:47:38 -040050}
51
khenaidoob9203542018-09-17 22:56:37 -040052func init() {
npujar1d86a522019-11-14 17:11:16 +053053 _, err := log.AddPackage(log.JSON, log.DebugLevel, nil)
54 if err != nil {
55 log.Errorw("unable-to-register-package-to-the-log-map", log.Fields{"error": err})
56 }
khenaidoob9203542018-09-17 22:56:37 -040057}
58
khenaidoocfee5f42018-07-19 22:47:38 -040059func newKVClient(storeType string, address string, timeout int) (kvstore.Client, error) {
60
khenaidoo5c11af72018-07-20 17:21:05 -040061 log.Infow("kv-store-type", log.Fields{"store": storeType})
khenaidoocfee5f42018-07-19 22:47:38 -040062 switch storeType {
63 case "consul":
64 return kvstore.NewConsulClient(address, timeout)
65 case "etcd":
66 return kvstore.NewEtcdClient(address, timeout)
67 }
68 return nil, errors.New("unsupported-kv-store")
69}
70
Scott Bakeree6a0872019-10-29 15:59:52 -070071func newKafkaClient(clientType string, host string, port int, instanceID string, livenessChannelInterval time.Duration) (kafka.Client, error) {
khenaidoo43c82122018-11-22 18:38:28 -050072
73 log.Infow("kafka-client-type", log.Fields{"client": clientType})
74 switch clientType {
75 case "sarama":
76 return kafka.NewSaramaClient(
77 kafka.Host(host),
khenaidoo90847922018-12-03 14:47:51 -050078 kafka.Port(port),
khenaidooca301322019-01-09 23:06:32 -050079 kafka.ConsumerType(kafka.GroupCustomer),
khenaidoo90847922018-12-03 14:47:51 -050080 kafka.ProducerReturnOnErrors(true),
81 kafka.ProducerReturnOnSuccess(true),
82 kafka.ProducerMaxRetries(6),
khenaidooca301322019-01-09 23:06:32 -050083 kafka.NumPartitions(3),
84 kafka.ConsumerGroupName(instanceID),
85 kafka.ConsumerGroupPrefix(instanceID),
khenaidoo54e0ddf2019-02-27 16:21:33 -050086 kafka.AutoCreateTopic(true),
khenaidooca301322019-01-09 23:06:32 -050087 kafka.ProducerFlushFrequency(5),
Scott Bakeree6a0872019-10-29 15:59:52 -070088 kafka.ProducerRetryBackoff(time.Millisecond*30),
89 kafka.LivenessChannelInterval(livenessChannelInterval),
90 ), nil
khenaidoo43c82122018-11-22 18:38:28 -050091 }
92 return nil, errors.New("unsupported-client-type")
93}
94
khenaidoocfee5f42018-07-19 22:47:38 -040095func newRWCore(cf *config.RWCoreFlags) *rwCore {
96 var rwCore rwCore
97 rwCore.config = cf
98 rwCore.halted = false
99 rwCore.exitChannel = make(chan int, 1)
khenaidoo79232702018-12-04 11:00:41 -0500100 rwCore.receiverChannels = make([]<-chan *ic.InterContainerMessage, 0)
khenaidoocfee5f42018-07-19 22:47:38 -0400101 return &rwCore
102}
103
khenaidoob9203542018-09-17 22:56:37 -0400104func (rw *rwCore) setKVClient() error {
105 addr := rw.config.KVStoreHost + ":" + strconv.Itoa(rw.config.KVStorePort)
106 client, err := newKVClient(rw.config.KVStoreType, addr, rw.config.KVStoreTimeout)
khenaidoocfee5f42018-07-19 22:47:38 -0400107 if err != nil {
Richard Jankowskie4d77662018-10-17 13:53:21 -0400108 rw.kvClient = nil
khenaidoocfee5f42018-07-19 22:47:38 -0400109 log.Error(err)
110 return err
111 }
khenaidoob9203542018-09-17 22:56:37 -0400112 rw.kvClient = client
khenaidoocfee5f42018-07-19 22:47:38 -0400113 return nil
114}
115
npujar1d86a522019-11-14 17:11:16 +0530116func (rw *rwCore) start(ctx context.Context, instanceID string) {
khenaidoo5c11af72018-07-20 17:21:05 -0400117 log.Info("Starting RW Core components")
khenaidoob9203542018-09-17 22:56:37 -0400118
khenaidoo90847922018-12-03 14:47:51 -0500119 // Setup KV Client
Richard Jankowskie4d77662018-10-17 13:53:21 -0400120 log.Debugw("create-kv-client", log.Fields{"kvstore": rw.config.KVStoreType})
121 err := rw.setKVClient()
122 if err == nil {
123 // Setup KV transaction context
khenaidoo9cdc1a62019-01-24 21:57:40 -0500124 txnPrefix := rw.config.KVStoreDataPrefix + "/transactions/"
npujar1d86a522019-11-14 17:11:16 +0530125 if err = c.SetTransactionContext(instanceID,
khenaidoo9cdc1a62019-01-24 21:57:40 -0500126 txnPrefix,
Richard Jankowskie4d77662018-10-17 13:53:21 -0400127 rw.kvClient,
khenaidoo09771ef2019-10-11 14:25:02 -0400128 rw.config.KVStoreTimeout); err != nil {
khenaidoo9cdc1a62019-01-24 21:57:40 -0500129 log.Fatal("creating-transaction-context-failed")
130 }
Richard Jankowskie4d77662018-10-17 13:53:21 -0400131 }
132
khenaidoo90847922018-12-03 14:47:51 -0500133 // Setup Kafka Client
Scott Bakeree6a0872019-10-29 15:59:52 -0700134 if rw.kafkaClient, err = newKafkaClient("sarama",
135 rw.config.KafkaAdapterHost,
136 rw.config.KafkaAdapterPort,
npujar1d86a522019-11-14 17:11:16 +0530137 instanceID,
Girish Kumar4d3887d2019-11-22 14:22:05 +0000138 rw.config.LiveProbeInterval/2); err != nil {
khenaidoo43c82122018-11-22 18:38:28 -0500139 log.Fatal("Unsupported-kafka-client")
140 }
141
khenaidoob9203542018-09-17 22:56:37 -0400142 // Create the core service
Thomas Lee Se5a44012019-11-07 20:32:24 +0530143 rw.core = c.NewCore(ctx, instanceID, rw.config, rw.kvClient, rw.kafkaClient)
khenaidoob9203542018-09-17 22:56:37 -0400144
145 // start the core
Thomas Lee Se5a44012019-11-07 20:32:24 +0530146 err = rw.core.Start(ctx)
147 if err != nil {
148 log.Fatalf("failed-to-start-rwcore", log.Fields{"error": err})
149 }
khenaidoocfee5f42018-07-19 22:47:38 -0400150}
151
David K. Bainbridgeb4a9ab02019-09-20 15:12:16 -0700152func (rw *rwCore) stop(ctx context.Context) {
khenaidoocfee5f42018-07-19 22:47:38 -0400153 // Stop leadership tracking
khenaidoob9203542018-09-17 22:56:37 -0400154 rw.halted = true
khenaidoocfee5f42018-07-19 22:47:38 -0400155
156 // send exit signal
khenaidoob9203542018-09-17 22:56:37 -0400157 rw.exitChannel <- 0
khenaidoocfee5f42018-07-19 22:47:38 -0400158
159 // Cleanup - applies only if we had a kvClient
khenaidoob9203542018-09-17 22:56:37 -0400160 if rw.kvClient != nil {
khenaidoocfee5f42018-07-19 22:47:38 -0400161 // Release all reservations
khenaidoob9203542018-09-17 22:56:37 -0400162 if err := rw.kvClient.ReleaseAllReservations(); err != nil {
khenaidoo5c11af72018-07-20 17:21:05 -0400163 log.Infow("fail-to-release-all-reservations", log.Fields{"error": err})
khenaidoocfee5f42018-07-19 22:47:38 -0400164 }
165 // Close the DB connection
khenaidoob9203542018-09-17 22:56:37 -0400166 rw.kvClient.Close()
khenaidoocfee5f42018-07-19 22:47:38 -0400167 }
khenaidoo43c82122018-11-22 18:38:28 -0500168
David K. Bainbridgeb4a9ab02019-09-20 15:12:16 -0700169 rw.core.Stop(ctx)
khenaidoo43c82122018-11-22 18:38:28 -0500170
171 //if rw.kafkaClient != nil {
172 // rw.kafkaClient.Stop()
173 //}
khenaidoocfee5f42018-07-19 22:47:38 -0400174}
175
176func waitForExit() int {
177 signalChannel := make(chan os.Signal, 1)
178 signal.Notify(signalChannel,
179 syscall.SIGHUP,
180 syscall.SIGINT,
181 syscall.SIGTERM,
182 syscall.SIGQUIT)
183
184 exitChannel := make(chan int)
185
186 go func() {
187 s := <-signalChannel
188 switch s {
189 case syscall.SIGHUP,
190 syscall.SIGINT,
191 syscall.SIGTERM,
192 syscall.SIGQUIT:
khenaidoo5c11af72018-07-20 17:21:05 -0400193 log.Infow("closing-signal-received", log.Fields{"signal": s})
khenaidoocfee5f42018-07-19 22:47:38 -0400194 exitChannel <- 0
195 default:
khenaidoo5c11af72018-07-20 17:21:05 -0400196 log.Infow("unexpected-signal-received", log.Fields{"signal": s})
khenaidoocfee5f42018-07-19 22:47:38 -0400197 exitChannel <- 1
198 }
199 }()
200
201 code := <-exitChannel
202 return code
203}
204
khenaidoo5c11af72018-07-20 17:21:05 -0400205func printBanner() {
206 fmt.Println(" ")
207 fmt.Println(" ______ ______ ")
208 fmt.Println("| _ \\ \\ / / ___|___ _ __ ___ ")
209 fmt.Println("| |_) \\ \\ /\\ / / | / _ \\| '__/ _ \\ ")
210 fmt.Println("| _ < \\ V V /| |__| (_) | | | __/ ")
211 fmt.Println("|_| \\_\\ \\_/\\_/ \\____\\___/|_| \\___| ")
212 fmt.Println(" ")
213}
214
David K. Bainbridgef430cd52019-05-28 15:00:35 -0700215func printVersion() {
216 fmt.Println("VOLTHA Read-Write Core")
217 fmt.Println(version.VersionInfo.String(" "))
218}
219
khenaidoocfee5f42018-07-19 22:47:38 -0400220func main() {
221 start := time.Now()
222
223 cf := config.NewRWCoreFlags()
224 cf.ParseCommandArguments()
225
khenaidoo631fe542019-05-31 15:44:43 -0400226 // Set the instance ID as the hostname
npujar1d86a522019-11-14 17:11:16 +0530227 var instanceID string
khenaidoo631fe542019-05-31 15:44:43 -0400228 hostName := utils.GetHostName()
229 if len(hostName) > 0 {
npujar1d86a522019-11-14 17:11:16 +0530230 instanceID = hostName
khenaidoo631fe542019-05-31 15:44:43 -0400231 } else {
232 log.Fatal("HOSTNAME not set")
233 }
khenaidoob9203542018-09-17 22:56:37 -0400234
235 //Setup default logger - applies for packages that do not have specific logger set
npujar1d86a522019-11-14 17:11:16 +0530236 if _, err := log.SetDefaultLogger(log.JSON, cf.LogLevel, log.Fields{"instanceId": instanceID}); err != nil {
khenaidoocfee5f42018-07-19 22:47:38 -0400237 log.With(log.Fields{"error": err}).Fatal("Cannot setup logging")
238 }
khenaidoob9203542018-09-17 22:56:37 -0400239
khenaidoo631fe542019-05-31 15:44:43 -0400240 // Update all loggers (provisioned via init) with a common field
npujar1d86a522019-11-14 17:11:16 +0530241 if err := log.UpdateAllLoggers(log.Fields{"instanceId": instanceID}); err != nil {
khenaidoob9203542018-09-17 22:56:37 -0400242 log.With(log.Fields{"error": err}).Fatal("Cannot setup logging")
243 }
244
khenaidoo631fe542019-05-31 15:44:43 -0400245 // Update all loggers to log level specified as input parameter
246 log.SetAllLogLevel(cf.LogLevel)
khenaidoo2c6a0992019-04-29 13:46:56 -0400247
khenaidoo631fe542019-05-31 15:44:43 -0400248 //log.SetPackageLogLevel("github.com/opencord/voltha-go/rw_core/core", log.DebugLevel)
khenaidoob9203542018-09-17 22:56:37 -0400249
npujar1d86a522019-11-14 17:11:16 +0530250 defer func() {
251 err := log.CleanUp()
252 if err != nil {
253 log.Errorw("unable-to-flush-any-buffered-log-entries", log.Fields{"error": err})
254 }
255 }()
khenaidoocfee5f42018-07-19 22:47:38 -0400256
khenaidoo631fe542019-05-31 15:44:43 -0400257 // Print version / build information and exit
David K. Bainbridgef430cd52019-05-28 15:00:35 -0700258 if cf.DisplayVersionOnly {
259 printVersion()
260 return
261 }
262
khenaidoo5c11af72018-07-20 17:21:05 -0400263 // Print banner if specified
264 if cf.Banner {
265 printBanner()
266 }
267
268 log.Infow("rw-core-config", log.Fields{"config": *cf})
269
David K. Bainbridgeb4a9ab02019-09-20 15:12:16 -0700270 // Create the core
271 rw := newRWCore(cf)
272
273 // Create a context adding the status update channel
khenaidoo5c11af72018-07-20 17:21:05 -0400274 ctx, cancel := context.WithCancel(context.Background())
275 defer cancel()
khenaidoocfee5f42018-07-19 22:47:38 -0400276
David K. Bainbridgeb4a9ab02019-09-20 15:12:16 -0700277 /*
278 * Create and start the liveness and readiness container management probes. This
279 * is done in the main function so just in case the main starts multiple other
280 * objects there can be a single probe end point for the process.
281 */
282 p := &probe.Probe{}
Kent Hagermanc4618832019-10-07 12:24:36 -0400283 go p.ListenAndServe(fmt.Sprintf("%s:%d", rw.config.ProbeHost, rw.config.ProbePort))
David K. Bainbridgeb4a9ab02019-09-20 15:12:16 -0700284
285 // Add the probe to the context to pass to all the services started
286 probeCtx := context.WithValue(ctx, probe.ProbeContextKey, p)
287
288 // Start the core
npujar1d86a522019-11-14 17:11:16 +0530289 go rw.start(probeCtx, instanceID)
khenaidoocfee5f42018-07-19 22:47:38 -0400290
291 code := waitForExit()
khenaidoo5c11af72018-07-20 17:21:05 -0400292 log.Infow("received-a-closing-signal", log.Fields{"code": code})
khenaidoocfee5f42018-07-19 22:47:38 -0400293
294 // Cleanup before leaving
David K. Bainbridgeb4a9ab02019-09-20 15:12:16 -0700295 rw.stop(probeCtx)
khenaidoocfee5f42018-07-19 22:47:38 -0400296
297 elapsed := time.Since(start)
npujar1d86a522019-11-14 17:11:16 +0530298 log.Infow("rw-core-run-time", log.Fields{"core": instanceID, "time": elapsed / time.Second})
khenaidoocfee5f42018-07-19 22:47:38 -0400299}