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