blob: b3bbc71427b4f5e285af557009819a9981ce674c [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 */
16package core
17
18import (
19 "context"
khenaidoo54e0ddf2019-02-27 16:21:33 -050020 "errors"
21 "fmt"
khenaidoob9203542018-09-17 22:56:37 -040022 grpcserver "github.com/opencord/voltha-go/common/grpc"
23 "github.com/opencord/voltha-go/common/log"
Richard Jankowskie4d77662018-10-17 13:53:21 -040024 "github.com/opencord/voltha-go/db/kvstore"
khenaidoob9203542018-09-17 22:56:37 -040025 "github.com/opencord/voltha-go/db/model"
26 "github.com/opencord/voltha-go/kafka"
27 "github.com/opencord/voltha-go/protos/voltha"
28 "github.com/opencord/voltha-go/rw_core/config"
29 "google.golang.org/grpc"
khenaidoo6417b6c2019-03-01 18:18:01 -050030 "google.golang.org/grpc/codes"
31 "google.golang.org/grpc/status"
32 "sync"
khenaidoob9203542018-09-17 22:56:37 -040033)
34
35type Core struct {
36 instanceId string
37 deviceMgr *DeviceManager
38 logicalDeviceMgr *LogicalDeviceManager
39 grpcServer *grpcserver.GrpcServer
Richard Jankowskidbab94a2018-12-06 16:20:25 -050040 grpcNBIAPIHandler *APIHandler
khenaidoo21d51152019-02-01 13:48:37 -050041 adapterMgr *AdapterManager
khenaidoob9203542018-09-17 22:56:37 -040042 config *config.RWCoreFlags
khenaidoo43c82122018-11-22 18:38:28 -050043 kmp *kafka.InterContainerProxy
khenaidoo92e62c52018-10-03 14:02:54 -040044 clusterDataRoot model.Root
45 localDataRoot model.Root
khenaidoob9203542018-09-17 22:56:37 -040046 clusterDataProxy *model.Proxy
47 localDataProxy *model.Proxy
48 exitChannel chan int
Richard Jankowskie4d77662018-10-17 13:53:21 -040049 kvClient kvstore.Client
khenaidoo43c82122018-11-22 18:38:28 -050050 kafkaClient kafka.Client
khenaidoo6417b6c2019-03-01 18:18:01 -050051 coreMembership *voltha.Membership
52 membershipLock *sync.RWMutex
khenaidoob9203542018-09-17 22:56:37 -040053}
54
55func init() {
56 log.AddPackage(log.JSON, log.WarnLevel, nil)
57}
58
khenaidoo43c82122018-11-22 18:38:28 -050059func NewCore(id string, cf *config.RWCoreFlags, kvClient kvstore.Client, kafkaClient kafka.Client) *Core {
khenaidoob9203542018-09-17 22:56:37 -040060 var core Core
61 core.instanceId = id
62 core.exitChannel = make(chan int, 1)
63 core.config = cf
Richard Jankowskie4d77662018-10-17 13:53:21 -040064 core.kvClient = kvClient
khenaidoo43c82122018-11-22 18:38:28 -050065 core.kafkaClient = kafkaClient
Richard Jankowskie4d77662018-10-17 13:53:21 -040066
67 // Setup the KV store
68 // Do not call NewBackend constructor; it creates its own KV client
khenaidoo91ecfd62018-11-04 17:13:42 -050069 // Commented the backend for now until the issue between the model and the KV store
70 // is resolved.
khenaidoo7ccedd52018-12-14 16:48:54 -050071 backend := model.Backend{
72 Client: kvClient,
73 StoreType: cf.KVStoreType,
74 Host: cf.KVStoreHost,
75 Port: cf.KVStorePort,
76 Timeout: cf.KVStoreTimeout,
khenaidoo9cdc1a62019-01-24 21:57:40 -050077 PathPrefix: cf.KVStoreDataPrefix}
khenaidoo7ccedd52018-12-14 16:48:54 -050078 core.clusterDataRoot = model.NewRoot(&voltha.Voltha{}, &backend)
79 core.localDataRoot = model.NewRoot(&voltha.CoreInstance{}, &backend)
khenaidoo43c82122018-11-22 18:38:28 -050080 core.clusterDataProxy = core.clusterDataRoot.CreateProxy("/", false)
81 core.localDataProxy = core.localDataRoot.CreateProxy("/", false)
khenaidoo6417b6c2019-03-01 18:18:01 -050082 core.membershipLock = &sync.RWMutex{}
khenaidoob9203542018-09-17 22:56:37 -040083 return &core
84}
85
86func (core *Core) Start(ctx context.Context) {
khenaidoo19374072018-12-11 11:05:15 -050087 log.Info("starting-adaptercore", log.Fields{"coreId": core.instanceId})
khenaidoo9cdc1a62019-01-24 21:57:40 -050088 if err := core.startKafkaMessagingProxy(ctx); err != nil {
89 log.Fatal("Failure-starting-kafkaMessagingProxy")
90 }
khenaidoob9203542018-09-17 22:56:37 -040091 log.Info("values", log.Fields{"kmp": core.kmp})
khenaidoo21d51152019-02-01 13:48:37 -050092 core.adapterMgr = newAdapterManager(core.clusterDataProxy, core.instanceId)
93 core.deviceMgr = newDeviceManager(core.kmp, core.clusterDataProxy, core.adapterMgr, core.instanceId)
khenaidoo4d4802d2018-10-04 21:59:49 -040094 core.logicalDeviceMgr = newLogicalDeviceManager(core.deviceMgr, core.kmp, core.clusterDataProxy)
khenaidoo54e0ddf2019-02-27 16:21:33 -050095
96 if err := core.registerAdapterRequestHandlers(ctx, core.instanceId, core.deviceMgr, core.logicalDeviceMgr, core.adapterMgr, core.clusterDataProxy, core.localDataProxy); err != nil {
khenaidoo9cdc1a62019-01-24 21:57:40 -050097 log.Fatal("Failure-registering-adapterRequestHandler")
98 }
khenaidoob9203542018-09-17 22:56:37 -040099 go core.startDeviceManager(ctx)
100 go core.startLogicalDeviceManager(ctx)
101 go core.startGRPCService(ctx)
khenaidoo21d51152019-02-01 13:48:37 -0500102 go core.startAdapterManager(ctx)
khenaidoob9203542018-09-17 22:56:37 -0400103
khenaidoo19374072018-12-11 11:05:15 -0500104 log.Info("adaptercore-started")
khenaidoob9203542018-09-17 22:56:37 -0400105}
106
107func (core *Core) Stop(ctx context.Context) {
khenaidoo19374072018-12-11 11:05:15 -0500108 log.Info("stopping-adaptercore")
khenaidoob9203542018-09-17 22:56:37 -0400109 core.exitChannel <- 1
khenaidoo43c82122018-11-22 18:38:28 -0500110 // Stop all the started services
111 core.grpcServer.Stop()
112 core.logicalDeviceMgr.stop(ctx)
113 core.deviceMgr.stop(ctx)
114 core.kmp.Stop()
khenaidoo19374072018-12-11 11:05:15 -0500115 log.Info("adaptercore-stopped")
khenaidoob9203542018-09-17 22:56:37 -0400116}
117
khenaidoo92e62c52018-10-03 14:02:54 -0400118//startGRPCService creates the grpc service handlers, registers it to the grpc server
khenaidoob9203542018-09-17 22:56:37 -0400119// and starts the server
120func (core *Core) startGRPCService(ctx context.Context) {
121 // create an insecure gserver server
122 core.grpcServer = grpcserver.NewGrpcServer(core.config.GrpcHost, core.config.GrpcPort, nil, false)
123 log.Info("grpc-server-created")
124
khenaidoo54e0ddf2019-02-27 16:21:33 -0500125 //core.grpcNBIAPIHandler = NewAPIHandler(core.deviceMgr, core.logicalDeviceMgr, core.adapterMgr, core.config.InCompetingMode, core.config.LongRunningRequestTimeout, core.config.DefaultRequestTimeout)
126 core.grpcNBIAPIHandler = NewAPIHandler(core)
Richard Jankowskidbab94a2018-12-06 16:20:25 -0500127 core.logicalDeviceMgr.setGrpcNbiHandler(core.grpcNBIAPIHandler)
khenaidoob9203542018-09-17 22:56:37 -0400128 // Create a function to register the core GRPC service with the GRPC server
129 f := func(gs *grpc.Server) {
130 voltha.RegisterVolthaServiceServer(
131 gs,
Richard Jankowskidbab94a2018-12-06 16:20:25 -0500132 core.grpcNBIAPIHandler,
khenaidoob9203542018-09-17 22:56:37 -0400133 )
134 }
135
136 core.grpcServer.AddService(f)
137 log.Info("grpc-service-added")
138
139 // Start the server
140 core.grpcServer.Start(context.Background())
141 log.Info("grpc-server-started")
142}
143
144func (core *Core) startKafkaMessagingProxy(ctx context.Context) error {
145 log.Infow("starting-kafka-messaging-proxy", log.Fields{"host": core.config.KafkaAdapterHost,
146 "port": core.config.KafkaAdapterPort, "topic": core.config.CoreTopic})
147 var err error
khenaidoo43c82122018-11-22 18:38:28 -0500148 if core.kmp, err = kafka.NewInterContainerProxy(
149 kafka.InterContainerHost(core.config.KafkaAdapterHost),
150 kafka.InterContainerPort(core.config.KafkaAdapterPort),
151 kafka.MsgClient(core.kafkaClient),
khenaidoo79232702018-12-04 11:00:41 -0500152 kafka.DefaultTopic(&kafka.Topic{Name: core.config.CoreTopic}),
153 kafka.DeviceDiscoveryTopic(&kafka.Topic{Name: core.config.AffinityRouterTopic})); err != nil {
khenaidoob9203542018-09-17 22:56:37 -0400154 log.Errorw("fail-to-create-kafka-proxy", log.Fields{"error": err})
155 return err
156 }
157
158 if err = core.kmp.Start(); err != nil {
159 log.Fatalw("error-starting-messaging-proxy", log.Fields{"error": err})
160 return err
161 }
162
163 log.Info("kafka-messaging-proxy-created")
164 return nil
165}
166
khenaidoo54e0ddf2019-02-27 16:21:33 -0500167//func (core *Core) registerAdapterRequestHandler(ctx context.Context, coreInstanceId string, dMgr *DeviceManager,
168// ldMgr *LogicalDeviceManager, aMgr *AdapterManager, cdProxy *model.Proxy, ldProxy *model.Proxy,
169// ) error {
170// requestProxy := NewAdapterRequestHandlerProxy(coreInstanceId, dMgr, ldMgr, aMgr, cdProxy, ldProxy,
171// core.config.InCompetingMode, core.config.LongRunningRequestTimeout, core.config.DefaultRequestTimeout)
172// core.kmp.SubscribeWithRequestHandlerInterface(kafka.Topic{Name: core.config.CoreTopic}, requestProxy)
173//
174// log.Info("request-handlers")
175// return nil
176//}
177
178func (core *Core) registerAdapterRequestHandlers(ctx context.Context, coreInstanceId string, dMgr *DeviceManager,
khenaidoo297cd252019-02-07 22:10:23 -0500179 ldMgr *LogicalDeviceManager, aMgr *AdapterManager, cdProxy *model.Proxy, ldProxy *model.Proxy,
khenaidoo54e0ddf2019-02-27 16:21:33 -0500180) error {
khenaidoo297cd252019-02-07 22:10:23 -0500181 requestProxy := NewAdapterRequestHandlerProxy(coreInstanceId, dMgr, ldMgr, aMgr, cdProxy, ldProxy,
182 core.config.InCompetingMode, core.config.LongRunningRequestTimeout, core.config.DefaultRequestTimeout)
khenaidoob9203542018-09-17 22:56:37 -0400183
khenaidoo54e0ddf2019-02-27 16:21:33 -0500184 // Register the broadcast topic to handle any core-bound broadcast requests
185 if err := core.kmp.SubscribeWithRequestHandlerInterface(kafka.Topic{Name: core.config.CoreTopic}, requestProxy); err != nil {
186 log.Fatalw("Failed-registering-broadcast-handler", log.Fields{"topic": core.config.CoreTopic})
187 return err
188 }
189
190 log.Info("request-handler-registered")
khenaidoob9203542018-09-17 22:56:37 -0400191 return nil
192}
193
khenaidoo54e0ddf2019-02-27 16:21:33 -0500194
khenaidoo6417b6c2019-03-01 18:18:01 -0500195func (core *Core) isMembershipRegistrationComplete() bool {
196 core.membershipLock.RLock()
197 defer core.membershipLock.RUnlock()
198 return core.coreMembership != nil
199}
200
201func (core *Core) updateCoreMembership(ctx context.Context, membership *voltha.Membership) error {
202 core.membershipLock.Lock()
203 defer core.membershipLock.Unlock()
khenaidoo54e0ddf2019-02-27 16:21:33 -0500204 // Sanity check - can only register once
khenaidoo6417b6c2019-03-01 18:18:01 -0500205 if core.coreMembership != nil {
206 log.Warnw("core-reregistration-not-allowed", log.Fields{"membership": membership})
207 return status.Errorf(codes.AlreadyExists, "%s", membership.GroupName)
khenaidoo54e0ddf2019-02-27 16:21:33 -0500208 }
209
khenaidoo6417b6c2019-03-01 18:18:01 -0500210 if membership == nil || membership.GroupName == "" {
211 return errors.New("invalid-membership-info")
212 }
213
214 core.coreMembership = membership
215
216
217 // Use the group name to register a specific kafka topic for this container
218 go func(groupName string) {
219 // Register the core-pair topic to handle core-bound requests destined to the core pair
220 pTopic := kafka.Topic{Name: fmt.Sprintf("%s_%s", core.config.CoreTopic, groupName)}
khenaidoo54e0ddf2019-02-27 16:21:33 -0500221 // TODO: Set a retry here to ensure this subscription happens
222 if err := core.kmp.SubscribeWithDefaultRequestHandler(pTopic, kafka.OffsetNewest); err != nil {
223 log.Fatalw("Failed-registering-pair-handler", log.Fields{"topic": core.config.CoreTopic})
224 }
khenaidoo6417b6c2019-03-01 18:18:01 -0500225 // Update the CoreTopic to be used by the adapter proxy
khenaidoo54e0ddf2019-02-27 16:21:33 -0500226 core.deviceMgr.adapterProxy.updateCoreTopic(&pTopic)
khenaidoo6417b6c2019-03-01 18:18:01 -0500227 }(membership.GroupName)
khenaidoo54e0ddf2019-02-27 16:21:33 -0500228
khenaidoo6417b6c2019-03-01 18:18:01 -0500229 log.Info("core-membership-registered")
khenaidoo54e0ddf2019-02-27 16:21:33 -0500230 return nil
231}
232
khenaidoo6417b6c2019-03-01 18:18:01 -0500233func (core *Core) getCoreMembership(ctx context.Context) *voltha.Membership {
234 core.membershipLock.RLock()
235 defer core.membershipLock.RUnlock()
236 return core.coreMembership
237}
238
khenaidoo54e0ddf2019-02-27 16:21:33 -0500239
khenaidoob9203542018-09-17 22:56:37 -0400240func (core *Core) startDeviceManager(ctx context.Context) {
241 // TODO: Interaction between the logicaldevicemanager and devicemanager should mostly occur via
242 // callbacks. For now, until the model is ready, devicemanager will keep a reference to the
243 // logicaldevicemanager to initiate the creation of logical devices
khenaidoo21d51152019-02-01 13:48:37 -0500244 log.Info("DeviceManager-Starting...")
khenaidoo4d4802d2018-10-04 21:59:49 -0400245 core.deviceMgr.start(ctx, core.logicalDeviceMgr)
khenaidoo21d51152019-02-01 13:48:37 -0500246 log.Info("DeviceManager-Started")
khenaidoob9203542018-09-17 22:56:37 -0400247}
248
249func (core *Core) startLogicalDeviceManager(ctx context.Context) {
khenaidoo21d51152019-02-01 13:48:37 -0500250 log.Info("Logical-DeviceManager-Starting...")
khenaidoo4d4802d2018-10-04 21:59:49 -0400251 core.logicalDeviceMgr.start(ctx)
khenaidoo21d51152019-02-01 13:48:37 -0500252 log.Info("Logical-DeviceManager-Started")
khenaidoob9203542018-09-17 22:56:37 -0400253}
khenaidoo21d51152019-02-01 13:48:37 -0500254
255func (core *Core) startAdapterManager(ctx context.Context) {
256 log.Info("Adapter-Manager-Starting...")
257 core.adapterMgr.start(ctx)
258 log.Info("Adapter-Manager-Started")
259}