blob: 20e1a520836d76991847d552ce7599a9446ff62e [file] [log] [blame]
Scott Baker2c1c4822019-10-16 11:02:41 -07001/*
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 common
17
18import (
19 "context"
Chaitrashree G S66d41352020-01-09 20:18:58 -050020 "sync"
21
Scott Baker2c1c4822019-10-16 11:02:41 -070022 "github.com/golang/protobuf/ptypes"
23 a "github.com/golang/protobuf/ptypes/any"
serkant.uluderyab38671c2019-11-01 09:35:38 -070024 "github.com/opencord/voltha-lib-go/v3/pkg/kafka"
25 "github.com/opencord/voltha-lib-go/v3/pkg/log"
26 ic "github.com/opencord/voltha-protos/v3/go/inter_container"
27 "github.com/opencord/voltha-protos/v3/go/voltha"
Scott Baker2c1c4822019-10-16 11:02:41 -070028 "google.golang.org/grpc/codes"
29 "google.golang.org/grpc/status"
Scott Baker2c1c4822019-10-16 11:02:41 -070030)
31
32type CoreProxy struct {
Matteo Scandolo2ba00d32020-01-16 17:33:03 -080033 kafkaICProxy kafka.InterContainerProxy
Scott Baker2c1c4822019-10-16 11:02:41 -070034 adapterTopic string
35 coreTopic string
36 deviceIdCoreMap map[string]string
37 lockDeviceIdCoreMap sync.RWMutex
38}
39
Matteo Scandolo2ba00d32020-01-16 17:33:03 -080040func NewCoreProxy(kafkaProxy kafka.InterContainerProxy, adapterTopic string, coreTopic string) *CoreProxy {
Scott Baker2c1c4822019-10-16 11:02:41 -070041 var proxy CoreProxy
42 proxy.kafkaICProxy = kafkaProxy
43 proxy.adapterTopic = adapterTopic
44 proxy.coreTopic = coreTopic
45 proxy.deviceIdCoreMap = make(map[string]string)
46 proxy.lockDeviceIdCoreMap = sync.RWMutex{}
khenaidoob332f9b2020-01-16 16:25:26 -050047 logger.Debugw("TOPICS", log.Fields{"core": proxy.coreTopic, "adapter": proxy.adapterTopic})
Scott Baker2c1c4822019-10-16 11:02:41 -070048
49 return &proxy
50}
51
52func unPackResponse(rpc string, deviceId string, success bool, response *a.Any) error {
53 if success {
54 return nil
55 } else {
56 unpackResult := &ic.Error{}
57 var err error
58 if err = ptypes.UnmarshalAny(response, unpackResult); err != nil {
khenaidoob332f9b2020-01-16 16:25:26 -050059 logger.Warnw("cannot-unmarshal-response", log.Fields{"error": err})
Scott Baker2c1c4822019-10-16 11:02:41 -070060 }
khenaidoob332f9b2020-01-16 16:25:26 -050061 logger.Debugw("response", log.Fields{"rpc": rpc, "deviceId": deviceId, "success": success, "error": err})
Scott Baker2c1c4822019-10-16 11:02:41 -070062 // TODO: Need to get the real error code
63 return status.Errorf(codes.Canceled, "%s", unpackResult.Reason)
64 }
65}
66
67// UpdateCoreReference adds or update a core reference (really the topic name) for a given device Id
68func (ap *CoreProxy) UpdateCoreReference(deviceId string, coreReference string) {
69 ap.lockDeviceIdCoreMap.Lock()
70 defer ap.lockDeviceIdCoreMap.Unlock()
71 ap.deviceIdCoreMap[deviceId] = coreReference
72}
73
74// DeleteCoreReference removes a core reference (really the topic name) for a given device Id
75func (ap *CoreProxy) DeleteCoreReference(deviceId string) {
76 ap.lockDeviceIdCoreMap.Lock()
77 defer ap.lockDeviceIdCoreMap.Unlock()
78 delete(ap.deviceIdCoreMap, deviceId)
79}
80
81func (ap *CoreProxy) getCoreTopic(deviceId string) kafka.Topic {
82 ap.lockDeviceIdCoreMap.Lock()
83 defer ap.lockDeviceIdCoreMap.Unlock()
84
85 if t, exist := ap.deviceIdCoreMap[deviceId]; exist {
86 return kafka.Topic{Name: t}
87 }
88
89 return kafka.Topic{Name: ap.coreTopic}
90}
91
92func (ap *CoreProxy) getAdapterTopic(args ...string) kafka.Topic {
93 return kafka.Topic{Name: ap.adapterTopic}
94}
95
96func (ap *CoreProxy) RegisterAdapter(ctx context.Context, adapter *voltha.Adapter, deviceTypes *voltha.DeviceTypes) error {
khenaidoob332f9b2020-01-16 16:25:26 -050097 logger.Debugw("registering-adapter", log.Fields{"coreTopic": ap.coreTopic, "adapterTopic": ap.adapterTopic})
Scott Baker2c1c4822019-10-16 11:02:41 -070098 rpc := "Register"
99 topic := kafka.Topic{Name: ap.coreTopic}
100 replyToTopic := ap.getAdapterTopic()
101 args := make([]*kafka.KVArg, 2)
Matteo Scandolod58eaef2020-03-30 12:30:02 -0700102
103 if adapter.TotalReplicas == 0 && adapter.CurrentReplica != 0 {
104 log.Fatal("totalReplicas can't be 0, since you're here you have at least one")
105 }
106
107 if adapter.CurrentReplica == 0 && adapter.TotalReplicas != 0 {
108 log.Fatal("currentReplica can't be 0, it has to start from 1")
109 }
110
111 if adapter.CurrentReplica == 0 && adapter.TotalReplicas == 0 {
112 // if the adapter is not setting these fields they default to 0,
113 // in that case it means the adapter is not ready to be scaled and thus it defaults
114 // to a single instance
115 adapter.CurrentReplica = 1
116 adapter.TotalReplicas = 1
117 }
118
119 if adapter.CurrentReplica > adapter.TotalReplicas {
120 log.Fatalf("CurrentReplica (%d) can't be greater than TotalReplicas (%d)",
121 adapter.CurrentReplica, adapter.TotalReplicas)
122 }
123
Scott Baker2c1c4822019-10-16 11:02:41 -0700124 args[0] = &kafka.KVArg{
125 Key: "adapter",
126 Value: adapter,
127 }
128 args[1] = &kafka.KVArg{
129 Key: "deviceTypes",
130 Value: deviceTypes,
131 }
132
133 success, result := ap.kafkaICProxy.InvokeRPC(ctx, rpc, &topic, &replyToTopic, true, "", args...)
khenaidoob332f9b2020-01-16 16:25:26 -0500134 logger.Debugw("Register-Adapter-response", log.Fields{"replyTopic": replyToTopic, "success": success})
Scott Baker2c1c4822019-10-16 11:02:41 -0700135 return unPackResponse(rpc, "", success, result)
136}
137
138func (ap *CoreProxy) DeviceUpdate(ctx context.Context, device *voltha.Device) error {
khenaidoob332f9b2020-01-16 16:25:26 -0500139 logger.Debugw("DeviceUpdate", log.Fields{"deviceId": device.Id})
Scott Baker2c1c4822019-10-16 11:02:41 -0700140 rpc := "DeviceUpdate"
141 toTopic := ap.getCoreTopic(device.Id)
142 args := make([]*kafka.KVArg, 1)
143 args[0] = &kafka.KVArg{
144 Key: "device",
145 Value: device,
146 }
147 // Use a device specific topic as we are the only adaptercore handling requests for this device
148 replyToTopic := ap.getAdapterTopic()
David K. Bainbridge7c75cac2020-02-19 08:53:46 -0800149 success, result := ap.kafkaICProxy.InvokeRPC(context.Background(), rpc, &toTopic, &replyToTopic, true, device.Id, args...)
khenaidoob332f9b2020-01-16 16:25:26 -0500150 logger.Debugw("DeviceUpdate-response", log.Fields{"deviceId": device.Id, "success": success})
Scott Baker2c1c4822019-10-16 11:02:41 -0700151 return unPackResponse(rpc, device.Id, success, result)
152}
153
154func (ap *CoreProxy) PortCreated(ctx context.Context, deviceId string, port *voltha.Port) error {
khenaidoob332f9b2020-01-16 16:25:26 -0500155 logger.Debugw("PortCreated", log.Fields{"portNo": port.PortNo})
Scott Baker2c1c4822019-10-16 11:02:41 -0700156 rpc := "PortCreated"
157 // Use a device specific topic to send the request. The adapter handling the device creates a device
158 // specific topic
159 toTopic := ap.getCoreTopic(deviceId)
160 args := make([]*kafka.KVArg, 2)
161 id := &voltha.ID{Id: deviceId}
162 args[0] = &kafka.KVArg{
163 Key: "device_id",
164 Value: id,
165 }
166 args[1] = &kafka.KVArg{
167 Key: "port",
168 Value: port,
169 }
170
171 // Use a device specific topic as we are the only adaptercore handling requests for this device
172 replyToTopic := ap.getAdapterTopic()
David K. Bainbridge7c75cac2020-02-19 08:53:46 -0800173 success, result := ap.kafkaICProxy.InvokeRPC(context.Background(), rpc, &toTopic, &replyToTopic, true, deviceId, args...)
khenaidoob332f9b2020-01-16 16:25:26 -0500174 logger.Debugw("PortCreated-response", log.Fields{"deviceId": deviceId, "success": success})
Scott Baker2c1c4822019-10-16 11:02:41 -0700175 return unPackResponse(rpc, deviceId, success, result)
176}
177
serkant.uluderyab38671c2019-11-01 09:35:38 -0700178func (ap *CoreProxy) PortsStateUpdate(ctx context.Context, deviceId string, operStatus voltha.OperStatus_Types) error {
Girish Kumare6f45e82020-03-20 10:46:54 +0000179 logger.Debugw("PortsStateUpdate", log.Fields{"deviceId": deviceId})
Scott Baker2c1c4822019-10-16 11:02:41 -0700180 rpc := "PortsStateUpdate"
181 // Use a device specific topic to send the request. The adapter handling the device creates a device
182 // specific topic
183 toTopic := ap.getCoreTopic(deviceId)
184 args := make([]*kafka.KVArg, 2)
185 id := &voltha.ID{Id: deviceId}
186 oStatus := &ic.IntType{Val: int64(operStatus)}
187
188 args[0] = &kafka.KVArg{
189 Key: "device_id",
190 Value: id,
191 }
192 args[1] = &kafka.KVArg{
193 Key: "oper_status",
194 Value: oStatus,
195 }
196
197 // Use a device specific topic as we are the only adaptercore handling requests for this device
198 replyToTopic := ap.getAdapterTopic()
David K. Bainbridge7c75cac2020-02-19 08:53:46 -0800199 success, result := ap.kafkaICProxy.InvokeRPC(context.Background(), rpc, &toTopic, &replyToTopic, true, deviceId, args...)
khenaidoob332f9b2020-01-16 16:25:26 -0500200 logger.Debugw("PortsStateUpdate-response", log.Fields{"deviceId": deviceId, "success": success})
Scott Baker2c1c4822019-10-16 11:02:41 -0700201 return unPackResponse(rpc, deviceId, success, result)
202}
203
204func (ap *CoreProxy) DeleteAllPorts(ctx context.Context, deviceId string) error {
khenaidoob332f9b2020-01-16 16:25:26 -0500205 logger.Debugw("DeleteAllPorts", log.Fields{"deviceId": deviceId})
Scott Baker2c1c4822019-10-16 11:02:41 -0700206 rpc := "DeleteAllPorts"
207 // Use a device specific topic to send the request. The adapter handling the device creates a device
208 // specific topic
209 toTopic := ap.getCoreTopic(deviceId)
210 args := make([]*kafka.KVArg, 2)
211 id := &voltha.ID{Id: deviceId}
212
213 args[0] = &kafka.KVArg{
214 Key: "device_id",
215 Value: id,
216 }
217
218 // Use a device specific topic as we are the only adaptercore handling requests for this device
219 replyToTopic := ap.getAdapterTopic()
David K. Bainbridge7c75cac2020-02-19 08:53:46 -0800220 success, result := ap.kafkaICProxy.InvokeRPC(context.Background(), rpc, &toTopic, &replyToTopic, true, deviceId, args...)
khenaidoob332f9b2020-01-16 16:25:26 -0500221 logger.Debugw("DeleteAllPorts-response", log.Fields{"deviceId": deviceId, "success": success})
Scott Baker2c1c4822019-10-16 11:02:41 -0700222 return unPackResponse(rpc, deviceId, success, result)
223}
224
225func (ap *CoreProxy) DeviceStateUpdate(ctx context.Context, deviceId string,
serkant.uluderyab38671c2019-11-01 09:35:38 -0700226 connStatus voltha.ConnectStatus_Types, operStatus voltha.OperStatus_Types) error {
Girish Kumare6f45e82020-03-20 10:46:54 +0000227 logger.Debugw("DeviceStateUpdate", log.Fields{"deviceId": deviceId})
Scott Baker2c1c4822019-10-16 11:02:41 -0700228 rpc := "DeviceStateUpdate"
229 // Use a device specific topic to send the request. The adapter handling the device creates a device
230 // specific topic
231 toTopic := ap.getCoreTopic(deviceId)
232 args := make([]*kafka.KVArg, 3)
233 id := &voltha.ID{Id: deviceId}
234 oStatus := &ic.IntType{Val: int64(operStatus)}
235 cStatus := &ic.IntType{Val: int64(connStatus)}
236
237 args[0] = &kafka.KVArg{
238 Key: "device_id",
239 Value: id,
240 }
241 args[1] = &kafka.KVArg{
242 Key: "oper_status",
243 Value: oStatus,
244 }
245 args[2] = &kafka.KVArg{
246 Key: "connect_status",
247 Value: cStatus,
248 }
249 // Use a device specific topic as we are the only adaptercore handling requests for this device
250 replyToTopic := ap.getAdapterTopic()
David K. Bainbridge7c75cac2020-02-19 08:53:46 -0800251 success, result := ap.kafkaICProxy.InvokeRPC(context.Background(), rpc, &toTopic, &replyToTopic, true, deviceId, args...)
khenaidoob332f9b2020-01-16 16:25:26 -0500252 logger.Debugw("DeviceStateUpdate-response", log.Fields{"deviceId": deviceId, "success": success})
Scott Baker2c1c4822019-10-16 11:02:41 -0700253 return unPackResponse(rpc, deviceId, success, result)
254}
255
256func (ap *CoreProxy) ChildDeviceDetected(ctx context.Context, parentDeviceId string, parentPortNo int,
257 childDeviceType string, channelId int, vendorId string, serialNumber string, onuId int64) (*voltha.Device, error) {
khenaidoob332f9b2020-01-16 16:25:26 -0500258 logger.Debugw("ChildDeviceDetected", log.Fields{"pDeviceId": parentDeviceId, "channelId": channelId})
Scott Baker2c1c4822019-10-16 11:02:41 -0700259 rpc := "ChildDeviceDetected"
260 // Use a device specific topic to send the request. The adapter handling the device creates a device
261 // specific topic
262 toTopic := ap.getCoreTopic(parentDeviceId)
263 replyToTopic := ap.getAdapterTopic()
264
265 args := make([]*kafka.KVArg, 7)
266 id := &voltha.ID{Id: parentDeviceId}
267 args[0] = &kafka.KVArg{
268 Key: "parent_device_id",
269 Value: id,
270 }
271 ppn := &ic.IntType{Val: int64(parentPortNo)}
272 args[1] = &kafka.KVArg{
273 Key: "parent_port_no",
274 Value: ppn,
275 }
276 cdt := &ic.StrType{Val: childDeviceType}
277 args[2] = &kafka.KVArg{
278 Key: "child_device_type",
279 Value: cdt,
280 }
281 channel := &ic.IntType{Val: int64(channelId)}
282 args[3] = &kafka.KVArg{
283 Key: "channel_id",
284 Value: channel,
285 }
286 vId := &ic.StrType{Val: vendorId}
287 args[4] = &kafka.KVArg{
288 Key: "vendor_id",
289 Value: vId,
290 }
291 sNo := &ic.StrType{Val: serialNumber}
292 args[5] = &kafka.KVArg{
293 Key: "serial_number",
294 Value: sNo,
295 }
296 oId := &ic.IntType{Val: int64(onuId)}
297 args[6] = &kafka.KVArg{
298 Key: "onu_id",
299 Value: oId,
300 }
301
David K. Bainbridge7c75cac2020-02-19 08:53:46 -0800302 success, result := ap.kafkaICProxy.InvokeRPC(context.Background(), rpc, &toTopic, &replyToTopic, true, parentDeviceId, args...)
khenaidoob332f9b2020-01-16 16:25:26 -0500303 logger.Debugw("ChildDeviceDetected-response", log.Fields{"pDeviceId": parentDeviceId, "success": success})
Scott Baker2c1c4822019-10-16 11:02:41 -0700304
305 if success {
306 volthaDevice := &voltha.Device{}
307 if err := ptypes.UnmarshalAny(result, volthaDevice); err != nil {
khenaidoob332f9b2020-01-16 16:25:26 -0500308 logger.Warnw("cannot-unmarshal-response", log.Fields{"error": err})
Matteo Scandolo1530d412020-01-30 14:57:46 -0800309 return nil, status.Error(codes.InvalidArgument, err.Error())
Scott Baker2c1c4822019-10-16 11:02:41 -0700310 }
311 return volthaDevice, nil
312 } else {
313 unpackResult := &ic.Error{}
314 var err error
315 if err = ptypes.UnmarshalAny(result, unpackResult); err != nil {
khenaidoob332f9b2020-01-16 16:25:26 -0500316 logger.Warnw("cannot-unmarshal-response", log.Fields{"error": err})
Scott Baker2c1c4822019-10-16 11:02:41 -0700317 }
khenaidoob332f9b2020-01-16 16:25:26 -0500318 logger.Debugw("ChildDeviceDetected-return", log.Fields{"deviceid": parentDeviceId, "success": success, "error": err})
Matteo Scandolo1530d412020-01-30 14:57:46 -0800319
320 return nil, status.Error(ICProxyErrorCodeToGrpcErrorCode(unpackResult.Code), unpackResult.Reason)
Scott Baker2c1c4822019-10-16 11:02:41 -0700321 }
322
323}
324
325func (ap *CoreProxy) ChildDevicesLost(ctx context.Context, parentDeviceId string) error {
khenaidoob332f9b2020-01-16 16:25:26 -0500326 logger.Debugw("ChildDevicesLost", log.Fields{"pDeviceId": parentDeviceId})
Scott Baker2c1c4822019-10-16 11:02:41 -0700327 rpc := "ChildDevicesLost"
328 // Use a device specific topic to send the request. The adapter handling the device creates a device
329 // specific topic
330 toTopic := ap.getCoreTopic(parentDeviceId)
331 replyToTopic := ap.getAdapterTopic()
332
333 args := make([]*kafka.KVArg, 1)
334 id := &voltha.ID{Id: parentDeviceId}
335 args[0] = &kafka.KVArg{
336 Key: "parent_device_id",
337 Value: id,
338 }
339
David K. Bainbridge7c75cac2020-02-19 08:53:46 -0800340 success, result := ap.kafkaICProxy.InvokeRPC(context.Background(), rpc, &toTopic, &replyToTopic, true, parentDeviceId, args...)
khenaidoob332f9b2020-01-16 16:25:26 -0500341 logger.Debugw("ChildDevicesLost-response", log.Fields{"pDeviceId": parentDeviceId, "success": success})
Scott Baker2c1c4822019-10-16 11:02:41 -0700342 return unPackResponse(rpc, parentDeviceId, success, result)
343}
344
345func (ap *CoreProxy) ChildDevicesDetected(ctx context.Context, parentDeviceId string) error {
khenaidoob332f9b2020-01-16 16:25:26 -0500346 logger.Debugw("ChildDevicesDetected", log.Fields{"pDeviceId": parentDeviceId})
Scott Baker2c1c4822019-10-16 11:02:41 -0700347 rpc := "ChildDevicesDetected"
348 // Use a device specific topic to send the request. The adapter handling the device creates a device
349 // specific topic
350 toTopic := ap.getCoreTopic(parentDeviceId)
351 replyToTopic := ap.getAdapterTopic()
352
353 args := make([]*kafka.KVArg, 1)
354 id := &voltha.ID{Id: parentDeviceId}
355 args[0] = &kafka.KVArg{
356 Key: "parent_device_id",
357 Value: id,
358 }
359
David K. Bainbridge7c75cac2020-02-19 08:53:46 -0800360 success, result := ap.kafkaICProxy.InvokeRPC(context.Background(), rpc, &toTopic, &replyToTopic, true, parentDeviceId, args...)
khenaidoob332f9b2020-01-16 16:25:26 -0500361 logger.Debugw("ChildDevicesDetected-response", log.Fields{"pDeviceId": parentDeviceId, "success": success})
Scott Baker2c1c4822019-10-16 11:02:41 -0700362 return unPackResponse(rpc, parentDeviceId, success, result)
363}
364
365func (ap *CoreProxy) GetDevice(ctx context.Context, parentDeviceId string, deviceId string) (*voltha.Device, error) {
khenaidoob332f9b2020-01-16 16:25:26 -0500366 logger.Debugw("GetDevice", log.Fields{"deviceId": deviceId})
Scott Baker2c1c4822019-10-16 11:02:41 -0700367 rpc := "GetDevice"
368
369 toTopic := ap.getCoreTopic(parentDeviceId)
370 replyToTopic := ap.getAdapterTopic()
371
372 args := make([]*kafka.KVArg, 1)
373 id := &voltha.ID{Id: deviceId}
374 args[0] = &kafka.KVArg{
375 Key: "device_id",
376 Value: id,
377 }
378
David K. Bainbridge7c75cac2020-02-19 08:53:46 -0800379 success, result := ap.kafkaICProxy.InvokeRPC(context.Background(), rpc, &toTopic, &replyToTopic, true, parentDeviceId, args...)
khenaidoob332f9b2020-01-16 16:25:26 -0500380 logger.Debugw("GetDevice-response", log.Fields{"pDeviceId": parentDeviceId, "success": success})
Scott Baker2c1c4822019-10-16 11:02:41 -0700381
382 if success {
383 volthaDevice := &voltha.Device{}
384 if err := ptypes.UnmarshalAny(result, volthaDevice); err != nil {
khenaidoob332f9b2020-01-16 16:25:26 -0500385 logger.Warnw("cannot-unmarshal-response", log.Fields{"error": err})
Matteo Scandolo1530d412020-01-30 14:57:46 -0800386 return nil, status.Error(codes.InvalidArgument, err.Error())
Scott Baker2c1c4822019-10-16 11:02:41 -0700387 }
388 return volthaDevice, nil
389 } else {
390 unpackResult := &ic.Error{}
391 var err error
392 if err = ptypes.UnmarshalAny(result, unpackResult); err != nil {
khenaidoob332f9b2020-01-16 16:25:26 -0500393 logger.Warnw("cannot-unmarshal-response", log.Fields{"error": err})
Scott Baker2c1c4822019-10-16 11:02:41 -0700394 }
khenaidoob332f9b2020-01-16 16:25:26 -0500395 logger.Debugw("GetDevice-return", log.Fields{"deviceid": parentDeviceId, "success": success, "error": err})
Scott Baker2c1c4822019-10-16 11:02:41 -0700396 // TODO: Need to get the real error code
Matteo Scandolo1530d412020-01-30 14:57:46 -0800397 return nil, status.Error(ICProxyErrorCodeToGrpcErrorCode(unpackResult.Code), unpackResult.Reason)
Scott Baker2c1c4822019-10-16 11:02:41 -0700398 }
399}
400
401func (ap *CoreProxy) GetChildDevice(ctx context.Context, parentDeviceId string, kwargs map[string]interface{}) (*voltha.Device, error) {
khenaidoob332f9b2020-01-16 16:25:26 -0500402 logger.Debugw("GetChildDevice", log.Fields{"parentDeviceId": parentDeviceId, "kwargs": kwargs})
Scott Baker2c1c4822019-10-16 11:02:41 -0700403 rpc := "GetChildDevice"
404
405 toTopic := ap.getCoreTopic(parentDeviceId)
406 replyToTopic := ap.getAdapterTopic()
407
408 args := make([]*kafka.KVArg, 4)
409 id := &voltha.ID{Id: parentDeviceId}
410 args[0] = &kafka.KVArg{
411 Key: "device_id",
412 Value: id,
413 }
414
415 var cnt uint8 = 0
416 for k, v := range kwargs {
417 cnt += 1
418 if k == "serial_number" {
419 val := &ic.StrType{Val: v.(string)}
420 args[cnt] = &kafka.KVArg{
421 Key: k,
422 Value: val,
423 }
424 } else if k == "onu_id" {
425 val := &ic.IntType{Val: int64(v.(uint32))}
426 args[cnt] = &kafka.KVArg{
427 Key: k,
428 Value: val,
429 }
430 } else if k == "parent_port_no" {
431 val := &ic.IntType{Val: int64(v.(uint32))}
432 args[cnt] = &kafka.KVArg{
433 Key: k,
434 Value: val,
435 }
436 }
437 }
438
David K. Bainbridge7c75cac2020-02-19 08:53:46 -0800439 success, result := ap.kafkaICProxy.InvokeRPC(context.Background(), rpc, &toTopic, &replyToTopic, true, parentDeviceId, args...)
khenaidoob332f9b2020-01-16 16:25:26 -0500440 logger.Debugw("GetChildDevice-response", log.Fields{"pDeviceId": parentDeviceId, "success": success})
Scott Baker2c1c4822019-10-16 11:02:41 -0700441
442 if success {
443 volthaDevice := &voltha.Device{}
444 if err := ptypes.UnmarshalAny(result, volthaDevice); err != nil {
khenaidoob332f9b2020-01-16 16:25:26 -0500445 logger.Warnw("cannot-unmarshal-response", log.Fields{"error": err})
Matteo Scandolo1530d412020-01-30 14:57:46 -0800446 return nil, status.Error(codes.InvalidArgument, err.Error())
Scott Baker2c1c4822019-10-16 11:02:41 -0700447 }
448 return volthaDevice, nil
449 } else {
450 unpackResult := &ic.Error{}
451 var err error
452 if err = ptypes.UnmarshalAny(result, unpackResult); err != nil {
khenaidoob332f9b2020-01-16 16:25:26 -0500453 logger.Warnw("cannot-unmarshal-response", log.Fields{"error": err})
Scott Baker2c1c4822019-10-16 11:02:41 -0700454 }
khenaidoob332f9b2020-01-16 16:25:26 -0500455 logger.Debugw("GetChildDevice-return", log.Fields{"deviceid": parentDeviceId, "success": success, "error": err})
Matteo Scandolob45cf592020-01-21 16:10:56 -0800456
Matteo Scandolo1530d412020-01-30 14:57:46 -0800457 return nil, status.Error(ICProxyErrorCodeToGrpcErrorCode(unpackResult.Code), unpackResult.Reason)
Scott Baker2c1c4822019-10-16 11:02:41 -0700458 }
459}
460
461func (ap *CoreProxy) GetChildDevices(ctx context.Context, parentDeviceId string) (*voltha.Devices, error) {
khenaidoob332f9b2020-01-16 16:25:26 -0500462 logger.Debugw("GetChildDevices", log.Fields{"parentDeviceId": parentDeviceId})
Scott Baker2c1c4822019-10-16 11:02:41 -0700463 rpc := "GetChildDevices"
464
465 toTopic := ap.getCoreTopic(parentDeviceId)
466 replyToTopic := ap.getAdapterTopic()
467
468 args := make([]*kafka.KVArg, 1)
469 id := &voltha.ID{Id: parentDeviceId}
470 args[0] = &kafka.KVArg{
471 Key: "device_id",
472 Value: id,
473 }
474
David K. Bainbridge7c75cac2020-02-19 08:53:46 -0800475 success, result := ap.kafkaICProxy.InvokeRPC(context.Background(), rpc, &toTopic, &replyToTopic, true, parentDeviceId, args...)
khenaidoob332f9b2020-01-16 16:25:26 -0500476 logger.Debugw("GetChildDevices-response", log.Fields{"pDeviceId": parentDeviceId, "success": success})
Scott Baker2c1c4822019-10-16 11:02:41 -0700477
478 if success {
479 volthaDevices := &voltha.Devices{}
480 if err := ptypes.UnmarshalAny(result, volthaDevices); err != nil {
khenaidoob332f9b2020-01-16 16:25:26 -0500481 logger.Warnw("cannot-unmarshal-response", log.Fields{"error": err})
Matteo Scandolo1530d412020-01-30 14:57:46 -0800482 return nil, status.Error(codes.InvalidArgument, err.Error())
Scott Baker2c1c4822019-10-16 11:02:41 -0700483 }
484 return volthaDevices, nil
485 } else {
486 unpackResult := &ic.Error{}
487 var err error
488 if err = ptypes.UnmarshalAny(result, unpackResult); err != nil {
khenaidoob332f9b2020-01-16 16:25:26 -0500489 logger.Warnw("cannot-unmarshal-response", log.Fields{"error": err})
Scott Baker2c1c4822019-10-16 11:02:41 -0700490 }
khenaidoob332f9b2020-01-16 16:25:26 -0500491 logger.Debugw("GetChildDevices-return", log.Fields{"deviceid": parentDeviceId, "success": success, "error": err})
Matteo Scandolob45cf592020-01-21 16:10:56 -0800492
Matteo Scandolo1530d412020-01-30 14:57:46 -0800493 return nil, status.Error(ICProxyErrorCodeToGrpcErrorCode(unpackResult.Code), unpackResult.Reason)
Scott Baker2c1c4822019-10-16 11:02:41 -0700494 }
495}
496
497func (ap *CoreProxy) SendPacketIn(ctx context.Context, deviceId string, port uint32, pktPayload []byte) error {
khenaidoob332f9b2020-01-16 16:25:26 -0500498 logger.Debugw("SendPacketIn", log.Fields{"deviceId": deviceId, "port": port, "pktPayload": pktPayload})
Scott Baker2c1c4822019-10-16 11:02:41 -0700499 rpc := "PacketIn"
500 // Use a device specific topic to send the request. The adapter handling the device creates a device
501 // specific topic
502 toTopic := ap.getCoreTopic(deviceId)
503 replyToTopic := ap.getAdapterTopic()
504
505 args := make([]*kafka.KVArg, 3)
506 id := &voltha.ID{Id: deviceId}
507 args[0] = &kafka.KVArg{
508 Key: "device_id",
509 Value: id,
510 }
511 portNo := &ic.IntType{Val: int64(port)}
512 args[1] = &kafka.KVArg{
513 Key: "port",
514 Value: portNo,
515 }
516 pkt := &ic.Packet{Payload: pktPayload}
517 args[2] = &kafka.KVArg{
518 Key: "packet",
519 Value: pkt,
520 }
David K. Bainbridge7c75cac2020-02-19 08:53:46 -0800521 success, result := ap.kafkaICProxy.InvokeRPC(context.Background(), rpc, &toTopic, &replyToTopic, true, deviceId, args...)
khenaidoob332f9b2020-01-16 16:25:26 -0500522 logger.Debugw("SendPacketIn-response", log.Fields{"pDeviceId": deviceId, "success": success})
Scott Baker2c1c4822019-10-16 11:02:41 -0700523 return unPackResponse(rpc, deviceId, success, result)
524}
525
David Bainbridge36d0b202019-10-23 18:33:34 +0000526func (ap *CoreProxy) DeviceReasonUpdate(ctx context.Context, deviceId string, deviceReason string) error {
khenaidoob332f9b2020-01-16 16:25:26 -0500527 logger.Debugw("DeviceReasonUpdate", log.Fields{"deviceId": deviceId, "deviceReason": deviceReason})
David Bainbridge36d0b202019-10-23 18:33:34 +0000528 rpc := "DeviceReasonUpdate"
529 // Use a device specific topic to send the request. The adapter handling the device creates a device
530 // specific topic
531 toTopic := ap.getCoreTopic(deviceId)
532 replyToTopic := ap.getAdapterTopic()
533
534 args := make([]*kafka.KVArg, 2)
535 id := &voltha.ID{Id: deviceId}
536 args[0] = &kafka.KVArg{
537 Key: "device_id",
538 Value: id,
539 }
540 reason := &ic.StrType{Val: deviceReason}
541 args[1] = &kafka.KVArg{
542 Key: "device_reason",
543 Value: reason,
544 }
David K. Bainbridge7c75cac2020-02-19 08:53:46 -0800545 success, result := ap.kafkaICProxy.InvokeRPC(context.Background(), rpc, &toTopic, &replyToTopic, true, deviceId, args...)
khenaidoob332f9b2020-01-16 16:25:26 -0500546 logger.Debugw("DeviceReason-response", log.Fields{"pDeviceId": deviceId, "success": success})
David Bainbridge36d0b202019-10-23 18:33:34 +0000547 return unPackResponse(rpc, deviceId, success, result)
548}
549
Scott Baker2c1c4822019-10-16 11:02:41 -0700550func (ap *CoreProxy) DevicePMConfigUpdate(ctx context.Context, pmConfigs *voltha.PmConfigs) error {
khenaidoob332f9b2020-01-16 16:25:26 -0500551 logger.Debugw("DevicePMConfigUpdate", log.Fields{"pmConfigs": pmConfigs})
Scott Baker2c1c4822019-10-16 11:02:41 -0700552 rpc := "DevicePMConfigUpdate"
553 // Use a device specific topic to send the request. The adapter handling the device creates a device
554 // specific topic
555 toTopic := ap.getCoreTopic(pmConfigs.Id)
556 replyToTopic := ap.getAdapterTopic()
557
558 args := make([]*kafka.KVArg, 1)
559 args[0] = &kafka.KVArg{
560 Key: "device_pm_config",
561 Value: pmConfigs,
562 }
David K. Bainbridge7c75cac2020-02-19 08:53:46 -0800563 success, result := ap.kafkaICProxy.InvokeRPC(context.Background(), rpc, &toTopic, &replyToTopic, true, pmConfigs.Id, args...)
khenaidoob332f9b2020-01-16 16:25:26 -0500564 logger.Debugw("DevicePMConfigUpdate-response", log.Fields{"pDeviceId": pmConfigs.Id, "success": success})
Scott Baker2c1c4822019-10-16 11:02:41 -0700565 return unPackResponse(rpc, pmConfigs.Id, success, result)
566}
567
568func (ap *CoreProxy) ReconcileChildDevices(ctx context.Context, parentDeviceId string) error {
khenaidoob332f9b2020-01-16 16:25:26 -0500569 logger.Debugw("ReconcileChildDevices", log.Fields{"parentDeviceId": parentDeviceId})
Scott Baker2c1c4822019-10-16 11:02:41 -0700570 rpc := "ReconcileChildDevices"
571 // Use a device specific topic to send the request. The adapter handling the device creates a device
572 // specific topic
573 toTopic := ap.getCoreTopic(parentDeviceId)
574 replyToTopic := ap.getAdapterTopic()
575
576 args := []*kafka.KVArg{
577 {Key: "parent_device_id", Value: &voltha.ID{Id: parentDeviceId}},
578 }
579
David K. Bainbridge7c75cac2020-02-19 08:53:46 -0800580 success, result := ap.kafkaICProxy.InvokeRPC(context.Background(), rpc, &toTopic, &replyToTopic, true, parentDeviceId, args...)
khenaidoob332f9b2020-01-16 16:25:26 -0500581 logger.Debugw("ReconcileChildDevices-response", log.Fields{"pDeviceId": parentDeviceId, "success": success})
Scott Baker2c1c4822019-10-16 11:02:41 -0700582 return unPackResponse(rpc, parentDeviceId, success, result)
583}
Chaitrashree G S66d41352020-01-09 20:18:58 -0500584
585func (ap *CoreProxy) PortStateUpdate(ctx context.Context, deviceId string, pType voltha.Port_PortType, portNum uint32,
serkant.uluderyab38671c2019-11-01 09:35:38 -0700586 operStatus voltha.OperStatus_Types) error {
khenaidoob332f9b2020-01-16 16:25:26 -0500587 logger.Debugw("PortStateUpdate", log.Fields{"deviceId": deviceId, "portType": pType, "portNo": portNum, "operation_status": operStatus})
Chaitrashree G S66d41352020-01-09 20:18:58 -0500588 rpc := "PortStateUpdate"
589 // Use a device specific topic to send the request. The adapter handling the device creates a device
590 // specific topic
591 toTopic := ap.getCoreTopic(deviceId)
592 args := make([]*kafka.KVArg, 4)
593 deviceID := &voltha.ID{Id: deviceId}
594 portNo := &ic.IntType{Val: int64(portNum)}
595 portType := &ic.IntType{Val: int64(pType)}
596 oStatus := &ic.IntType{Val: int64(operStatus)}
597
598 args[0] = &kafka.KVArg{
599 Key: "device_id",
600 Value: deviceID,
601 }
602 args[1] = &kafka.KVArg{
603 Key: "oper_status",
604 Value: oStatus,
605 }
606 args[2] = &kafka.KVArg{
607 Key: "port_type",
608 Value: portType,
609 }
610 args[3] = &kafka.KVArg{
611 Key: "port_no",
612 Value: portNo,
613 }
614
615 // Use a device specific topic as we are the only adaptercore handling requests for this device
616 replyToTopic := ap.getAdapterTopic()
David K. Bainbridge7c75cac2020-02-19 08:53:46 -0800617 success, result := ap.kafkaICProxy.InvokeRPC(context.Background(), rpc, &toTopic, &replyToTopic, true, deviceId, args...)
khenaidoob332f9b2020-01-16 16:25:26 -0500618 logger.Debugw("PortStateUpdate-response", log.Fields{"deviceId": deviceId, "success": success})
Chaitrashree G S66d41352020-01-09 20:18:58 -0500619 return unPackResponse(rpc, deviceId, success, result)
620}