blob: 9f6adcd618f15f47e2edadd58d3c971ad55a8187 [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
Richard Jankowski199fd862019-03-18 14:49:51 -040053 deviceOwnership *DeviceOwnership
khenaidoob9203542018-09-17 22:56:37 -040054}
55
56func init() {
57 log.AddPackage(log.JSON, log.WarnLevel, nil)
58}
59
khenaidoo43c82122018-11-22 18:38:28 -050060func NewCore(id string, cf *config.RWCoreFlags, kvClient kvstore.Client, kafkaClient kafka.Client) *Core {
khenaidoob9203542018-09-17 22:56:37 -040061 var core Core
62 core.instanceId = id
63 core.exitChannel = make(chan int, 1)
64 core.config = cf
Richard Jankowskie4d77662018-10-17 13:53:21 -040065 core.kvClient = kvClient
khenaidoo43c82122018-11-22 18:38:28 -050066 core.kafkaClient = kafkaClient
Richard Jankowskie4d77662018-10-17 13:53:21 -040067
68 // Setup the KV store
69 // Do not call NewBackend constructor; it creates its own KV client
khenaidoo91ecfd62018-11-04 17:13:42 -050070 // Commented the backend for now until the issue between the model and the KV store
71 // is resolved.
khenaidoo7ccedd52018-12-14 16:48:54 -050072 backend := model.Backend{
73 Client: kvClient,
74 StoreType: cf.KVStoreType,
75 Host: cf.KVStoreHost,
76 Port: cf.KVStorePort,
77 Timeout: cf.KVStoreTimeout,
khenaidoo9cdc1a62019-01-24 21:57:40 -050078 PathPrefix: cf.KVStoreDataPrefix}
khenaidoo7ccedd52018-12-14 16:48:54 -050079 core.clusterDataRoot = model.NewRoot(&voltha.Voltha{}, &backend)
80 core.localDataRoot = model.NewRoot(&voltha.CoreInstance{}, &backend)
khenaidoo43c82122018-11-22 18:38:28 -050081 core.clusterDataProxy = core.clusterDataRoot.CreateProxy("/", false)
82 core.localDataProxy = core.localDataRoot.CreateProxy("/", false)
khenaidoo6417b6c2019-03-01 18:18:01 -050083 core.membershipLock = &sync.RWMutex{}
khenaidoob9203542018-09-17 22:56:37 -040084 return &core
85}
86
87func (core *Core) Start(ctx context.Context) {
khenaidoo19374072018-12-11 11:05:15 -050088 log.Info("starting-adaptercore", log.Fields{"coreId": core.instanceId})
khenaidoo9cdc1a62019-01-24 21:57:40 -050089 if err := core.startKafkaMessagingProxy(ctx); err != nil {
90 log.Fatal("Failure-starting-kafkaMessagingProxy")
91 }
khenaidoob9203542018-09-17 22:56:37 -040092 log.Info("values", log.Fields{"kmp": core.kmp})
khenaidoo21d51152019-02-01 13:48:37 -050093 core.adapterMgr = newAdapterManager(core.clusterDataProxy, core.instanceId)
Richard Jankowski199fd862019-03-18 14:49:51 -040094 core.deviceMgr = newDeviceManager(core)
95 core.logicalDeviceMgr = newLogicalDeviceManager(core, core.deviceMgr, core.kmp, core.clusterDataProxy)
khenaidoo54e0ddf2019-02-27 16:21:33 -050096
97 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 -050098 log.Fatal("Failure-registering-adapterRequestHandler")
99 }
khenaidoob9203542018-09-17 22:56:37 -0400100 go core.startDeviceManager(ctx)
101 go core.startLogicalDeviceManager(ctx)
102 go core.startGRPCService(ctx)
khenaidoo21d51152019-02-01 13:48:37 -0500103 go core.startAdapterManager(ctx)
khenaidoob9203542018-09-17 22:56:37 -0400104
khenaidoo1ce37ad2019-03-24 22:07:24 -0400105 // Setup device ownership context
106 core.deviceOwnership = NewDeviceOwnership(core.instanceId, core.kvClient, core.deviceMgr, core.logicalDeviceMgr,
107 "service/voltha/owns_device", 10)
108
khenaidoo19374072018-12-11 11:05:15 -0500109 log.Info("adaptercore-started")
khenaidoob9203542018-09-17 22:56:37 -0400110}
111
112func (core *Core) Stop(ctx context.Context) {
khenaidoo19374072018-12-11 11:05:15 -0500113 log.Info("stopping-adaptercore")
khenaidoob9203542018-09-17 22:56:37 -0400114 core.exitChannel <- 1
khenaidoo43c82122018-11-22 18:38:28 -0500115 // Stop all the started services
116 core.grpcServer.Stop()
117 core.logicalDeviceMgr.stop(ctx)
118 core.deviceMgr.stop(ctx)
119 core.kmp.Stop()
khenaidoo19374072018-12-11 11:05:15 -0500120 log.Info("adaptercore-stopped")
khenaidoob9203542018-09-17 22:56:37 -0400121}
122
khenaidoo92e62c52018-10-03 14:02:54 -0400123//startGRPCService creates the grpc service handlers, registers it to the grpc server
khenaidoob9203542018-09-17 22:56:37 -0400124// and starts the server
125func (core *Core) startGRPCService(ctx context.Context) {
126 // create an insecure gserver server
127 core.grpcServer = grpcserver.NewGrpcServer(core.config.GrpcHost, core.config.GrpcPort, nil, false)
128 log.Info("grpc-server-created")
129
khenaidoo54e0ddf2019-02-27 16:21:33 -0500130 //core.grpcNBIAPIHandler = NewAPIHandler(core.deviceMgr, core.logicalDeviceMgr, core.adapterMgr, core.config.InCompetingMode, core.config.LongRunningRequestTimeout, core.config.DefaultRequestTimeout)
131 core.grpcNBIAPIHandler = NewAPIHandler(core)
Richard Jankowski46464e92019-03-05 11:53:55 -0500132 log.Infow("grpc-handler", log.Fields{"core_binding_key": core.config.CoreBindingKey})
Richard Jankowskidbab94a2018-12-06 16:20:25 -0500133 core.logicalDeviceMgr.setGrpcNbiHandler(core.grpcNBIAPIHandler)
khenaidoob9203542018-09-17 22:56:37 -0400134 // Create a function to register the core GRPC service with the GRPC server
135 f := func(gs *grpc.Server) {
136 voltha.RegisterVolthaServiceServer(
137 gs,
Richard Jankowskidbab94a2018-12-06 16:20:25 -0500138 core.grpcNBIAPIHandler,
khenaidoob9203542018-09-17 22:56:37 -0400139 )
140 }
141
142 core.grpcServer.AddService(f)
143 log.Info("grpc-service-added")
144
145 // Start the server
146 core.grpcServer.Start(context.Background())
147 log.Info("grpc-server-started")
148}
149
150func (core *Core) startKafkaMessagingProxy(ctx context.Context) error {
151 log.Infow("starting-kafka-messaging-proxy", log.Fields{"host": core.config.KafkaAdapterHost,
152 "port": core.config.KafkaAdapterPort, "topic": core.config.CoreTopic})
153 var err error
khenaidoo43c82122018-11-22 18:38:28 -0500154 if core.kmp, err = kafka.NewInterContainerProxy(
155 kafka.InterContainerHost(core.config.KafkaAdapterHost),
156 kafka.InterContainerPort(core.config.KafkaAdapterPort),
157 kafka.MsgClient(core.kafkaClient),
khenaidoo79232702018-12-04 11:00:41 -0500158 kafka.DefaultTopic(&kafka.Topic{Name: core.config.CoreTopic}),
159 kafka.DeviceDiscoveryTopic(&kafka.Topic{Name: core.config.AffinityRouterTopic})); err != nil {
khenaidoob9203542018-09-17 22:56:37 -0400160 log.Errorw("fail-to-create-kafka-proxy", log.Fields{"error": err})
161 return err
162 }
163
164 if err = core.kmp.Start(); err != nil {
165 log.Fatalw("error-starting-messaging-proxy", log.Fields{"error": err})
166 return err
167 }
168
169 log.Info("kafka-messaging-proxy-created")
170 return nil
171}
172
khenaidoo54e0ddf2019-02-27 16:21:33 -0500173//func (core *Core) registerAdapterRequestHandler(ctx context.Context, coreInstanceId string, dMgr *DeviceManager,
174// ldMgr *LogicalDeviceManager, aMgr *AdapterManager, cdProxy *model.Proxy, ldProxy *model.Proxy,
175// ) error {
176// requestProxy := NewAdapterRequestHandlerProxy(coreInstanceId, dMgr, ldMgr, aMgr, cdProxy, ldProxy,
177// core.config.InCompetingMode, core.config.LongRunningRequestTimeout, core.config.DefaultRequestTimeout)
178// core.kmp.SubscribeWithRequestHandlerInterface(kafka.Topic{Name: core.config.CoreTopic}, requestProxy)
179//
180// log.Info("request-handlers")
181// return nil
182//}
183
184func (core *Core) registerAdapterRequestHandlers(ctx context.Context, coreInstanceId string, dMgr *DeviceManager,
khenaidoo297cd252019-02-07 22:10:23 -0500185 ldMgr *LogicalDeviceManager, aMgr *AdapterManager, cdProxy *model.Proxy, ldProxy *model.Proxy,
khenaidoo54e0ddf2019-02-27 16:21:33 -0500186) error {
Richard Jankowski199fd862019-03-18 14:49:51 -0400187 requestProxy := NewAdapterRequestHandlerProxy(core, coreInstanceId, dMgr, ldMgr, aMgr, cdProxy, ldProxy,
khenaidoo297cd252019-02-07 22:10:23 -0500188 core.config.InCompetingMode, core.config.LongRunningRequestTimeout, core.config.DefaultRequestTimeout)
khenaidoob9203542018-09-17 22:56:37 -0400189
khenaidoo54e0ddf2019-02-27 16:21:33 -0500190 // Register the broadcast topic to handle any core-bound broadcast requests
191 if err := core.kmp.SubscribeWithRequestHandlerInterface(kafka.Topic{Name: core.config.CoreTopic}, requestProxy); err != nil {
192 log.Fatalw("Failed-registering-broadcast-handler", log.Fields{"topic": core.config.CoreTopic})
193 return err
194 }
195
196 log.Info("request-handler-registered")
khenaidoob9203542018-09-17 22:56:37 -0400197 return nil
198}
199
khenaidoo54e0ddf2019-02-27 16:21:33 -0500200
khenaidoo6417b6c2019-03-01 18:18:01 -0500201func (core *Core) isMembershipRegistrationComplete() bool {
202 core.membershipLock.RLock()
203 defer core.membershipLock.RUnlock()
204 return core.coreMembership != nil
205}
206
207func (core *Core) updateCoreMembership(ctx context.Context, membership *voltha.Membership) error {
208 core.membershipLock.Lock()
209 defer core.membershipLock.Unlock()
khenaidoo54e0ddf2019-02-27 16:21:33 -0500210 // Sanity check - can only register once
khenaidoo6417b6c2019-03-01 18:18:01 -0500211 if core.coreMembership != nil {
212 log.Warnw("core-reregistration-not-allowed", log.Fields{"membership": membership})
213 return status.Errorf(codes.AlreadyExists, "%s", membership.GroupName)
khenaidoo54e0ddf2019-02-27 16:21:33 -0500214 }
215
khenaidoo6417b6c2019-03-01 18:18:01 -0500216 if membership == nil || membership.GroupName == "" {
217 return errors.New("invalid-membership-info")
218 }
219
220 core.coreMembership = membership
221
222
223 // Use the group name to register a specific kafka topic for this container
224 go func(groupName string) {
225 // Register the core-pair topic to handle core-bound requests destined to the core pair
226 pTopic := kafka.Topic{Name: fmt.Sprintf("%s_%s", core.config.CoreTopic, groupName)}
khenaidoo54e0ddf2019-02-27 16:21:33 -0500227 // TODO: Set a retry here to ensure this subscription happens
228 if err := core.kmp.SubscribeWithDefaultRequestHandler(pTopic, kafka.OffsetNewest); err != nil {
229 log.Fatalw("Failed-registering-pair-handler", log.Fields{"topic": core.config.CoreTopic})
230 }
khenaidoo6417b6c2019-03-01 18:18:01 -0500231 // Update the CoreTopic to be used by the adapter proxy
khenaidoo54e0ddf2019-02-27 16:21:33 -0500232 core.deviceMgr.adapterProxy.updateCoreTopic(&pTopic)
khenaidoo6417b6c2019-03-01 18:18:01 -0500233 }(membership.GroupName)
khenaidoo54e0ddf2019-02-27 16:21:33 -0500234
khenaidoo6417b6c2019-03-01 18:18:01 -0500235 log.Info("core-membership-registered")
khenaidoo54e0ddf2019-02-27 16:21:33 -0500236 return nil
237}
238
khenaidoo6417b6c2019-03-01 18:18:01 -0500239func (core *Core) getCoreMembership(ctx context.Context) *voltha.Membership {
240 core.membershipLock.RLock()
241 defer core.membershipLock.RUnlock()
242 return core.coreMembership
243}
244
khenaidoo54e0ddf2019-02-27 16:21:33 -0500245
khenaidoob9203542018-09-17 22:56:37 -0400246func (core *Core) startDeviceManager(ctx context.Context) {
247 // TODO: Interaction between the logicaldevicemanager and devicemanager should mostly occur via
248 // callbacks. For now, until the model is ready, devicemanager will keep a reference to the
249 // logicaldevicemanager to initiate the creation of logical devices
khenaidoo21d51152019-02-01 13:48:37 -0500250 log.Info("DeviceManager-Starting...")
khenaidoo4d4802d2018-10-04 21:59:49 -0400251 core.deviceMgr.start(ctx, core.logicalDeviceMgr)
khenaidoo21d51152019-02-01 13:48:37 -0500252 log.Info("DeviceManager-Started")
khenaidoob9203542018-09-17 22:56:37 -0400253}
254
255func (core *Core) startLogicalDeviceManager(ctx context.Context) {
khenaidoo21d51152019-02-01 13:48:37 -0500256 log.Info("Logical-DeviceManager-Starting...")
khenaidoo4d4802d2018-10-04 21:59:49 -0400257 core.logicalDeviceMgr.start(ctx)
khenaidoo21d51152019-02-01 13:48:37 -0500258 log.Info("Logical-DeviceManager-Started")
khenaidoob9203542018-09-17 22:56:37 -0400259}
khenaidoo21d51152019-02-01 13:48:37 -0500260
261func (core *Core) startAdapterManager(ctx context.Context) {
262 log.Info("Adapter-Manager-Starting...")
263 core.adapterMgr.start(ctx)
264 log.Info("Adapter-Manager-Started")
265}