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" |
khenaidoo | d948f77 | 2021-08-11 17:49:24 -0400 | [diff] [blame] | 28 | conf "github.com/opencord/voltha-lib-go/v7/pkg/config" |
| 29 | "github.com/opencord/voltha-lib-go/v7/pkg/events" |
| 30 | grpcserver "github.com/opencord/voltha-lib-go/v7/pkg/grpc" |
| 31 | "github.com/opencord/voltha-lib-go/v7/pkg/kafka" |
| 32 | "github.com/opencord/voltha-lib-go/v7/pkg/log" |
| 33 | "github.com/opencord/voltha-lib-go/v7/pkg/probe" |
khenaidoo | 9beaaf1 | 2021-10-19 17:32:01 -0400 | [diff] [blame] | 34 | "github.com/opencord/voltha-protos/v5/go/core_service" |
khenaidoo | d948f77 | 2021-08-11 17:49:24 -0400 | [diff] [blame] | 35 | "github.com/opencord/voltha-protos/v5/go/extension" |
| 36 | "github.com/opencord/voltha-protos/v5/go/voltha" |
khenaidoo | b920354 | 2018-09-17 22:56:37 -0400 | [diff] [blame] | 37 | "google.golang.org/grpc" |
khenaidoo | b920354 | 2018-09-17 22:56:37 -0400 | [diff] [blame] | 38 | ) |
| 39 | |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 40 | // Core represent read,write core attributes |
khenaidoo | b920354 | 2018-09-17 22:56:37 -0400 | [diff] [blame] | 41 | type Core struct { |
khenaidoo | d948f77 | 2021-08-11 17:49:24 -0400 | [diff] [blame] | 42 | Shutdown context.CancelFunc |
| 43 | Stopped chan struct{} |
| 44 | KafkaClient kafka.Client |
khenaidoo | b920354 | 2018-09-17 22:56:37 -0400 | [diff] [blame] | 45 | } |
| 46 | |
Matteo Scandolo | b3ba79c | 2021-03-01 10:53:23 -0800 | [diff] [blame] | 47 | const ( |
khenaidoo | d948f77 | 2021-08-11 17:49:24 -0400 | [diff] [blame] | 48 | clusterMessagingService = "cluster-message-service" |
| 49 | grpcNBIService = "grpc-nbi-service" |
| 50 | grpcSBIService = "grpc-sbi-service" |
| 51 | adapterService = "adapter-service" |
| 52 | kvService = "kv-service" |
| 53 | deviceService = "device-service" |
| 54 | logicalDeviceService = "logical-device-service" |
Matteo Scandolo | b3ba79c | 2021-03-01 10:53:23 -0800 | [diff] [blame] | 55 | ) |
| 56 | |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 57 | // NewCore creates instance of rw core |
khenaidoo | d948f77 | 2021-08-11 17:49:24 -0400 | [diff] [blame] | 58 | func NewCore(ctx context.Context, id string, cf *config.RWCoreFlags) (*Core, context.Context) { |
David K. Bainbridge | b4a9ab0 | 2019-09-20 15:12:16 -0700 | [diff] [blame] | 59 | // 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] | 60 | if p := probe.GetProbeFromContext(ctx); p != nil { |
| 61 | p.RegisterService( |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 62 | ctx, |
khenaidoo | d948f77 | 2021-08-11 17:49:24 -0400 | [diff] [blame] | 63 | kvService, |
| 64 | adapterService, |
| 65 | grpcSBIService, |
| 66 | clusterMessagingService, |
| 67 | deviceService, |
| 68 | logicalDeviceService, |
khenaidoo | b920354 | 2018-09-17 22:56:37 -0400 | [diff] [blame] | 69 | ) |
| 70 | } |
| 71 | |
khenaidoo | d948f77 | 2021-08-11 17:49:24 -0400 | [diff] [blame] | 72 | // create kafka client for events |
| 73 | KafkaClient := kafka.NewSaramaClient( |
| 74 | kafka.Address(cf.KafkaClusterAddress), |
| 75 | kafka.ProducerReturnOnErrors(true), |
| 76 | kafka.ProducerReturnOnSuccess(true), |
| 77 | kafka.ProducerMaxRetries(6), |
| 78 | kafka.ProducerRetryBackoff(time.Millisecond*30), |
| 79 | kafka.AutoCreateTopic(true), |
| 80 | kafka.MetadatMaxRetries(15), |
| 81 | ) |
| 82 | |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 83 | // new threads will be given a new cancelable context, so that they can be aborted later when Stop() is called |
| 84 | shutdownCtx, cancelCtx := context.WithCancel(ctx) |
| 85 | |
khenaidoo | d948f77 | 2021-08-11 17:49:24 -0400 | [diff] [blame] | 86 | rwCore := &Core{Shutdown: cancelCtx, Stopped: make(chan struct{}), KafkaClient: KafkaClient} |
| 87 | return rwCore, shutdownCtx |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 88 | } |
| 89 | |
khenaidoo | d948f77 | 2021-08-11 17:49:24 -0400 | [diff] [blame] | 90 | func (core *Core) Start(ctx context.Context, id string, cf *config.RWCoreFlags) { |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 91 | logger.Info(ctx, "starting-core-services", log.Fields{"coreId": id}) |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 92 | |
| 93 | // deferred functions are used to run cleanup |
| 94 | // failing partway will stop anything that's been started |
khenaidoo | d948f77 | 2021-08-11 17:49:24 -0400 | [diff] [blame] | 95 | defer close(core.Stopped) |
| 96 | defer core.Shutdown() |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 97 | |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 98 | logger.Info(ctx, "starting-rw-core-components") |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 99 | |
| 100 | // setup kv client |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 101 | logger.Debugw(ctx, "create-kv-client", log.Fields{"kvstore": cf.KVStoreType}) |
| 102 | kvClient, err := newKVClient(ctx, cf.KVStoreType, cf.KVStoreAddress, cf.KVStoreTimeout) |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 103 | if err != nil { |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 104 | logger.Fatal(ctx, err) |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 105 | } |
Rohan Agrawal | cf12f20 | 2020-08-03 04:42:01 +0000 | [diff] [blame] | 106 | defer stopKVClient(log.WithSpanFromContext(context.Background(), ctx), kvClient) |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 107 | |
| 108 | // sync logging config with kv store |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 109 | cm := conf.NewConfigManager(ctx, kvClient, cf.KVStoreType, cf.KVStoreAddress, cf.KVStoreTimeout) |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 110 | go conf.StartLogLevelConfigProcessing(cm, ctx) |
Girish Kumar | f8d4f8d | 2020-08-18 11:45:30 +0000 | [diff] [blame] | 111 | go conf.StartLogFeaturesConfigProcessing(cm, ctx) |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 112 | |
serkant.uluderya | 8ff291d | 2020-05-20 00:58:00 -0700 | [diff] [blame] | 113 | backend := cm.Backend |
| 114 | backend.LivenessChannelInterval = cf.LiveProbeInterval / 2 |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 115 | |
| 116 | // wait until connection to KV Store is up |
khenaidoo | d948f77 | 2021-08-11 17:49:24 -0400 | [diff] [blame] | 117 | if err := waitUntilKVStoreReachableOrMaxTries(ctx, kvClient, cf.MaxConnectionRetries, cf.ConnectionRetryInterval, kvService); err != nil { |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 118 | logger.Fatal(ctx, "unable-to-connect-to-kv-store") |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 119 | } |
khenaidoo | d948f77 | 2021-08-11 17:49:24 -0400 | [diff] [blame] | 120 | go monitorKVStoreLiveness(ctx, backend, kvService, cf.LiveProbeInterval, cf.NotLiveProbeInterval) |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 121 | |
khenaidoo | d948f77 | 2021-08-11 17:49:24 -0400 | [diff] [blame] | 122 | // Start kafka communications and artefacts |
| 123 | if err := kafka.StartAndWaitUntilKafkaConnectionIsUp(ctx, core.KafkaClient, cf.ConnectionRetryInterval, clusterMessagingService); err != nil { |
| 124 | logger.Fatal(ctx, "unable-to-connect-to-kafka") |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 125 | } |
khenaidoo | d948f77 | 2021-08-11 17:49:24 -0400 | [diff] [blame] | 126 | defer core.KafkaClient.Stop(ctx) |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 127 | |
khenaidoo | d948f77 | 2021-08-11 17:49:24 -0400 | [diff] [blame] | 128 | // Create the event proxy to post events to KAFKA |
| 129 | eventProxy := events.NewEventProxy(events.MsgClient(core.KafkaClient), events.MsgTopic(kafka.Topic{Name: cf.EventTopic})) |
| 130 | go func() { |
| 131 | if err := eventProxy.Start(); err != nil { |
| 132 | logger.Fatalw(ctx, "event-proxy-cannot-start", log.Fields{"error": err}) |
| 133 | } |
| 134 | }() |
| 135 | defer eventProxy.Stop() |
| 136 | |
| 137 | // Start the kafka monitoring routine |
| 138 | go kafka.MonitorKafkaReadiness(ctx, core.KafkaClient, cf.LiveProbeInterval, cf.NotLiveProbeInterval, clusterMessagingService) |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 139 | |
Kent Hagerman | f5a6735 | 2020-04-30 15:15:26 -0400 | [diff] [blame] | 140 | // create kv path |
| 141 | dbPath := model.NewDBPath(backend) |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 142 | |
| 143 | // load adapters & device types while other things are starting |
khenaidoo | 25057da | 2021-12-08 14:40:45 -0500 | [diff] [blame] | 144 | adapterMgr := adapter.NewAdapterManager(cf.GrpcSBIAddress, dbPath, id, backend, cf.LiveProbeInterval) |
khenaidoo | d948f77 | 2021-08-11 17:49:24 -0400 | [diff] [blame] | 145 | adapterMgr.Start(ctx, adapterService) |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 146 | |
| 147 | // create the core of the system, the device managers |
khenaidoo | d948f77 | 2021-08-11 17:49:24 -0400 | [diff] [blame] | 148 | deviceMgr, logicalDeviceMgr := device.NewManagers(dbPath, adapterMgr, cf, id, eventProxy) |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 149 | |
khenaidoo | 7585a96 | 2021-06-10 16:15:38 -0400 | [diff] [blame] | 150 | // Start the device manager to load the devices. Wait until it is completed to prevent multiple loading happening |
| 151 | // triggered by logicalDeviceMgr.Start(Ctx) |
khenaidoo | d948f77 | 2021-08-11 17:49:24 -0400 | [diff] [blame] | 152 | err = deviceMgr.Start(ctx, deviceService) |
| 153 | if err != nil { |
| 154 | logger.Fatalw(ctx, "failure-starting-device-manager", log.Fields{"error": err}) |
| 155 | } |
khenaidoo | 7585a96 | 2021-06-10 16:15:38 -0400 | [diff] [blame] | 156 | |
| 157 | // Start the logical device manager to load the logical devices. |
khenaidoo | d948f77 | 2021-08-11 17:49:24 -0400 | [diff] [blame] | 158 | logicalDeviceMgr.Start(ctx, logicalDeviceService) |
khenaidoo | 7585a96 | 2021-06-10 16:15:38 -0400 | [diff] [blame] | 159 | |
khenaidoo | d948f77 | 2021-08-11 17:49:24 -0400 | [diff] [blame] | 160 | // Create and start the SBI gRPC service |
| 161 | grpcSBIServer := grpcserver.NewGrpcServer(cf.GrpcSBIAddress, nil, false, probe.GetProbeFromContext(ctx)) |
| 162 | go startGrpcSbiService(ctx, grpcSBIServer, grpcSBIService, api.NewAPIHandler(deviceMgr, nil, adapterMgr)) |
| 163 | defer grpcSBIServer.Stop() |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 164 | |
khenaidoo | d948f77 | 2021-08-11 17:49:24 -0400 | [diff] [blame] | 165 | // In the case of a restart, let's wait until all the registered adapters are connected to the Core |
| 166 | // before starting the grpc server that handles NBI requests. |
| 167 | err = adapterMgr.WaitUntilConnectionsToAdaptersAreUp(ctx, cf.ConnectionRetryInterval) |
| 168 | if err != nil { |
| 169 | logger.Fatalw(ctx, "failure-connecting-to-adapters", log.Fields{"error": err}) |
| 170 | } |
| 171 | |
| 172 | // Create the NBI gRPC server |
| 173 | grpcNBIServer := grpcserver.NewGrpcServer(cf.GrpcNBIAddress, nil, false, probe.GetProbeFromContext(ctx)) |
Salman Siddiqui | 1cf9504 | 2020-11-19 00:42:56 +0530 | [diff] [blame] | 174 | |
| 175 | //Register the 'Extension' service on this gRPC server |
khenaidoo | d948f77 | 2021-08-11 17:49:24 -0400 | [diff] [blame] | 176 | addGRPCExtensionService(ctx, grpcNBIServer, device.GetNewExtensionManager(deviceMgr)) |
Salman Siddiqui | 1cf9504 | 2020-11-19 00:42:56 +0530 | [diff] [blame] | 177 | |
khenaidoo | d948f77 | 2021-08-11 17:49:24 -0400 | [diff] [blame] | 178 | go startGrpcNbiService(ctx, grpcNBIServer, grpcNBIService, api.NewAPIHandler(deviceMgr, logicalDeviceMgr, adapterMgr)) |
| 179 | defer grpcNBIServer.Stop() |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 180 | |
| 181 | // wait for core to be stopped, via Stop() or context cancellation, before running deferred functions |
| 182 | <-ctx.Done() |
| 183 | } |
| 184 | |
| 185 | // Stop brings down core services |
| 186 | func (core *Core) Stop() { |
khenaidoo | d948f77 | 2021-08-11 17:49:24 -0400 | [diff] [blame] | 187 | core.Shutdown() |
| 188 | <-core.Stopped |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 189 | } |
| 190 | |
khenaidoo | d948f77 | 2021-08-11 17:49:24 -0400 | [diff] [blame] | 191 | // startGrpcSbiService creates the grpc core service handlers, registers it to the grpc server and starts the server |
khenaidoo | 9beaaf1 | 2021-10-19 17:32:01 -0400 | [diff] [blame] | 192 | func startGrpcSbiService(ctx context.Context, server *grpcserver.GrpcServer, serviceName string, handler core_service.CoreServiceServer) { |
khenaidoo | d948f77 | 2021-08-11 17:49:24 -0400 | [diff] [blame] | 193 | logger.Infow(ctx, "starting-grpc-sbi-service", log.Fields{"service": serviceName}) |
| 194 | |
khenaidoo | 9beaaf1 | 2021-10-19 17:32:01 -0400 | [diff] [blame] | 195 | server.AddService(func(server *grpc.Server) { core_service.RegisterCoreServiceServer(server, handler) }) |
khenaidoo | d948f77 | 2021-08-11 17:49:24 -0400 | [diff] [blame] | 196 | logger.Infow(ctx, "grpc-sbi-service-added", log.Fields{"service": serviceName}) |
| 197 | |
| 198 | probe.UpdateStatusFromContext(ctx, serviceName, probe.ServiceStatusRunning) |
| 199 | logger.Infow(ctx, "grpc-sbi-server-started", log.Fields{"service": serviceName}) |
| 200 | server.Start(ctx) |
| 201 | probe.UpdateStatusFromContext(ctx, serviceName, probe.ServiceStatusStopped) |
| 202 | } |
| 203 | |
| 204 | // startGrpcNbiService creates the grpc NBI service handlers, registers it to the grpc server and starts the server |
| 205 | func startGrpcNbiService(ctx context.Context, server *grpcserver.GrpcServer, serviceName string, handler voltha.VolthaServiceServer) { |
| 206 | logger.Infow(ctx, "starting-grpc-nbi-service", log.Fields{"service": serviceName}) |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 207 | |
| 208 | server.AddService(func(gs *grpc.Server) { voltha.RegisterVolthaServiceServer(gs, handler) }) |
khenaidoo | d948f77 | 2021-08-11 17:49:24 -0400 | [diff] [blame] | 209 | logger.Infow(ctx, "grpc-nbi-service-added-and-started", log.Fields{"service": serviceName}) |
khenaidoo | b920354 | 2018-09-17 22:56:37 -0400 | [diff] [blame] | 210 | |
Kent Hagerman | 2f0d055 | 2020-04-23 17:28:52 -0400 | [diff] [blame] | 211 | // Note that there is a small window here in which the core could return its status as ready, |
| 212 | // when it really isn't. This is unlikely to cause issues, as the delay is incredibly short. |
| 213 | server.Start(ctx) |
khenaidoo | b920354 | 2018-09-17 22:56:37 -0400 | [diff] [blame] | 214 | } |
Salman Siddiqui | 1cf9504 | 2020-11-19 00:42:56 +0530 | [diff] [blame] | 215 | |
| 216 | func addGRPCExtensionService(ctx context.Context, server *grpcserver.GrpcServer, handler extension.ExtensionServer) { |
| 217 | logger.Info(ctx, "extension-grpc-server-created") |
| 218 | |
| 219 | server.AddService(func(server *grpc.Server) { |
| 220 | extension.RegisterExtensionServer(server, handler) |
| 221 | }) |
Salman Siddiqui | 1cf9504 | 2020-11-19 00:42:56 +0530 | [diff] [blame] | 222 | } |