khenaidoo | b920354 | 2018-09-17 22:56:37 -0400 | [diff] [blame] | 1 | /* |
| 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 | */ |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 16 | |
khenaidoo | b920354 | 2018-09-17 22:56:37 -0400 | [diff] [blame] | 17 | package core |
| 18 | |
| 19 | import ( |
| 20 | "context" |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 21 | "time" |
| 22 | |
sbarbari | 17d7e22 | 2019-11-05 10:02:29 -0500 | [diff] [blame] | 23 | "github.com/opencord/voltha-go/db/model" |
khenaidoo | b920354 | 2018-09-17 22:56:37 -0400 | [diff] [blame] | 24 | "github.com/opencord/voltha-go/rw_core/config" |
Kent Hagerman | 2b21604 | 2020-04-03 18:28:56 -0400 | [diff] [blame] | 25 | "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" |
Maninder | dfadc98 | 2020-10-28 14:04:33 +0530 | [diff] [blame] | 28 | conf "github.com/opencord/voltha-lib-go/v4/pkg/config" |
| 29 | grpcserver "github.com/opencord/voltha-lib-go/v4/pkg/grpc" |
| 30 | "github.com/opencord/voltha-lib-go/v4/pkg/kafka" |
| 31 | "github.com/opencord/voltha-lib-go/v4/pkg/log" |
| 32 | "github.com/opencord/voltha-lib-go/v4/pkg/probe" |
Salman Siddiqui | 1cf9504 | 2020-11-19 00:42:56 +0530 | [diff] [blame] | 33 | "github.com/opencord/voltha-protos/v4/go/extension" |
Maninder | dfadc98 | 2020-10-28 14:04:33 +0530 | [diff] [blame] | 34 | "github.com/opencord/voltha-protos/v4/go/voltha" |
khenaidoo | b920354 | 2018-09-17 22:56:37 -0400 | [diff] [blame] | 35 | "google.golang.org/grpc" |
khenaidoo | b920354 | 2018-09-17 22:56:37 -0400 | [diff] [blame] | 36 | ) |
| 37 | |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 38 | // Core represent read,write core attributes |
khenaidoo | b920354 | 2018-09-17 22:56:37 -0400 | [diff] [blame] | 39 | type Core struct { |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 40 | shutdown context.CancelFunc |
| 41 | stopped chan struct{} |
khenaidoo | b920354 | 2018-09-17 22:56:37 -0400 | [diff] [blame] | 42 | } |
| 43 | |
Matteo Scandolo | b3ba79c | 2021-03-01 10:53:23 -0800 | [diff] [blame] | 44 | const ( |
| 45 | adapterMessageBus = "adapter-message-bus" |
| 46 | clusterMessageBus = "cluster-message-bus" |
| 47 | ) |
| 48 | |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 49 | // NewCore creates instance of rw core |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 50 | func NewCore(ctx context.Context, id string, cf *config.RWCoreFlags) *Core { |
David K. Bainbridge | b4a9ab0 | 2019-09-20 15:12:16 -0700 | [diff] [blame] | 51 | // If the context has a probe then fetch it and register our services |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 52 | if p := probe.GetProbeFromContext(ctx); p != nil { |
| 53 | p.RegisterService( |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 54 | ctx, |
Matteo Scandolo | b3ba79c | 2021-03-01 10:53:23 -0800 | [diff] [blame] | 55 | adapterMessageBus, |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 56 | "kv-store", |
| 57 | "adapter-manager", |
| 58 | "grpc-service", |
khenaidoo | b920354 | 2018-09-17 22:56:37 -0400 | [diff] [blame] | 59 | ) |
Matteo Scandolo | b3ba79c | 2021-03-01 10:53:23 -0800 | [diff] [blame] | 60 | |
| 61 | if cf.KafkaAdapterAddress != cf.KafkaClusterAddress { |
| 62 | p.RegisterService( |
| 63 | ctx, |
| 64 | clusterMessageBus, |
| 65 | ) |
| 66 | } |
khenaidoo | b920354 | 2018-09-17 22:56:37 -0400 | [diff] [blame] | 67 | } |
| 68 | |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 69 | // new threads will be given a new cancelable context, so that they can be aborted later when Stop() is called |
| 70 | shutdownCtx, cancelCtx := context.WithCancel(ctx) |
| 71 | |
| 72 | core := &Core{shutdown: cancelCtx, stopped: make(chan struct{})} |
| 73 | go core.start(shutdownCtx, id, cf) |
| 74 | return core |
| 75 | } |
| 76 | |
| 77 | func (core *Core) start(ctx context.Context, id string, cf *config.RWCoreFlags) { |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 78 | logger.Info(ctx, "starting-core-services", log.Fields{"coreId": id}) |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 79 | |
| 80 | // deferred functions are used to run cleanup |
| 81 | // failing partway will stop anything that's been started |
| 82 | defer close(core.stopped) |
| 83 | defer core.shutdown() |
| 84 | |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 85 | logger.Info(ctx, "starting-rw-core-components") |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 86 | |
| 87 | // setup kv client |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 88 | logger.Debugw(ctx, "create-kv-client", log.Fields{"kvstore": cf.KVStoreType}) |
| 89 | kvClient, err := newKVClient(ctx, cf.KVStoreType, cf.KVStoreAddress, cf.KVStoreTimeout) |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 90 | if err != nil { |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 91 | logger.Fatal(ctx, err) |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 92 | } |
Rohan Agrawal | cf12f20 | 2020-08-03 04:42:01 +0000 | [diff] [blame] | 93 | defer stopKVClient(log.WithSpanFromContext(context.Background(), ctx), kvClient) |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 94 | |
| 95 | // sync logging config with kv store |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 96 | cm := conf.NewConfigManager(ctx, kvClient, cf.KVStoreType, cf.KVStoreAddress, cf.KVStoreTimeout) |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 97 | go conf.StartLogLevelConfigProcessing(cm, ctx) |
Girish Kumar | f8d4f8d | 2020-08-18 11:45:30 +0000 | [diff] [blame] | 98 | go conf.StartLogFeaturesConfigProcessing(cm, ctx) |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 99 | |
serkant.uluderya | 8ff291d | 2020-05-20 00:58:00 -0700 | [diff] [blame] | 100 | backend := cm.Backend |
| 101 | backend.LivenessChannelInterval = cf.LiveProbeInterval / 2 |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 102 | |
| 103 | // wait until connection to KV Store is up |
| 104 | if err := waitUntilKVStoreReachableOrMaxTries(ctx, kvClient, cf.MaxConnectionRetries, cf.ConnectionRetryInterval); err != nil { |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 105 | logger.Fatal(ctx, "unable-to-connect-to-kv-store") |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 106 | } |
| 107 | go monitorKVStoreLiveness(ctx, backend, cf.LiveProbeInterval, cf.NotLiveProbeInterval) |
| 108 | |
| 109 | // create kafka client |
| 110 | kafkaClient := kafka.NewSaramaClient( |
Neha Sharma | d1387da | 2020-05-07 20:07:28 +0000 | [diff] [blame] | 111 | kafka.Address(cf.KafkaAdapterAddress), |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 112 | kafka.ConsumerType(kafka.GroupCustomer), |
| 113 | kafka.ProducerReturnOnErrors(true), |
| 114 | kafka.ProducerReturnOnSuccess(true), |
| 115 | kafka.ProducerMaxRetries(6), |
| 116 | kafka.NumPartitions(3), |
| 117 | kafka.ConsumerGroupName(id), |
| 118 | kafka.ConsumerGroupPrefix(id), |
| 119 | kafka.AutoCreateTopic(true), |
| 120 | kafka.ProducerFlushFrequency(5), |
| 121 | kafka.ProducerRetryBackoff(time.Millisecond*30), |
| 122 | kafka.LivenessChannelInterval(cf.LiveProbeInterval/2), |
| 123 | ) |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 124 | |
| 125 | // create kafka client for events |
| 126 | kafkaClientEvent := kafka.NewSaramaClient( |
| 127 | kafka.Address(cf.KafkaClusterAddress), |
| 128 | kafka.ProducerReturnOnErrors(true), |
| 129 | kafka.ProducerReturnOnSuccess(true), |
| 130 | kafka.ProducerMaxRetries(6), |
| 131 | kafka.ProducerRetryBackoff(time.Millisecond*30), |
| 132 | kafka.AutoCreateTopic(true), |
| 133 | kafka.MetadatMaxRetries(15), |
| 134 | ) |
Matteo Scandolo | b3ba79c | 2021-03-01 10:53:23 -0800 | [diff] [blame] | 135 | |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 136 | // create event proxy |
Matteo Scandolo | b3ba79c | 2021-03-01 10:53:23 -0800 | [diff] [blame] | 137 | eventProxy, err := startEventProxy(ctx, kafkaClientEvent, cf.EventTopic, cf.ConnectionRetryInterval) |
| 138 | if err != nil { |
| 139 | logger.Warn(ctx, "failed-to-setup-kafka-event-proxy-connection") |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 140 | return |
| 141 | } |
Matteo Scandolo | b3ba79c | 2021-03-01 10:53:23 -0800 | [diff] [blame] | 142 | if cf.KafkaAdapterAddress != cf.KafkaClusterAddress { |
| 143 | // if we're using a single kafka cluster we don't need two liveliness probes on the same cluster |
| 144 | go monitorKafkaLiveness(ctx, eventProxy, cf.LiveProbeInterval, cf.NotLiveProbeInterval, clusterMessageBus) |
| 145 | } |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 146 | |
| 147 | defer kafkaClientEvent.Stop(ctx) |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 148 | |
Kent Hagerman | f5a6735 | 2020-04-30 15:15:26 -0400 | [diff] [blame] | 149 | // create kv path |
| 150 | dbPath := model.NewDBPath(backend) |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 151 | |
| 152 | // load adapters & device types while other things are starting |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 153 | adapterMgr := adapter.NewAdapterManager(ctx, dbPath, id, kafkaClient) |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 154 | go adapterMgr.Start(ctx) |
| 155 | |
| 156 | // connect to kafka, then wait until reachable and publisher/consumer created |
| 157 | // core.kmp must be created before deviceMgr and adapterMgr |
David Bainbridge | 9ae1313 | 2020-06-22 17:28:01 -0700 | [diff] [blame] | 158 | kmp, err := startKafkInterContainerProxy(ctx, kafkaClient, cf.KafkaAdapterAddress, cf.CoreTopic, cf.ConnectionRetryInterval) |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 159 | if err != nil { |
Matteo Scandolo | b3ba79c | 2021-03-01 10:53:23 -0800 | [diff] [blame] | 160 | logger.Warn(ctx, "failed-to-setup-kafka-adapter-proxy-connection") |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 161 | return |
| 162 | } |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 163 | defer kmp.Stop(ctx) |
Matteo Scandolo | b3ba79c | 2021-03-01 10:53:23 -0800 | [diff] [blame] | 164 | go monitorKafkaLiveness(ctx, kmp, cf.LiveProbeInterval, cf.NotLiveProbeInterval, adapterMessageBus) |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 165 | |
| 166 | // create the core of the system, the device managers |
| 167 | endpointMgr := kafka.NewEndpointManager(backend) |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 168 | deviceMgr, logicalDeviceMgr := device.NewManagers(dbPath, adapterMgr, kmp, endpointMgr, cf.CoreTopic, id, cf.DefaultCoreTimeout, eventProxy) |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 169 | |
| 170 | // register kafka RPC handler |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 171 | registerAdapterRequestHandlers(ctx, kmp, deviceMgr, adapterMgr, cf.CoreTopic) |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 172 | |
| 173 | // start gRPC handler |
Neha Sharma | d1387da | 2020-05-07 20:07:28 +0000 | [diff] [blame] | 174 | grpcServer := grpcserver.NewGrpcServer(cf.GrpcAddress, nil, false, probe.GetProbeFromContext(ctx)) |
Salman Siddiqui | 1cf9504 | 2020-11-19 00:42:56 +0530 | [diff] [blame] | 175 | |
| 176 | //Register the 'Extension' service on this gRPC server |
| 177 | addGRPCExtensionService(ctx, grpcServer, device.GetNewExtensionManager(deviceMgr)) |
| 178 | |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 179 | go startGRPCService(ctx, grpcServer, api.NewNBIHandler(deviceMgr, logicalDeviceMgr, adapterMgr)) |
| 180 | defer grpcServer.Stop() |
| 181 | |
| 182 | // wait for core to be stopped, via Stop() or context cancellation, before running deferred functions |
| 183 | <-ctx.Done() |
| 184 | } |
| 185 | |
| 186 | // Stop brings down core services |
| 187 | func (core *Core) Stop() { |
| 188 | core.shutdown() |
| 189 | <-core.stopped |
| 190 | } |
| 191 | |
| 192 | // startGRPCService creates the grpc service handlers, registers it to the grpc server and starts the server |
| 193 | func startGRPCService(ctx context.Context, server *grpcserver.GrpcServer, handler voltha.VolthaServiceServer) { |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 194 | logger.Info(ctx, "grpc-server-created") |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 195 | |
| 196 | server.AddService(func(gs *grpc.Server) { voltha.RegisterVolthaServiceServer(gs, handler) }) |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 197 | logger.Info(ctx, "grpc-service-added") |
khenaidoo | b920354 | 2018-09-17 22:56:37 -0400 | [diff] [blame] | 198 | |
David K. Bainbridge | b4a9ab0 | 2019-09-20 15:12:16 -0700 | [diff] [blame] | 199 | probe.UpdateStatusFromContext(ctx, "grpc-service", probe.ServiceStatusRunning) |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 200 | logger.Info(ctx, "grpc-server-started") |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 201 | // Note that there is a small window here in which the core could return its status as ready, |
| 202 | // when it really isn't. This is unlikely to cause issues, as the delay is incredibly short. |
| 203 | server.Start(ctx) |
David K. Bainbridge | b4a9ab0 | 2019-09-20 15:12:16 -0700 | [diff] [blame] | 204 | probe.UpdateStatusFromContext(ctx, "grpc-service", probe.ServiceStatusStopped) |
khenaidoo | b920354 | 2018-09-17 22:56:37 -0400 | [diff] [blame] | 205 | } |
Salman Siddiqui | 1cf9504 | 2020-11-19 00:42:56 +0530 | [diff] [blame] | 206 | |
| 207 | func addGRPCExtensionService(ctx context.Context, server *grpcserver.GrpcServer, handler extension.ExtensionServer) { |
| 208 | logger.Info(ctx, "extension-grpc-server-created") |
| 209 | |
| 210 | server.AddService(func(server *grpc.Server) { |
| 211 | extension.RegisterExtensionServer(server, handler) |
| 212 | }) |
| 213 | |
| 214 | } |