blob: 3fa68eff9dc99febd9330a1095628962c13caecf [file] [log] [blame]
* Copyright 2018-present Open Networking Foundation
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* See the License for the specific language governing permissions and
* limitations under the License.
package core
import (
conf ""
grpcserver ""
// Core represent read,write core attributes
type Core struct {
Shutdown context.CancelFunc
Stopped chan struct{}
KafkaClient kafka.Client
const (
clusterMessagingService = "cluster-message-service"
grpcNBIService = "grpc-nbi-service"
grpcSBIService = "grpc-sbi-service"
adapterService = "adapter-service"
kvService = "kv-service"
deviceService = "device-service"
logicalDeviceService = "logical-device-service"
// NewCore creates instance of rw core
func NewCore(ctx context.Context, id string, cf *config.RWCoreFlags) (*Core, context.Context) {
// If the context has a probe then fetch it and register our services
if p := probe.GetProbeFromContext(ctx); p != nil {
// create kafka client for events
KafkaClient := kafka.NewSaramaClient(
// new threads will be given a new cancelable context, so that they can be aborted later when Stop() is called
shutdownCtx, cancelCtx := context.WithCancel(ctx)
rwCore := &Core{Shutdown: cancelCtx, Stopped: make(chan struct{}), KafkaClient: KafkaClient}
return rwCore, shutdownCtx
func (core *Core) Start(ctx context.Context, id string, cf *config.RWCoreFlags) {
logger.Info(ctx, "starting-core-services", log.Fields{"coreId": id})
// deferred functions are used to run cleanup
// failing partway will stop anything that's been started
defer close(core.Stopped)
defer core.Shutdown()
logger.Info(ctx, "starting-rw-core-components")
// setup kv client
logger.Debugw(ctx, "create-kv-client", log.Fields{"kvstore": cf.KVStoreType})
kvClient, err := newKVClient(ctx, cf.KVStoreType, cf.KVStoreAddress, cf.KVStoreTimeout)
if err != nil {
logger.Fatal(ctx, err)
defer stopKVClient(log.WithSpanFromContext(context.Background(), ctx), kvClient)
// sync logging config with kv store
cm := conf.NewConfigManager(ctx, kvClient, cf.KVStoreType, cf.KVStoreAddress, cf.KVStoreTimeout)
go conf.StartLogLevelConfigProcessing(cm, ctx)
go conf.StartLogFeaturesConfigProcessing(cm, ctx)
backend := cm.Backend
backend.LivenessChannelInterval = cf.LiveProbeInterval / 2
// wait until connection to KV Store is up
if err := waitUntilKVStoreReachableOrMaxTries(ctx, kvClient, cf.MaxConnectionRetries, cf.ConnectionRetryInterval, kvService); err != nil {
logger.Fatal(ctx, "unable-to-connect-to-kv-store")
go monitorKVStoreLiveness(ctx, backend, kvService, cf.LiveProbeInterval, cf.NotLiveProbeInterval)
// Start kafka communications and artefacts
if err := kafka.StartAndWaitUntilKafkaConnectionIsUp(ctx, core.KafkaClient, cf.ConnectionRetryInterval, clusterMessagingService); err != nil {
logger.Fatal(ctx, "unable-to-connect-to-kafka")
defer core.KafkaClient.Stop(ctx)
// Create the event proxy to post events to KAFKA
eventProxy := events.NewEventProxy(events.MsgClient(core.KafkaClient), events.MsgTopic(kafka.Topic{Name: cf.EventTopic}))
go func() {
if err := eventProxy.Start(); err != nil {
logger.Fatalw(ctx, "event-proxy-cannot-start", log.Fields{"error": err})
defer eventProxy.Stop()
// Start the kafka monitoring routine
go kafka.MonitorKafkaReadiness(ctx, core.KafkaClient, cf.LiveProbeInterval, cf.NotLiveProbeInterval, clusterMessagingService)
// create kv path
dbPath := model.NewDBPath(backend)
// load adapters & device types while other things are starting
adapterMgr := adapter.NewAdapterManager(dbPath, id, backend, cf.LiveProbeInterval)
adapterMgr.Start(ctx, adapterService)
// create the core of the system, the device managers
deviceMgr, logicalDeviceMgr := device.NewManagers(dbPath, adapterMgr, cf, id, eventProxy)
// Start the device manager to load the devices. Wait until it is completed to prevent multiple loading happening
// triggered by logicalDeviceMgr.Start(Ctx)
err = deviceMgr.Start(ctx, deviceService)
if err != nil {
logger.Fatalw(ctx, "failure-starting-device-manager", log.Fields{"error": err})
// Start the logical device manager to load the logical devices.
logicalDeviceMgr.Start(ctx, logicalDeviceService)
// Create and start the SBI gRPC service
grpcSBIServer := grpcserver.NewGrpcServer(cf.GrpcSBIAddress, nil, false, probe.GetProbeFromContext(ctx))
go startGrpcSbiService(ctx, grpcSBIServer, grpcSBIService, api.NewAPIHandler(deviceMgr, nil, adapterMgr))
defer grpcSBIServer.Stop()
// In the case of a restart, let's wait until all the registered adapters are connected to the Core
// before starting the grpc server that handles NBI requests.
err = adapterMgr.WaitUntilConnectionsToAdaptersAreUp(ctx, cf.ConnectionRetryInterval)
if err != nil {
logger.Fatalw(ctx, "failure-connecting-to-adapters", log.Fields{"error": err})
// Create the NBI gRPC server
grpcNBIServer := grpcserver.NewGrpcServer(cf.GrpcNBIAddress, nil, false, probe.GetProbeFromContext(ctx))
//Register the 'Extension' service on this gRPC server
addGRPCExtensionService(ctx, grpcNBIServer, device.GetNewExtensionManager(deviceMgr))
go startGrpcNbiService(ctx, grpcNBIServer, grpcNBIService, api.NewAPIHandler(deviceMgr, logicalDeviceMgr, adapterMgr))
defer grpcNBIServer.Stop()
// wait for core to be stopped, via Stop() or context cancellation, before running deferred functions
// Stop brings down core services
func (core *Core) Stop() {
// startGrpcSbiService creates the grpc core service handlers, registers it to the grpc server and starts the server
func startGrpcSbiService(ctx context.Context, server *grpcserver.GrpcServer, serviceName string, handler core.CoreServiceServer) {
logger.Infow(ctx, "starting-grpc-sbi-service", log.Fields{"service": serviceName})
server.AddService(func(server *grpc.Server) { core.RegisterCoreServiceServer(server, handler) })
logger.Infow(ctx, "grpc-sbi-service-added", log.Fields{"service": serviceName})
probe.UpdateStatusFromContext(ctx, serviceName, probe.ServiceStatusRunning)
logger.Infow(ctx, "grpc-sbi-server-started", log.Fields{"service": serviceName})
probe.UpdateStatusFromContext(ctx, serviceName, probe.ServiceStatusStopped)
// startGrpcNbiService creates the grpc NBI service handlers, registers it to the grpc server and starts the server
func startGrpcNbiService(ctx context.Context, server *grpcserver.GrpcServer, serviceName string, handler voltha.VolthaServiceServer) {
logger.Infow(ctx, "starting-grpc-nbi-service", log.Fields{"service": serviceName})
server.AddService(func(gs *grpc.Server) { voltha.RegisterVolthaServiceServer(gs, handler) })
logger.Infow(ctx, "grpc-nbi-service-added-and-started", log.Fields{"service": serviceName})
// Note that there is a small window here in which the core could return its status as ready,
// when it really isn't. This is unlikely to cause issues, as the delay is incredibly short.
func addGRPCExtensionService(ctx context.Context, server *grpcserver.GrpcServer, handler extension.ExtensionServer) {
logger.Info(ctx, "extension-grpc-server-created")
server.AddService(func(server *grpc.Server) {
extension.RegisterExtensionServer(server, handler)