blob: 801b72c12871dd81fe9b2c23fa65cc7e2bc2f462 [file] [log] [blame]
khenaidoob9203542018-09-17 22:56:37 -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
khenaidoob9203542018-09-17 22:56:37 -040017package core
18
19import (
20 "context"
npujar1d86a522019-11-14 17:11:16 +053021 "time"
22
sbarbari17d7e222019-11-05 10:02:29 -050023 "github.com/opencord/voltha-go/db/model"
khenaidoob9203542018-09-17 22:56:37 -040024 "github.com/opencord/voltha-go/rw_core/config"
Kent Hagerman2b216042020-04-03 18:28:56 -040025 "github.com/opencord/voltha-go/rw_core/core/adapter"
26 "github.com/opencord/voltha-go/rw_core/core/api"
27 "github.com/opencord/voltha-go/rw_core/core/device"
Kent Hagerman2f0d0552020-04-23 17:28:52 -040028 conf "github.com/opencord/voltha-lib-go/v3/pkg/config"
serkant.uluderya2ae470f2020-01-21 11:13:09 -080029 grpcserver "github.com/opencord/voltha-lib-go/v3/pkg/grpc"
30 "github.com/opencord/voltha-lib-go/v3/pkg/kafka"
31 "github.com/opencord/voltha-lib-go/v3/pkg/log"
32 "github.com/opencord/voltha-lib-go/v3/pkg/probe"
33 "github.com/opencord/voltha-protos/v3/go/voltha"
khenaidoob9203542018-09-17 22:56:37 -040034 "google.golang.org/grpc"
khenaidoob9203542018-09-17 22:56:37 -040035)
36
npujar1d86a522019-11-14 17:11:16 +053037// Core represent read,write core attributes
khenaidoob9203542018-09-17 22:56:37 -040038type Core struct {
Kent Hagerman2f0d0552020-04-23 17:28:52 -040039 shutdown context.CancelFunc
40 stopped chan struct{}
khenaidoob9203542018-09-17 22:56:37 -040041}
42
npujar1d86a522019-11-14 17:11:16 +053043// NewCore creates instance of rw core
Kent Hagerman2f0d0552020-04-23 17:28:52 -040044func NewCore(ctx context.Context, id string, cf *config.RWCoreFlags) *Core {
David K. Bainbridgeb4a9ab02019-09-20 15:12:16 -070045 // If the context has a probe then fetch it and register our services
Kent Hagerman2f0d0552020-04-23 17:28:52 -040046 if p := probe.GetProbeFromContext(ctx); p != nil {
47 p.RegisterService(
48 "message-bus",
49 "kv-store",
50 "adapter-manager",
51 "grpc-service",
khenaidoob9203542018-09-17 22:56:37 -040052 )
53 }
54
Kent Hagerman2f0d0552020-04-23 17:28:52 -040055 // new threads will be given a new cancelable context, so that they can be aborted later when Stop() is called
56 shutdownCtx, cancelCtx := context.WithCancel(ctx)
57
58 core := &Core{shutdown: cancelCtx, stopped: make(chan struct{})}
59 go core.start(shutdownCtx, id, cf)
60 return core
61}
62
63func (core *Core) start(ctx context.Context, id string, cf *config.RWCoreFlags) {
64 logger.Info("starting-core-services", log.Fields{"coreId": id})
65
66 // deferred functions are used to run cleanup
67 // failing partway will stop anything that's been started
68 defer close(core.stopped)
69 defer core.shutdown()
70
71 logger.Info("Starting RW Core components")
72
73 // setup kv client
74 logger.Debugw("create-kv-client", log.Fields{"kvstore": cf.KVStoreType})
Neha Sharmad1387da2020-05-07 20:07:28 +000075 kvClient, err := newKVClient(cf.KVStoreType, cf.KVStoreAddress, cf.KVStoreTimeout)
Kent Hagerman2f0d0552020-04-23 17:28:52 -040076 if err != nil {
77 logger.Fatal(err)
78 }
79 defer stopKVClient(context.Background(), kvClient)
80
81 // sync logging config with kv store
Neha Sharmad1387da2020-05-07 20:07:28 +000082 cm := conf.NewConfigManager(kvClient, cf.KVStoreType, cf.KVStoreAddress, cf.KVStoreTimeout)
Kent Hagerman2f0d0552020-04-23 17:28:52 -040083 go conf.StartLogLevelConfigProcessing(cm, ctx)
84
serkant.uluderya8ff291d2020-05-20 00:58:00 -070085 backend := cm.Backend
86 backend.LivenessChannelInterval = cf.LiveProbeInterval / 2
Kent Hagerman2f0d0552020-04-23 17:28:52 -040087
88 // wait until connection to KV Store is up
89 if err := waitUntilKVStoreReachableOrMaxTries(ctx, kvClient, cf.MaxConnectionRetries, cf.ConnectionRetryInterval); err != nil {
90 logger.Fatal("Unable-to-connect-to-KV-store")
91 }
92 go monitorKVStoreLiveness(ctx, backend, cf.LiveProbeInterval, cf.NotLiveProbeInterval)
93
94 // create kafka client
95 kafkaClient := kafka.NewSaramaClient(
Neha Sharmad1387da2020-05-07 20:07:28 +000096 kafka.Address(cf.KafkaAdapterAddress),
Kent Hagerman2f0d0552020-04-23 17:28:52 -040097 kafka.ConsumerType(kafka.GroupCustomer),
98 kafka.ProducerReturnOnErrors(true),
99 kafka.ProducerReturnOnSuccess(true),
100 kafka.ProducerMaxRetries(6),
101 kafka.NumPartitions(3),
102 kafka.ConsumerGroupName(id),
103 kafka.ConsumerGroupPrefix(id),
104 kafka.AutoCreateTopic(true),
105 kafka.ProducerFlushFrequency(5),
106 kafka.ProducerRetryBackoff(time.Millisecond*30),
107 kafka.LivenessChannelInterval(cf.LiveProbeInterval/2),
108 )
109 // defer kafkaClient.Stop()
110
Kent Hagermanf5a67352020-04-30 15:15:26 -0400111 // create kv path
112 dbPath := model.NewDBPath(backend)
Kent Hagerman2f0d0552020-04-23 17:28:52 -0400113
114 // load adapters & device types while other things are starting
Kent Hagermanf5a67352020-04-30 15:15:26 -0400115 adapterMgr := adapter.NewAdapterManager(dbPath, id, kafkaClient)
Kent Hagerman2f0d0552020-04-23 17:28:52 -0400116 go adapterMgr.Start(ctx)
117
118 // connect to kafka, then wait until reachable and publisher/consumer created
119 // core.kmp must be created before deviceMgr and adapterMgr
David Bainbridge9ae13132020-06-22 17:28:01 -0700120 kmp, err := startKafkInterContainerProxy(ctx, kafkaClient, cf.KafkaAdapterAddress, cf.CoreTopic, cf.ConnectionRetryInterval)
Kent Hagerman2f0d0552020-04-23 17:28:52 -0400121 if err != nil {
122 logger.Warn("Failed to setup kafka connection")
123 return
124 }
125 defer kmp.Stop()
126 go monitorKafkaLiveness(ctx, kmp, cf.LiveProbeInterval, cf.NotLiveProbeInterval)
127
128 // create the core of the system, the device managers
129 endpointMgr := kafka.NewEndpointManager(backend)
serkant.uluderya8ff291d2020-05-20 00:58:00 -0700130 deviceMgr, logicalDeviceMgr := device.NewManagers(dbPath, adapterMgr, kmp, endpointMgr, cf.CoreTopic, id, cf.DefaultCoreTimeout)
Kent Hagerman2f0d0552020-04-23 17:28:52 -0400131
132 // register kafka RPC handler
serkant.uluderya8ff291d2020-05-20 00:58:00 -0700133 registerAdapterRequestHandlers(kmp, deviceMgr, adapterMgr, cf.CoreTopic)
Kent Hagerman2f0d0552020-04-23 17:28:52 -0400134
135 // start gRPC handler
Neha Sharmad1387da2020-05-07 20:07:28 +0000136 grpcServer := grpcserver.NewGrpcServer(cf.GrpcAddress, nil, false, probe.GetProbeFromContext(ctx))
Kent Hagerman2f0d0552020-04-23 17:28:52 -0400137 go startGRPCService(ctx, grpcServer, api.NewNBIHandler(deviceMgr, logicalDeviceMgr, adapterMgr))
138 defer grpcServer.Stop()
139
140 // wait for core to be stopped, via Stop() or context cancellation, before running deferred functions
141 <-ctx.Done()
142}
143
144// Stop brings down core services
145func (core *Core) Stop() {
146 core.shutdown()
147 <-core.stopped
148}
149
150// startGRPCService creates the grpc service handlers, registers it to the grpc server and starts the server
151func startGRPCService(ctx context.Context, server *grpcserver.GrpcServer, handler voltha.VolthaServiceServer) {
152 logger.Info("grpc-server-created")
153
154 server.AddService(func(gs *grpc.Server) { voltha.RegisterVolthaServiceServer(gs, handler) })
Girish Kumarf56a4682020-03-20 20:07:46 +0000155 logger.Info("grpc-service-added")
khenaidoob9203542018-09-17 22:56:37 -0400156
David K. Bainbridgeb4a9ab02019-09-20 15:12:16 -0700157 probe.UpdateStatusFromContext(ctx, "grpc-service", probe.ServiceStatusRunning)
Girish Kumarf56a4682020-03-20 20:07:46 +0000158 logger.Info("grpc-server-started")
Kent Hagerman2f0d0552020-04-23 17:28:52 -0400159 // Note that there is a small window here in which the core could return its status as ready,
160 // when it really isn't. This is unlikely to cause issues, as the delay is incredibly short.
161 server.Start(ctx)
David K. Bainbridgeb4a9ab02019-09-20 15:12:16 -0700162 probe.UpdateStatusFromContext(ctx, "grpc-service", probe.ServiceStatusStopped)
khenaidoob9203542018-09-17 22:56:37 -0400163}