VOL-2909 - Disaggregating rw_core/core/.
This breaks the core package into logical components. (adapter manager, adapter proxy, devices, nbi/api), as well as the "core" which aggregates all these.
Change-Id: I257ac64024a1cf3efe3f5d89d508e60e6e681fb1
diff --git a/rw_core/core/api/adapter_request_handler.go b/rw_core/core/api/adapter_request_handler.go
new file mode 100644
index 0000000..7175a2b
--- /dev/null
+++ b/rw_core/core/api/adapter_request_handler.go
@@ -0,0 +1,882 @@
+/*
+ * 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
+
+ * http://www.apache.org/licenses/LICENSE-2.0
+
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package api
+
+import (
+ "context"
+ "errors"
+ "github.com/opencord/voltha-go/rw_core/core/adapter"
+ "github.com/opencord/voltha-go/rw_core/core/device"
+ "time"
+
+ "github.com/golang/protobuf/ptypes"
+ "github.com/golang/protobuf/ptypes/empty"
+ "github.com/opencord/voltha-go/db/model"
+ "github.com/opencord/voltha-lib-go/v3/pkg/kafka"
+ "github.com/opencord/voltha-lib-go/v3/pkg/log"
+ ic "github.com/opencord/voltha-protos/v3/go/inter_container"
+ "github.com/opencord/voltha-protos/v3/go/voltha"
+)
+
+// AdapterRequestHandlerProxy represent adapter request handler proxy attributes
+type AdapterRequestHandlerProxy struct {
+ coreInstanceID string
+ deviceMgr *device.Manager
+ adapterMgr *adapter.Manager
+ localDataProxy *model.Proxy
+ clusterDataProxy *model.Proxy
+ defaultRequestTimeout time.Duration
+ longRunningRequestTimeout time.Duration
+}
+
+// NewAdapterRequestHandlerProxy assigns values for adapter request handler proxy attributes and returns the new instance
+func NewAdapterRequestHandlerProxy(coreInstanceID string, dMgr *device.Manager,
+ aMgr *adapter.Manager, cdProxy *model.Proxy, ldProxy *model.Proxy, longRunningRequestTimeout time.Duration,
+ defaultRequestTimeout time.Duration) *AdapterRequestHandlerProxy {
+ var proxy AdapterRequestHandlerProxy
+ proxy.coreInstanceID = coreInstanceID
+ proxy.deviceMgr = dMgr
+ proxy.clusterDataProxy = cdProxy
+ proxy.localDataProxy = ldProxy
+ proxy.adapterMgr = aMgr
+ proxy.defaultRequestTimeout = defaultRequestTimeout
+ proxy.longRunningRequestTimeout = longRunningRequestTimeout
+ return &proxy
+}
+
+func (rhp *AdapterRequestHandlerProxy) Register(args []*ic.Argument) (*voltha.CoreInstance, error) {
+ if len(args) < 3 {
+ logger.Warn("invalid-number-of-args", log.Fields{"args": args})
+ err := errors.New("invalid-number-of-args")
+ return nil, err
+ }
+ adapter := &voltha.Adapter{}
+ deviceTypes := &voltha.DeviceTypes{}
+ transactionID := &ic.StrType{}
+ for _, arg := range args {
+ switch arg.Key {
+ case "adapter":
+ if err := ptypes.UnmarshalAny(arg.Value, adapter); err != nil {
+ logger.Warnw("cannot-unmarshal-adapter", log.Fields{"error": err})
+ return nil, err
+ }
+ case "deviceTypes":
+ if err := ptypes.UnmarshalAny(arg.Value, deviceTypes); err != nil {
+ logger.Warnw("cannot-unmarshal-device-types", log.Fields{"error": err})
+ return nil, err
+ }
+ case kafka.TransactionKey:
+ if err := ptypes.UnmarshalAny(arg.Value, transactionID); err != nil {
+ logger.Warnw("cannot-unmarshal-transaction-ID", log.Fields{"error": err})
+ return nil, err
+ }
+ }
+ }
+ logger.Debugw("Register", log.Fields{"adapter": *adapter, "device-types": deviceTypes, "transaction-id": transactionID.Val, "core-id": rhp.coreInstanceID})
+
+ return rhp.adapterMgr.RegisterAdapter(adapter, deviceTypes)
+}
+
+// GetDevice returns device info
+func (rhp *AdapterRequestHandlerProxy) GetDevice(args []*ic.Argument) (*voltha.Device, error) {
+ if len(args) < 2 {
+ logger.Warn("invalid-number-of-args", log.Fields{"args": args})
+ err := errors.New("invalid-number-of-args")
+ return nil, err
+ }
+
+ pID := &voltha.ID{}
+ transactionID := &ic.StrType{}
+ for _, arg := range args {
+ switch arg.Key {
+ case "device_id":
+ if err := ptypes.UnmarshalAny(arg.Value, pID); err != nil {
+ logger.Warnw("cannot-unmarshal-ID", log.Fields{"error": err})
+ return nil, err
+ }
+ case kafka.TransactionKey:
+ if err := ptypes.UnmarshalAny(arg.Value, transactionID); err != nil {
+ logger.Warnw("cannot-unmarshal-transaction-ID", log.Fields{"error": err})
+ return nil, err
+ }
+ }
+ }
+ logger.Debugw("GetDevice", log.Fields{"deviceID": pID.Id, "transactionID": transactionID.Val})
+
+ // Get the device via the device manager
+ device, err := rhp.deviceMgr.GetDevice(context.TODO(), pID.Id)
+ if err != nil {
+ logger.Debugw("get-device-failed", log.Fields{"deviceID": pID.Id, "error": err})
+ }
+ return device, err
+}
+
+// DeviceUpdate updates device using adapter data
+func (rhp *AdapterRequestHandlerProxy) DeviceUpdate(args []*ic.Argument) (*empty.Empty, error) {
+ if len(args) < 2 {
+ logger.Warn("invalid-number-of-args", log.Fields{"args": args})
+ err := errors.New("invalid-number-of-args")
+ return nil, err
+ }
+
+ device := &voltha.Device{}
+ transactionID := &ic.StrType{}
+ for _, arg := range args {
+ switch arg.Key {
+ case "device":
+ if err := ptypes.UnmarshalAny(arg.Value, device); err != nil {
+ logger.Warnw("cannot-unmarshal-ID", log.Fields{"error": err})
+ return nil, err
+ }
+ case kafka.TransactionKey:
+ if err := ptypes.UnmarshalAny(arg.Value, transactionID); err != nil {
+ logger.Warnw("cannot-unmarshal-transaction-ID", log.Fields{"error": err})
+ return nil, err
+ }
+ }
+ }
+ logger.Debugw("DeviceUpdate", log.Fields{"deviceID": device.Id, "transactionID": transactionID.Val})
+
+ if err := rhp.deviceMgr.UpdateDeviceUsingAdapterData(context.TODO(), device); err != nil {
+ logger.Debugw("unable-to-update-device-using-adapter-data", log.Fields{"error": err})
+ return nil, err
+ }
+ return &empty.Empty{}, nil
+}
+
+// GetChildDevice returns details of child device
+func (rhp *AdapterRequestHandlerProxy) GetChildDevice(args []*ic.Argument) (*voltha.Device, error) {
+ if len(args) < 3 {
+ logger.Warn("invalid-number-of-args", log.Fields{"args": args})
+ err := errors.New("invalid-number-of-args")
+ return nil, err
+ }
+
+ pID := &voltha.ID{}
+ transactionID := &ic.StrType{}
+ serialNumber := &ic.StrType{}
+ onuID := &ic.IntType{}
+ parentPortNo := &ic.IntType{}
+ for _, arg := range args {
+ switch arg.Key {
+ case "device_id":
+ if err := ptypes.UnmarshalAny(arg.Value, pID); err != nil {
+ logger.Warnw("cannot-unmarshal-ID", log.Fields{"error": err})
+ return nil, err
+ }
+ case "serial_number":
+ if err := ptypes.UnmarshalAny(arg.Value, serialNumber); err != nil {
+ logger.Warnw("cannot-unmarshal-ID", log.Fields{"error": err})
+ return nil, err
+ }
+ case "onu_id":
+ if err := ptypes.UnmarshalAny(arg.Value, onuID); err != nil {
+ logger.Warnw("cannot-unmarshal-ID", log.Fields{"error": err})
+ return nil, err
+ }
+ case "parent_port_no":
+ if err := ptypes.UnmarshalAny(arg.Value, parentPortNo); err != nil {
+ logger.Warnw("cannot-unmarshal-ID", log.Fields{"error": err})
+ return nil, err
+ }
+ case kafka.TransactionKey:
+ if err := ptypes.UnmarshalAny(arg.Value, transactionID); err != nil {
+ logger.Warnw("cannot-unmarshal-transaction-ID", log.Fields{"error": err})
+ return nil, err
+ }
+ }
+ }
+ logger.Debugw("GetChildDevice", log.Fields{"parentDeviceID": pID.Id, "args": args, "transactionID": transactionID.Val})
+
+ return rhp.deviceMgr.GetChildDevice(context.TODO(), pID.Id, serialNumber.Val, onuID.Val, parentPortNo.Val)
+}
+
+// GetChildDeviceWithProxyAddress returns details of child device with proxy address
+func (rhp *AdapterRequestHandlerProxy) GetChildDeviceWithProxyAddress(args []*ic.Argument) (*voltha.Device, error) {
+ if len(args) < 2 {
+ logger.Warn("invalid-number-of-args", log.Fields{"args": args})
+ err := errors.New("invalid-number-of-args")
+ return nil, err
+ }
+
+ proxyAddress := &voltha.Device_ProxyAddress{}
+ transactionID := &ic.StrType{}
+ for _, arg := range args {
+ switch arg.Key {
+ case "proxy_address":
+ if err := ptypes.UnmarshalAny(arg.Value, proxyAddress); err != nil {
+ logger.Warnw("cannot-unmarshal-proxy-address", log.Fields{"error": err})
+ return nil, err
+ }
+ case kafka.TransactionKey:
+ if err := ptypes.UnmarshalAny(arg.Value, transactionID); err != nil {
+ logger.Warnw("cannot-unmarshal-transaction-ID", log.Fields{"error": err})
+ return nil, err
+ }
+ }
+ }
+ logger.Debugw("GetChildDeviceWithProxyAddress", log.Fields{"proxyAddress": proxyAddress, "transactionID": transactionID.Val})
+
+ return rhp.deviceMgr.GetChildDeviceWithProxyAddress(context.TODO(), proxyAddress)
+}
+
+// GetPorts returns the ports information of the device based on the port type.
+func (rhp *AdapterRequestHandlerProxy) GetPorts(args []*ic.Argument) (*voltha.Ports, error) {
+ if len(args) < 3 {
+ logger.Warn("invalid-number-of-args", log.Fields{"args": args})
+ err := errors.New("invalid-number-of-args")
+ return nil, err
+ }
+ deviceID := &voltha.ID{}
+ pt := &ic.IntType{}
+ transactionID := &ic.StrType{}
+ for _, arg := range args {
+ switch arg.Key {
+ case "device_id":
+ if err := ptypes.UnmarshalAny(arg.Value, deviceID); err != nil {
+ logger.Warnw("cannot-unmarshal-device-id", log.Fields{"error": err})
+ return nil, err
+ }
+ case "port_type":
+ if err := ptypes.UnmarshalAny(arg.Value, pt); err != nil {
+ logger.Warnw("cannot-unmarshal-porttype", log.Fields{"error": err})
+ return nil, err
+ }
+ case kafka.TransactionKey:
+ if err := ptypes.UnmarshalAny(arg.Value, transactionID); err != nil {
+ logger.Warnw("cannot-unmarshal-transaction-ID", log.Fields{"error": err})
+ return nil, err
+ }
+ }
+ }
+ logger.Debugw("GetPorts", log.Fields{"deviceID": deviceID.Id, "portype": pt.Val, "transactionID": transactionID.Val})
+
+ return rhp.deviceMgr.GetPorts(context.TODO(), deviceID.Id, voltha.Port_PortType(pt.Val))
+}
+
+// GetChildDevices gets all the child device IDs from the device passed as parameter
+func (rhp *AdapterRequestHandlerProxy) GetChildDevices(args []*ic.Argument) (*voltha.Devices, error) {
+ if len(args) < 2 {
+ logger.Warn("invalid-number-of-args", log.Fields{"args": args})
+ err := errors.New("invalid-number-of-args")
+ return nil, err
+ }
+
+ pID := &voltha.ID{}
+ transactionID := &ic.StrType{}
+ for _, arg := range args {
+ switch arg.Key {
+ case "device_id":
+ if err := ptypes.UnmarshalAny(arg.Value, pID); err != nil {
+ logger.Warnw("cannot-unmarshal-ID", log.Fields{"error": err})
+ return nil, err
+ }
+ case kafka.TransactionKey:
+ if err := ptypes.UnmarshalAny(arg.Value, transactionID); err != nil {
+ logger.Warnw("cannot-unmarshal-transaction-ID", log.Fields{"error": err})
+ return nil, err
+ }
+ }
+ }
+ logger.Debugw("GetChildDevices", log.Fields{"deviceID": pID.Id, "transactionID": transactionID.Val})
+
+ return rhp.deviceMgr.GetAllChildDevices(context.TODO(), pID.Id)
+}
+
+// ChildDeviceDetected is invoked when a child device is detected. The following parameters are expected:
+// {parent_device_id, parent_port_no, child_device_type, channel_id, vendor_id, serial_number)
+func (rhp *AdapterRequestHandlerProxy) ChildDeviceDetected(args []*ic.Argument) (*voltha.Device, error) {
+ if len(args) < 5 {
+ logger.Warn("invalid-number-of-args", log.Fields{"args": args})
+ err := errors.New("invalid-number-of-args")
+ return nil, err
+ }
+
+ pID := &voltha.ID{}
+ portNo := &ic.IntType{}
+ dt := &ic.StrType{}
+ chnlID := &ic.IntType{}
+ transactionID := &ic.StrType{}
+ serialNumber := &ic.StrType{}
+ vendorID := &ic.StrType{}
+ onuID := &ic.IntType{}
+ for _, arg := range args {
+ switch arg.Key {
+ case "parent_device_id":
+ if err := ptypes.UnmarshalAny(arg.Value, pID); err != nil {
+ logger.Warnw("cannot-unmarshal-parent-device-id", log.Fields{"error": err})
+ return nil, err
+ }
+ case "parent_port_no":
+ if err := ptypes.UnmarshalAny(arg.Value, portNo); err != nil {
+ logger.Warnw("cannot-unmarshal-parent-port", log.Fields{"error": err})
+ return nil, err
+ }
+ case "child_device_type":
+ if err := ptypes.UnmarshalAny(arg.Value, dt); err != nil {
+ logger.Warnw("cannot-unmarshal-child-device-type", log.Fields{"error": err})
+ return nil, err
+ }
+ case "channel_id":
+ if err := ptypes.UnmarshalAny(arg.Value, chnlID); err != nil {
+ logger.Warnw("cannot-unmarshal-channel-id", log.Fields{"error": err})
+ return nil, err
+ }
+ case "vendor_id":
+ if err := ptypes.UnmarshalAny(arg.Value, vendorID); err != nil {
+ logger.Warnw("cannot-unmarshal-vendor-id", log.Fields{"error": err})
+ return nil, err
+ }
+ case "serial_number":
+ if err := ptypes.UnmarshalAny(arg.Value, serialNumber); err != nil {
+ logger.Warnw("cannot-unmarshal-serial-number", log.Fields{"error": err})
+ return nil, err
+ }
+ case "onu_id":
+ if err := ptypes.UnmarshalAny(arg.Value, onuID); err != nil {
+ logger.Warnw("cannot-unmarshal-onu-id", log.Fields{"error": err})
+ return nil, err
+ }
+ case kafka.TransactionKey:
+ if err := ptypes.UnmarshalAny(arg.Value, transactionID); err != nil {
+ logger.Warnw("cannot-unmarshal-transaction-ID", log.Fields{"error": err})
+ return nil, err
+ }
+ }
+ }
+ logger.Debugw("ChildDeviceDetected", log.Fields{"parentDeviceID": pID.Id, "parentPortNo": portNo.Val,
+ "deviceType": dt.Val, "channelID": chnlID.Val, "serialNumber": serialNumber.Val,
+ "vendorID": vendorID.Val, "onuID": onuID.Val, "transactionID": transactionID.Val})
+
+ device, err := rhp.deviceMgr.ChildDeviceDetected(context.TODO(), pID.Id, portNo.Val, dt.Val, chnlID.Val, vendorID.Val, serialNumber.Val, onuID.Val)
+ if err != nil {
+ logger.Debugw("child-detection-failed", log.Fields{"parentID": pID.Id, "onuID": onuID.Val, "error": err})
+ }
+ return device, err
+}
+
+// DeviceStateUpdate updates device status
+func (rhp *AdapterRequestHandlerProxy) DeviceStateUpdate(args []*ic.Argument) (*empty.Empty, error) {
+ if len(args) < 3 {
+ logger.Warn("invalid-number-of-args", log.Fields{"args": args})
+ err := errors.New("invalid-number-of-args")
+ return nil, err
+ }
+ deviceID := &voltha.ID{}
+ operStatus := &ic.IntType{}
+ connStatus := &ic.IntType{}
+ transactionID := &ic.StrType{}
+ for _, arg := range args {
+ switch arg.Key {
+ case "device_id":
+ if err := ptypes.UnmarshalAny(arg.Value, deviceID); err != nil {
+ logger.Warnw("cannot-unmarshal-device-id", log.Fields{"error": err})
+ return nil, err
+ }
+ case "oper_status":
+ if err := ptypes.UnmarshalAny(arg.Value, operStatus); err != nil {
+ logger.Warnw("cannot-unmarshal-operStatus", log.Fields{"error": err})
+ return nil, err
+ }
+ case "connect_status":
+ if err := ptypes.UnmarshalAny(arg.Value, connStatus); err != nil {
+ logger.Warnw("cannot-unmarshal-connStatus", log.Fields{"error": err})
+ return nil, err
+ }
+ case kafka.TransactionKey:
+ if err := ptypes.UnmarshalAny(arg.Value, transactionID); err != nil {
+ logger.Warnw("cannot-unmarshal-transaction-ID", log.Fields{"error": err})
+ return nil, err
+ }
+ }
+ }
+ logger.Debugw("DeviceStateUpdate", log.Fields{"deviceID": deviceID.Id, "oper-status": operStatus,
+ "conn-status": connStatus, "transactionID": transactionID.Val})
+
+ if err := rhp.deviceMgr.UpdateDeviceStatus(context.TODO(), deviceID.Id, voltha.OperStatus_Types(operStatus.Val),
+ voltha.ConnectStatus_Types(connStatus.Val)); err != nil {
+ logger.Debugw("unable-to-update-device-status", log.Fields{"error": err})
+ return nil, err
+ }
+ return &empty.Empty{}, nil
+}
+
+// ChildrenStateUpdate updates child device status
+func (rhp *AdapterRequestHandlerProxy) ChildrenStateUpdate(args []*ic.Argument) (*empty.Empty, error) {
+ if len(args) < 3 {
+ logger.Warn("invalid-number-of-args", log.Fields{"args": args})
+ err := errors.New("invalid-number-of-args")
+ return nil, err
+ }
+ deviceID := &voltha.ID{}
+ operStatus := &ic.IntType{}
+ connStatus := &ic.IntType{}
+ transactionID := &ic.StrType{}
+ for _, arg := range args {
+ switch arg.Key {
+ case "device_id":
+ if err := ptypes.UnmarshalAny(arg.Value, deviceID); err != nil {
+ logger.Warnw("cannot-unmarshal-device-id", log.Fields{"error": err})
+ return nil, err
+ }
+ case "oper_status":
+ if err := ptypes.UnmarshalAny(arg.Value, operStatus); err != nil {
+ logger.Warnw("cannot-unmarshal-operStatus", log.Fields{"error": err})
+ return nil, err
+ }
+ case "connect_status":
+ if err := ptypes.UnmarshalAny(arg.Value, connStatus); err != nil {
+ logger.Warnw("cannot-unmarshal-connStatus", log.Fields{"error": err})
+ return nil, err
+ }
+ case kafka.TransactionKey:
+ if err := ptypes.UnmarshalAny(arg.Value, transactionID); err != nil {
+ logger.Warnw("cannot-unmarshal-transaction-ID", log.Fields{"error": err})
+ return nil, err
+ }
+ }
+ }
+ logger.Debugw("ChildrenStateUpdate", log.Fields{"deviceID": deviceID.Id, "oper-status": operStatus,
+ "conn-status": connStatus, "transactionID": transactionID.Val})
+
+ // When the enum is not set (i.e. -1), Go still convert to the Enum type with the value being -1
+ if err := rhp.deviceMgr.UpdateChildrenStatus(context.TODO(), deviceID.Id, voltha.OperStatus_Types(operStatus.Val),
+ voltha.ConnectStatus_Types(connStatus.Val)); err != nil {
+ logger.Debugw("unable-to-update-children-status", log.Fields{"error": err})
+ return nil, err
+ }
+ return &empty.Empty{}, nil
+}
+
+// PortsStateUpdate updates the ports state related to the device
+func (rhp *AdapterRequestHandlerProxy) PortsStateUpdate(args []*ic.Argument) (*empty.Empty, error) {
+ if len(args) < 2 {
+ logger.Warn("invalid-number-of-args", log.Fields{"args": args})
+ err := errors.New("invalid-number-of-args")
+ return nil, err
+ }
+ deviceID := &voltha.ID{}
+ operStatus := &ic.IntType{}
+ transactionID := &ic.StrType{}
+ for _, arg := range args {
+ switch arg.Key {
+ case "device_id":
+ if err := ptypes.UnmarshalAny(arg.Value, deviceID); err != nil {
+ logger.Warnw("cannot-unmarshal-device-id", log.Fields{"error": err})
+ return nil, err
+ }
+ case "oper_status":
+ if err := ptypes.UnmarshalAny(arg.Value, operStatus); err != nil {
+ logger.Warnw("cannot-unmarshal-operStatus", log.Fields{"error": err})
+ return nil, err
+ }
+ case kafka.TransactionKey:
+ if err := ptypes.UnmarshalAny(arg.Value, transactionID); err != nil {
+ logger.Warnw("cannot-unmarshal-transaction-ID", log.Fields{"error": err})
+ return nil, err
+ }
+ }
+ }
+ logger.Debugw("PortsStateUpdate", log.Fields{"deviceID": deviceID.Id, "operStatus": operStatus, "transactionID": transactionID.Val})
+
+ if err := rhp.deviceMgr.UpdatePortsState(context.TODO(), deviceID.Id, voltha.OperStatus_Types(operStatus.Val)); err != nil {
+ logger.Debugw("unable-to-update-ports-state", log.Fields{"error": err})
+ return nil, err
+ }
+ return &empty.Empty{}, nil
+}
+
+// PortStateUpdate updates the port state of the device
+func (rhp *AdapterRequestHandlerProxy) PortStateUpdate(args []*ic.Argument) (*empty.Empty, error) {
+ if len(args) < 3 {
+ logger.Warn("invalid-number-of-args", log.Fields{"args": args})
+ err := errors.New("invalid-number-of-args")
+ return nil, err
+ }
+ deviceID := &voltha.ID{}
+ portType := &ic.IntType{}
+ portNo := &ic.IntType{}
+ operStatus := &ic.IntType{}
+ transactionID := &ic.StrType{}
+ for _, arg := range args {
+ switch arg.Key {
+ case "device_id":
+ if err := ptypes.UnmarshalAny(arg.Value, deviceID); err != nil {
+ logger.Warnw("cannot-unmarshal-device-id", log.Fields{"error": err})
+ return nil, err
+ }
+ case "oper_status":
+ if err := ptypes.UnmarshalAny(arg.Value, operStatus); err != nil {
+ logger.Warnw("cannot-unmarshal-operStatus", log.Fields{"error": err})
+ return nil, err
+ }
+ case "port_type":
+ if err := ptypes.UnmarshalAny(arg.Value, portType); err != nil {
+ logger.Warnw("cannot-unmarshal-porttype", log.Fields{"error": err})
+ return nil, err
+ }
+ case "port_no":
+ if err := ptypes.UnmarshalAny(arg.Value, portNo); err != nil {
+ logger.Warnw("cannot-unmarshal-portno", log.Fields{"error": err})
+ return nil, err
+ }
+ case kafka.TransactionKey:
+ if err := ptypes.UnmarshalAny(arg.Value, transactionID); err != nil {
+ logger.Warnw("cannot-unmarshal-transaction-ID", log.Fields{"error": err})
+ return nil, err
+ }
+ }
+ }
+ logger.Debugw("PortStateUpdate", log.Fields{"deviceID": deviceID.Id, "operStatus": operStatus,
+ "portType": portType, "portNo": portNo, "transactionID": transactionID.Val})
+
+ if err := rhp.deviceMgr.UpdatePortState(context.TODO(), deviceID.Id, voltha.Port_PortType(portType.Val), uint32(portNo.Val),
+ voltha.OperStatus_Types(operStatus.Val)); err != nil {
+ // If the error doesn't change behavior and is essentially ignored, it is not an error, it is a
+ // warning.
+ // TODO: VOL-2707
+ logger.Debugw("unable-to-update-port-state", log.Fields{"error": err})
+ return nil, err
+ }
+ return &empty.Empty{}, nil
+}
+
+// DeleteAllPorts deletes all ports of device
+func (rhp *AdapterRequestHandlerProxy) DeleteAllPorts(args []*ic.Argument) (*empty.Empty, error) {
+ if len(args) < 3 {
+ logger.Warn("invalid-number-of-args", log.Fields{"args": args})
+ err := errors.New("invalid-number-of-args")
+ return nil, err
+ }
+ deviceID := &voltha.ID{}
+ transactionID := &ic.StrType{}
+ for _, arg := range args {
+ switch arg.Key {
+ case "device_id":
+ if err := ptypes.UnmarshalAny(arg.Value, deviceID); err != nil {
+ logger.Warnw("cannot-unmarshal-device-id", log.Fields{"error": err})
+ return nil, err
+ }
+ case kafka.TransactionKey:
+ if err := ptypes.UnmarshalAny(arg.Value, transactionID); err != nil {
+ logger.Warnw("cannot-unmarshal-transaction-ID", log.Fields{"error": err})
+ return nil, err
+ }
+ }
+ }
+ logger.Debugw("DeleteAllPorts", log.Fields{"deviceID": deviceID.Id, "transactionID": transactionID.Val})
+
+ if err := rhp.deviceMgr.DeleteAllPorts(context.TODO(), deviceID.Id); err != nil {
+ logger.Debugw("unable-to-delete-ports", log.Fields{"error": err})
+ return nil, err
+ }
+ return &empty.Empty{}, nil
+}
+
+// ChildDevicesLost indicates that a parent device is in a state (Disabled) where it cannot manage the child devices.
+// This will trigger the Core to disable all the child devices.
+func (rhp *AdapterRequestHandlerProxy) ChildDevicesLost(args []*ic.Argument) (*empty.Empty, error) {
+ if len(args) < 2 {
+ logger.Warn("invalid-number-of-args", log.Fields{"args": args})
+ err := errors.New("invalid-number-of-args")
+ return nil, err
+ }
+ parentDeviceID := &voltha.ID{}
+ transactionID := &ic.StrType{}
+ for _, arg := range args {
+ switch arg.Key {
+ case "parent_device_id":
+ if err := ptypes.UnmarshalAny(arg.Value, parentDeviceID); err != nil {
+ logger.Warnw("cannot-unmarshal-device-id", log.Fields{"error": err})
+ return nil, err
+ }
+ case kafka.TransactionKey:
+ if err := ptypes.UnmarshalAny(arg.Value, transactionID); err != nil {
+ logger.Warnw("cannot-unmarshal-transaction-ID", log.Fields{"error": err})
+ return nil, err
+ }
+ }
+ }
+ logger.Debugw("ChildDevicesLost", log.Fields{"deviceID": parentDeviceID.Id, "transactionID": transactionID.Val})
+
+ if err := rhp.deviceMgr.ChildDevicesLost(context.TODO(), parentDeviceID.Id); err != nil {
+ logger.Debugw("unable-to-disable-child-devices", log.Fields{"error": err})
+ return nil, err
+ }
+ return &empty.Empty{}, nil
+}
+
+// ChildDevicesDetected invoked by an adapter when child devices are found, typically after after a disable/enable sequence.
+// This will trigger the Core to Enable all the child devices of that parent.
+func (rhp *AdapterRequestHandlerProxy) ChildDevicesDetected(args []*ic.Argument) (*empty.Empty, error) {
+ if len(args) < 2 {
+ logger.Warn("invalid-number-of-args", log.Fields{"args": args})
+ err := errors.New("invalid-number-of-args")
+ return nil, err
+ }
+ parentDeviceID := &voltha.ID{}
+ transactionID := &ic.StrType{}
+ for _, arg := range args {
+ switch arg.Key {
+ case "parent_device_id":
+ if err := ptypes.UnmarshalAny(arg.Value, parentDeviceID); err != nil {
+ logger.Warnw("cannot-unmarshal-device-id", log.Fields{"error": err})
+ return nil, err
+ }
+ case kafka.TransactionKey:
+ if err := ptypes.UnmarshalAny(arg.Value, transactionID); err != nil {
+ logger.Warnw("cannot-unmarshal-transaction-ID", log.Fields{"error": err})
+ return nil, err
+ }
+ }
+ }
+ logger.Debugw("ChildDevicesDetected", log.Fields{"deviceID": parentDeviceID.Id, "transactionID": transactionID.Val})
+
+ if err := rhp.deviceMgr.ChildDevicesDetected(context.TODO(), parentDeviceID.Id); err != nil {
+ logger.Debugw("child-devices-detection-failed", log.Fields{"parentID": parentDeviceID.Id, "error": err})
+ return nil, err
+ }
+ return &empty.Empty{}, nil
+}
+
+// PortCreated adds port to device
+func (rhp *AdapterRequestHandlerProxy) PortCreated(args []*ic.Argument) (*empty.Empty, error) {
+ if len(args) < 3 {
+ logger.Warn("invalid-number-of-args", log.Fields{"args": args})
+ err := errors.New("invalid-number-of-args")
+ return nil, err
+ }
+ deviceID := &voltha.ID{}
+ port := &voltha.Port{}
+ transactionID := &ic.StrType{}
+ for _, arg := range args {
+ switch arg.Key {
+ case "device_id":
+ if err := ptypes.UnmarshalAny(arg.Value, deviceID); err != nil {
+ logger.Warnw("cannot-unmarshal-device-id", log.Fields{"error": err})
+ return nil, err
+ }
+ case "port":
+ if err := ptypes.UnmarshalAny(arg.Value, port); err != nil {
+ logger.Warnw("cannot-unmarshal-port", log.Fields{"error": err})
+ return nil, err
+ }
+ case kafka.TransactionKey:
+ if err := ptypes.UnmarshalAny(arg.Value, transactionID); err != nil {
+ logger.Warnw("cannot-unmarshal-transaction-ID", log.Fields{"error": err})
+ return nil, err
+ }
+ }
+ }
+ logger.Debugw("PortCreated", log.Fields{"deviceID": deviceID.Id, "port": port, "transactionID": transactionID.Val})
+
+ if err := rhp.deviceMgr.AddPort(context.TODO(), deviceID.Id, port); err != nil {
+ logger.Debugw("unable-to-add-port", log.Fields{"error": err})
+ return nil, err
+ }
+ return &empty.Empty{}, nil
+}
+
+// DevicePMConfigUpdate initializes the pm configs as defined by the adapter.
+func (rhp *AdapterRequestHandlerProxy) DevicePMConfigUpdate(args []*ic.Argument) (*empty.Empty, error) {
+ if len(args) < 2 {
+ logger.Warn("invalid-number-of-args", log.Fields{"args": args})
+ err := errors.New("invalid-number-of-args")
+ return nil, err
+ }
+ pmConfigs := &voltha.PmConfigs{}
+ transactionID := &ic.StrType{}
+ for _, arg := range args {
+ switch arg.Key {
+ case "device_pm_config":
+ if err := ptypes.UnmarshalAny(arg.Value, pmConfigs); err != nil {
+ logger.Warnw("cannot-unmarshal-pm-config", log.Fields{"error": err})
+ return nil, err
+ }
+ case kafka.TransactionKey:
+ if err := ptypes.UnmarshalAny(arg.Value, transactionID); err != nil {
+ logger.Warnw("cannot-unmarshal-transaction-ID", log.Fields{"error": err})
+ return nil, err
+ }
+ }
+ }
+ logger.Debugw("DevicePMConfigUpdate", log.Fields{"deviceID": pmConfigs.Id, "configs": pmConfigs,
+ "transactionID": transactionID.Val})
+
+ if err := rhp.deviceMgr.InitPmConfigs(context.TODO(), pmConfigs.Id, pmConfigs); err != nil {
+ logger.Debugw("unable-to-initialize-pm-configs", log.Fields{"error": err})
+ return nil, err
+ }
+ return &empty.Empty{}, nil
+}
+
+// PacketIn sends the incoming packet of device
+func (rhp *AdapterRequestHandlerProxy) PacketIn(args []*ic.Argument) (*empty.Empty, error) {
+ if len(args) < 4 {
+ logger.Warn("invalid-number-of-args", log.Fields{"args": args})
+ err := errors.New("invalid-number-of-args")
+ return nil, err
+ }
+ deviceID := &voltha.ID{}
+ portNo := &ic.IntType{}
+ packet := &ic.Packet{}
+ transactionID := &ic.StrType{}
+ for _, arg := range args {
+ switch arg.Key {
+ case "device_id":
+ if err := ptypes.UnmarshalAny(arg.Value, deviceID); err != nil {
+ logger.Warnw("cannot-unmarshal-device-id", log.Fields{"error": err})
+ return nil, err
+ }
+ case "port":
+ if err := ptypes.UnmarshalAny(arg.Value, portNo); err != nil {
+ logger.Warnw("cannot-unmarshal-port-no", log.Fields{"error": err})
+ return nil, err
+ }
+ case "packet":
+ if err := ptypes.UnmarshalAny(arg.Value, packet); err != nil {
+ logger.Warnw("cannot-unmarshal-packet", log.Fields{"error": err})
+ return nil, err
+ }
+ case kafka.TransactionKey:
+ if err := ptypes.UnmarshalAny(arg.Value, transactionID); err != nil {
+ logger.Warnw("cannot-unmarshal-transaction-ID", log.Fields{"error": err})
+ return nil, err
+ }
+ }
+ }
+ logger.Debugw("PacketIn", log.Fields{"deviceID": deviceID.Id, "port": portNo.Val, "packet": packet,
+ "transactionID": transactionID.Val})
+
+ if err := rhp.deviceMgr.PacketIn(context.TODO(), deviceID.Id, uint32(portNo.Val), transactionID.Val, packet.Payload); err != nil {
+ logger.Debugw("unable-to-receive-packet-from-adapter", log.Fields{"error": err})
+ return nil, err
+
+ }
+ return &empty.Empty{}, nil
+}
+
+// UpdateImageDownload updates image download
+func (rhp *AdapterRequestHandlerProxy) UpdateImageDownload(args []*ic.Argument) (*empty.Empty, error) {
+ if len(args) < 2 {
+ logger.Warn("invalid-number-of-args", log.Fields{"args": args})
+ err := errors.New("invalid-number-of-args")
+ return nil, err
+ }
+ deviceID := &voltha.ID{}
+ img := &voltha.ImageDownload{}
+ transactionID := &ic.StrType{}
+ for _, arg := range args {
+ switch arg.Key {
+ case "device_id":
+ if err := ptypes.UnmarshalAny(arg.Value, deviceID); err != nil {
+ logger.Warnw("cannot-unmarshal-device-id", log.Fields{"error": err})
+ return nil, err
+ }
+ case "image_download":
+ if err := ptypes.UnmarshalAny(arg.Value, img); err != nil {
+ logger.Warnw("cannot-unmarshal-imgaeDownload", log.Fields{"error": err})
+ return nil, err
+ }
+ case kafka.TransactionKey:
+ if err := ptypes.UnmarshalAny(arg.Value, transactionID); err != nil {
+ logger.Warnw("cannot-unmarshal-transaction-ID", log.Fields{"error": err})
+ return nil, err
+ }
+ }
+ }
+ logger.Debugw("UpdateImageDownload", log.Fields{"deviceID": deviceID.Id, "image-download": img,
+ "transactionID": transactionID.Val})
+
+ if err := rhp.deviceMgr.UpdateImageDownload(context.TODO(), deviceID.Id, img); err != nil {
+ logger.Debugw("unable-to-update-image-download", log.Fields{"error": err})
+ return nil, err
+ }
+ return &empty.Empty{}, nil
+}
+
+// ReconcileChildDevices reconciles child devices
+func (rhp *AdapterRequestHandlerProxy) ReconcileChildDevices(args []*ic.Argument) (*empty.Empty, error) {
+ if len(args) < 2 {
+ logger.Warn("invalid-number-of-args", log.Fields{"args": args})
+ err := errors.New("invalid-number-of-args")
+ return nil, err
+ }
+ parentDeviceID := &voltha.ID{}
+ transactionID := &ic.StrType{}
+ for _, arg := range args {
+ switch arg.Key {
+ case "parent_device_id":
+ if err := ptypes.UnmarshalAny(arg.Value, parentDeviceID); err != nil {
+ logger.Warnw("cannot-unmarshal-device-id", log.Fields{"error": err})
+ return nil, err
+ }
+ case kafka.TransactionKey:
+ if err := ptypes.UnmarshalAny(arg.Value, transactionID); err != nil {
+ logger.Warnw("cannot-unmarshal-transaction-ID", log.Fields{"error": err})
+ return nil, err
+ }
+ }
+ }
+ logger.Debugw("ReconcileChildDevices", log.Fields{"deviceID": parentDeviceID.Id, "transactionID": transactionID.Val})
+
+ if err := rhp.deviceMgr.ReconcileChildDevices(context.TODO(), parentDeviceID.Id); err != nil {
+ logger.Debugw("unable-to-reconcile-child-devices", log.Fields{"error": err})
+ return nil, err
+ }
+ return &empty.Empty{}, nil
+}
+
+// DeviceReasonUpdate updates device reason
+func (rhp *AdapterRequestHandlerProxy) DeviceReasonUpdate(args []*ic.Argument) (*empty.Empty, error) {
+ if len(args) < 2 {
+ logger.Warn("DeviceReasonUpdate: invalid-number-of-args", log.Fields{"args": args})
+ err := errors.New("DeviceReasonUpdate: invalid-number-of-args")
+ return nil, err
+ }
+ deviceID := &voltha.ID{}
+ reason := &ic.StrType{}
+ transactionID := &ic.StrType{}
+ for _, arg := range args {
+ switch arg.Key {
+ case "device_id":
+ if err := ptypes.UnmarshalAny(arg.Value, deviceID); err != nil {
+ logger.Warnw("cannot-unmarshal-device-id", log.Fields{"error": err})
+ return nil, err
+ }
+ case "device_reason":
+ if err := ptypes.UnmarshalAny(arg.Value, reason); err != nil {
+ logger.Warnw("cannot-unmarshal-reason", log.Fields{"error": err})
+ return nil, err
+ }
+ case kafka.TransactionKey:
+ if err := ptypes.UnmarshalAny(arg.Value, transactionID); err != nil {
+ logger.Warnw("cannot-unmarshal-transaction-ID", log.Fields{"error": err})
+ return nil, err
+ }
+ }
+ }
+ logger.Debugw("DeviceReasonUpdate", log.Fields{"deviceId": deviceID.Id, "reason": reason.Val,
+ "transactionID": transactionID.Val})
+
+ if err := rhp.deviceMgr.UpdateDeviceReason(context.TODO(), deviceID.Id, reason.Val); err != nil {
+ logger.Debugw("unable-to-update-device-reason", log.Fields{"error": err})
+ return nil, err
+
+ }
+ return &empty.Empty{}, nil
+}
diff --git a/rw_core/core/api/common.go b/rw_core/core/api/common.go
new file mode 100644
index 0000000..fd2910a
--- /dev/null
+++ b/rw_core/core/api/common.go
@@ -0,0 +1,33 @@
+/*
+ * Copyright 2020-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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// Package core Common Logger initialization
+package api
+
+import (
+ "github.com/opencord/voltha-lib-go/v3/pkg/log"
+)
+
+var logger log.Logger
+
+func init() {
+ // Setup this package so that it's log level can be modified at run time
+ var err error
+ logger, err = log.AddPackage(log.JSON, log.ErrorLevel, log.Fields{"pkg": "nbi"})
+ if err != nil {
+ panic(err)
+ }
+}
diff --git a/rw_core/core/api/common_test.go b/rw_core/core/api/common_test.go
new file mode 100644
index 0000000..f6f07f5
--- /dev/null
+++ b/rw_core/core/api/common_test.go
@@ -0,0 +1,291 @@
+/*
+ * Copyright 2019-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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package api
+
+import (
+ "context"
+ "fmt"
+ "strconv"
+ "time"
+
+ "github.com/golang/protobuf/ptypes/empty"
+ "github.com/google/uuid"
+ "github.com/opencord/voltha-go/rw_core/config"
+ cm "github.com/opencord/voltha-go/rw_core/mocks"
+ "github.com/opencord/voltha-lib-go/v3/pkg/adapters"
+ com "github.com/opencord/voltha-lib-go/v3/pkg/adapters/common"
+ "github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore"
+ "github.com/opencord/voltha-lib-go/v3/pkg/kafka"
+ "github.com/opencord/voltha-lib-go/v3/pkg/log"
+ mock_etcd "github.com/opencord/voltha-lib-go/v3/pkg/mocks/etcd"
+ "github.com/opencord/voltha-protos/v3/go/voltha"
+ "github.com/phayes/freeport"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/metadata"
+ "google.golang.org/grpc/status"
+)
+
+const (
+ volthaSerialNumberKey = "voltha_serial_number"
+ retryInterval = 50 * time.Millisecond
+)
+
+const (
+ OltAdapter = iota
+ OnuAdapter
+)
+
+var (
+ coreInCompeteMode bool
+)
+
+type isLogicalDeviceConditionSatisfied func(ld *voltha.LogicalDevice) bool
+type isDeviceConditionSatisfied func(ld *voltha.Device) bool
+type isDevicesConditionSatisfied func(ds *voltha.Devices) bool
+type isLogicalDevicesConditionSatisfied func(lds *voltha.LogicalDevices) bool
+type isConditionSatisfied func() bool
+
+func init() {
+ //Default mode is two rw-core running in a pair of competing cores
+ coreInCompeteMode = true
+}
+
+func setCoreCompeteMode(mode bool) {
+ coreInCompeteMode = mode
+}
+
+func getContext() context.Context {
+ if coreInCompeteMode {
+ return metadata.NewIncomingContext(context.Background(), metadata.Pairs(volthaSerialNumberKey, uuid.New().String()))
+ }
+ return context.Background()
+}
+
+//startEmbeddedEtcdServer creates and starts an Embedded etcd server locally.
+func startEmbeddedEtcdServer(configName, storageDir, logLevel string) (*mock_etcd.EtcdServer, int, error) {
+ kvClientPort, err := freeport.GetFreePort()
+ if err != nil {
+ return nil, 0, err
+ }
+ peerPort, err := freeport.GetFreePort()
+ if err != nil {
+ return nil, 0, err
+ }
+ etcdServer := mock_etcd.StartEtcdServer(mock_etcd.MKConfig(configName, kvClientPort, peerPort, storageDir, logLevel))
+ if etcdServer == nil {
+ return nil, 0, status.Error(codes.Internal, "Embedded server failed to start")
+ }
+ return etcdServer, kvClientPort, nil
+}
+
+func stopEmbeddedEtcdServer(server *mock_etcd.EtcdServer) {
+ if server != nil {
+ server.Stop()
+ }
+}
+
+func setupKVClient(cf *config.RWCoreFlags, coreInstanceID string) kvstore.Client {
+ addr := cf.KVStoreHost + ":" + strconv.Itoa(cf.KVStorePort)
+ client, err := kvstore.NewEtcdClient(addr, cf.KVStoreTimeout)
+ if err != nil {
+ panic("no kv client")
+ }
+ return client
+}
+
+func createMockAdapter(adapterType int, kafkaClient kafka.Client, coreInstanceID string, coreName string, adapterName string) (adapters.IAdapter, error) {
+ var err error
+ var adapter adapters.IAdapter
+ adapterKafkaICProxy := kafka.NewInterContainerProxy(
+ kafka.MsgClient(kafkaClient),
+ kafka.DefaultTopic(&kafka.Topic{Name: adapterName}))
+ adapterCoreProxy := com.NewCoreProxy(adapterKafkaICProxy, adapterName, coreName)
+ var adapterReqHandler *com.RequestHandlerProxy
+ switch adapterType {
+ case OltAdapter:
+ adapter = cm.NewOLTAdapter(adapterCoreProxy)
+ case OnuAdapter:
+ adapter = cm.NewONUAdapter(adapterCoreProxy)
+ default:
+ logger.Fatalf("invalid-adapter-type-%d", adapterType)
+ }
+ adapterReqHandler = com.NewRequestHandlerProxy(coreInstanceID, adapter, adapterCoreProxy)
+
+ if err = adapterKafkaICProxy.Start(); err != nil {
+ logger.Errorw("Failure-starting-adapter-intercontainerProxy", log.Fields{"error": err})
+ return nil, err
+ }
+ if err = adapterKafkaICProxy.SubscribeWithRequestHandlerInterface(kafka.Topic{Name: adapterName}, adapterReqHandler); err != nil {
+ logger.Errorw("Failure-to-subscribe-onu-request-handler", log.Fields{"error": err})
+ return nil, err
+ }
+ return adapter, nil
+}
+
+func waitUntilDeviceReadiness(deviceID string,
+ timeout time.Duration,
+ verificationFunction isDeviceConditionSatisfied,
+ nbi *NBIHandler) error {
+ ch := make(chan int, 1)
+ done := false
+ go func() {
+ for {
+ device, _ := nbi.GetDevice(getContext(), &voltha.ID{Id: deviceID})
+ if verificationFunction(device) {
+ ch <- 1
+ break
+ }
+ if done {
+ break
+ }
+ time.Sleep(retryInterval)
+ }
+ }()
+ timer := time.NewTimer(timeout)
+ defer timer.Stop()
+ select {
+ case <-ch:
+ return nil
+ case <-timer.C:
+ done = true
+ return fmt.Errorf("expected-states-not-reached-for-device%s", deviceID)
+ }
+}
+
+func waitUntilLogicalDeviceReadiness(oltDeviceID string,
+ timeout time.Duration,
+ nbi *NBIHandler,
+ verificationFunction isLogicalDeviceConditionSatisfied,
+) error {
+ ch := make(chan int, 1)
+ done := false
+ go func() {
+ for {
+ // Get the logical device from the olt device
+ d, _ := nbi.GetDevice(getContext(), &voltha.ID{Id: oltDeviceID})
+ if d != nil && d.ParentId != "" {
+ ld, _ := nbi.GetLogicalDevice(getContext(), &voltha.ID{Id: d.ParentId})
+ if verificationFunction(ld) {
+ ch <- 1
+ break
+ }
+ if done {
+ break
+ }
+ } else if d != nil && d.ParentId == "" { // case where logical device deleted
+ if verificationFunction(nil) {
+ ch <- 1
+ break
+ }
+ if done {
+ break
+ }
+ }
+ time.Sleep(retryInterval)
+ }
+ }()
+ timer := time.NewTimer(timeout)
+ defer timer.Stop()
+ select {
+ case <-ch:
+ return nil
+ case <-timer.C:
+ done = true
+ return fmt.Errorf("timeout-waiting-for-logical-device-readiness%s", oltDeviceID)
+ }
+}
+
+func waitUntilConditionForDevices(timeout time.Duration, nbi *NBIHandler, verificationFunction isDevicesConditionSatisfied) error {
+ ch := make(chan int, 1)
+ done := false
+ go func() {
+ for {
+ devices, _ := nbi.ListDevices(getContext(), &empty.Empty{})
+ if verificationFunction(devices) {
+ ch <- 1
+ break
+ }
+ if done {
+ break
+ }
+
+ time.Sleep(retryInterval)
+ }
+ }()
+ timer := time.NewTimer(timeout)
+ defer timer.Stop()
+ select {
+ case <-ch:
+ return nil
+ case <-timer.C:
+ done = true
+ return fmt.Errorf("timeout-waiting-devices")
+ }
+}
+
+func waitUntilConditionForLogicalDevices(timeout time.Duration, nbi *NBIHandler, verificationFunction isLogicalDevicesConditionSatisfied) error {
+ ch := make(chan int, 1)
+ done := false
+ go func() {
+ for {
+ lDevices, _ := nbi.ListLogicalDevices(getContext(), &empty.Empty{})
+ if verificationFunction(lDevices) {
+ ch <- 1
+ break
+ }
+ if done {
+ break
+ }
+
+ time.Sleep(retryInterval)
+ }
+ }()
+ timer := time.NewTimer(timeout)
+ defer timer.Stop()
+ select {
+ case <-ch:
+ return nil
+ case <-timer.C:
+ done = true
+ return fmt.Errorf("timeout-waiting-logical-devices")
+ }
+}
+
+func waitUntilCondition(timeout time.Duration, nbi *NBIHandler, verificationFunction isConditionSatisfied) error {
+ ch := make(chan int, 1)
+ done := false
+ go func() {
+ for {
+ if verificationFunction() {
+ ch <- 1
+ break
+ }
+ if done {
+ break
+ }
+ time.Sleep(retryInterval)
+ }
+ }()
+ timer := time.NewTimer(timeout)
+ defer timer.Stop()
+ select {
+ case <-ch:
+ return nil
+ case <-timer.C:
+ done = true
+ return fmt.Errorf("timeout-waiting-for-condition")
+ }
+}
diff --git a/rw_core/core/api/grpc_nbi_handler.go b/rw_core/core/api/grpc_nbi_handler.go
new file mode 100755
index 0000000..1c296dd
--- /dev/null
+++ b/rw_core/core/api/grpc_nbi_handler.go
@@ -0,0 +1,807 @@
+/*
+ * 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
+
+ * http://www.apache.org/licenses/LICENSE-2.0
+
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package api
+
+import (
+ "context"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "github.com/golang/protobuf/ptypes/empty"
+ da "github.com/opencord/voltha-go/common/core/northbound/grpc"
+ "github.com/opencord/voltha-go/rw_core/core/adapter"
+ "github.com/opencord/voltha-go/rw_core/core/device"
+ "github.com/opencord/voltha-lib-go/v3/pkg/log"
+ "github.com/opencord/voltha-lib-go/v3/pkg/version"
+ "github.com/opencord/voltha-protos/v3/go/common"
+ "github.com/opencord/voltha-protos/v3/go/omci"
+ "github.com/opencord/voltha-protos/v3/go/openflow_13"
+ "github.com/opencord/voltha-protos/v3/go/voltha"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/status"
+ "io"
+ "sync"
+)
+
+// Image related constants
+const (
+ ImageDownload = iota
+ CancelImageDownload = iota
+ ActivateImage = iota
+ RevertImage = iota
+)
+
+// NBIHandler represent attributes of API handler
+type NBIHandler struct {
+ deviceMgr *device.Manager
+ logicalDeviceMgr *device.LogicalManager
+ adapterMgr *adapter.Manager
+ packetInQueue chan openflow_13.PacketIn
+ changeEventQueue chan openflow_13.ChangeEvent
+ packetInQueueDone chan bool
+ changeEventQueueDone chan bool
+ da.DefaultAPIHandler
+}
+
+// NewAPIHandler creates API handler instance
+func NewAPIHandler(deviceMgr *device.Manager, logicalDeviceMgr *device.LogicalManager, adapterMgr *adapter.Manager) *NBIHandler {
+ return &NBIHandler{
+ deviceMgr: deviceMgr,
+ logicalDeviceMgr: logicalDeviceMgr,
+ adapterMgr: adapterMgr,
+ packetInQueue: make(chan openflow_13.PacketIn, 100),
+ changeEventQueue: make(chan openflow_13.ChangeEvent, 100),
+ packetInQueueDone: make(chan bool, 1),
+ changeEventQueueDone: make(chan bool, 1),
+ }
+}
+
+// waitForNilResponseOnSuccess is a helper function to wait for a response on channel monitorCh where an nil
+// response is expected in a successful scenario
+func waitForNilResponseOnSuccess(ctx context.Context, ch chan interface{}) (*empty.Empty, error) {
+ select {
+ case res := <-ch:
+ if res == nil {
+ return &empty.Empty{}, nil
+ } else if err, ok := res.(error); ok {
+ return &empty.Empty{}, err
+ } else {
+ logger.Warnw("unexpected-return-type", log.Fields{"result": res})
+ err = status.Errorf(codes.Internal, "%s", res)
+ return &empty.Empty{}, err
+ }
+ case <-ctx.Done():
+ logger.Debug("client-timeout")
+ return nil, ctx.Err()
+ }
+}
+
+// ListCoreInstances returns details on the running core containers
+func (handler *NBIHandler) ListCoreInstances(ctx context.Context, empty *empty.Empty) (*voltha.CoreInstances, error) {
+ logger.Debug("ListCoreInstances")
+ // TODO: unused stub
+ return &voltha.CoreInstances{}, status.Errorf(codes.NotFound, "no-core-instances")
+}
+
+// GetCoreInstance returns the details of a specific core container
+func (handler *NBIHandler) GetCoreInstance(ctx context.Context, id *voltha.ID) (*voltha.CoreInstance, error) {
+ logger.Debugw("GetCoreInstance", log.Fields{"id": id})
+ //TODO: unused stub
+ return &voltha.CoreInstance{}, status.Errorf(codes.NotFound, "core-instance-%s", id.Id)
+}
+
+// GetLogicalDevicePort returns logical device port details
+func (handler *NBIHandler) GetLogicalDevicePort(ctx context.Context, id *voltha.LogicalPortId) (*voltha.LogicalPort, error) {
+ logger.Debugw("GetLogicalDevicePort-request", log.Fields{"id": *id})
+
+ return handler.logicalDeviceMgr.GetLogicalPort(ctx, id)
+}
+
+// EnableLogicalDevicePort enables logical device port
+func (handler *NBIHandler) EnableLogicalDevicePort(ctx context.Context, id *voltha.LogicalPortId) (*empty.Empty, error) {
+ logger.Debugw("EnableLogicalDevicePort-request", log.Fields{"id": id, "test": common.TestModeKeys_api_test.String()})
+
+ ch := make(chan interface{})
+ defer close(ch)
+ go handler.logicalDeviceMgr.EnableLogicalPort(ctx, id, ch)
+ return waitForNilResponseOnSuccess(ctx, ch)
+}
+
+// DisableLogicalDevicePort disables logical device port
+func (handler *NBIHandler) DisableLogicalDevicePort(ctx context.Context, id *voltha.LogicalPortId) (*empty.Empty, error) {
+ logger.Debugw("DisableLogicalDevicePort-request", log.Fields{"id": id, "test": common.TestModeKeys_api_test.String()})
+
+ ch := make(chan interface{})
+ defer close(ch)
+ go handler.logicalDeviceMgr.DisableLogicalPort(ctx, id, ch)
+ return waitForNilResponseOnSuccess(ctx, ch)
+}
+
+// UpdateLogicalDeviceFlowTable updates logical device flow table
+func (handler *NBIHandler) UpdateLogicalDeviceFlowTable(ctx context.Context, flow *openflow_13.FlowTableUpdate) (*empty.Empty, error) {
+ logger.Debugw("UpdateLogicalDeviceFlowTable-request", log.Fields{"flow": flow, "test": common.TestModeKeys_api_test.String()})
+
+ ch := make(chan interface{})
+ defer close(ch)
+ go handler.logicalDeviceMgr.UpdateFlowTable(ctx, flow.Id, flow.FlowMod, ch)
+ return waitForNilResponseOnSuccess(ctx, ch)
+}
+
+// UpdateLogicalDeviceFlowGroupTable updates logical device flow group table
+func (handler *NBIHandler) UpdateLogicalDeviceFlowGroupTable(ctx context.Context, flow *openflow_13.FlowGroupTableUpdate) (*empty.Empty, error) {
+ logger.Debugw("UpdateLogicalDeviceFlowGroupTable-request", log.Fields{"flow": flow, "test": common.TestModeKeys_api_test.String()})
+ ch := make(chan interface{})
+ defer close(ch)
+ go handler.logicalDeviceMgr.UpdateGroupTable(ctx, flow.Id, flow.GroupMod, ch)
+ return waitForNilResponseOnSuccess(ctx, ch)
+}
+
+// GetDevice must be implemented in the read-only containers - should it also be implemented here?
+func (handler *NBIHandler) GetDevice(ctx context.Context, id *voltha.ID) (*voltha.Device, error) {
+ logger.Debugw("GetDevice-request", log.Fields{"id": id})
+ return handler.deviceMgr.GetDevice(ctx, id.Id)
+}
+
+// GetDevice must be implemented in the read-only containers - should it also be implemented here?
+
+// ListDevices retrieves the latest devices from the data model
+func (handler *NBIHandler) ListDevices(ctx context.Context, empty *empty.Empty) (*voltha.Devices, error) {
+ logger.Debug("ListDevices")
+ devices, err := handler.deviceMgr.ListDevices(ctx)
+ if err != nil {
+ logger.Errorw("Failed to list devices", log.Fields{"error": err})
+ return nil, err
+ }
+ return devices, nil
+}
+
+// ListDeviceIds returns the list of device ids managed by a voltha core
+func (handler *NBIHandler) ListDeviceIds(ctx context.Context, empty *empty.Empty) (*voltha.IDs, error) {
+ logger.Debug("ListDeviceIDs")
+ return handler.deviceMgr.ListDeviceIds()
+}
+
+//ReconcileDevices is a request to a voltha core to managed a list of devices based on their IDs
+func (handler *NBIHandler) ReconcileDevices(ctx context.Context, ids *voltha.IDs) (*empty.Empty, error) {
+ logger.Debug("ReconcileDevices")
+
+ ch := make(chan interface{})
+ defer close(ch)
+ go handler.deviceMgr.ReconcileDevices(ctx, ids, ch)
+ return waitForNilResponseOnSuccess(ctx, ch)
+}
+
+// GetLogicalDevice provides a cloned most up to date logical device
+func (handler *NBIHandler) GetLogicalDevice(ctx context.Context, id *voltha.ID) (*voltha.LogicalDevice, error) {
+ logger.Debugw("GetLogicalDevice-request", log.Fields{"id": id})
+ return handler.logicalDeviceMgr.GetLogicalDevice(ctx, id.Id)
+}
+
+// ListLogicalDevices returns the list of all logical devices
+func (handler *NBIHandler) ListLogicalDevices(ctx context.Context, empty *empty.Empty) (*voltha.LogicalDevices, error) {
+ logger.Debug("ListLogicalDevices-request")
+ return handler.logicalDeviceMgr.ListLogicalDevices(ctx)
+}
+
+// ListAdapters returns the contents of all adapters known to the system
+func (handler *NBIHandler) ListAdapters(ctx context.Context, empty *empty.Empty) (*voltha.Adapters, error) {
+ logger.Debug("ListAdapters")
+ return handler.adapterMgr.ListAdapters(ctx)
+}
+
+// ListLogicalDeviceFlows returns the flows of logical device
+func (handler *NBIHandler) ListLogicalDeviceFlows(ctx context.Context, id *voltha.ID) (*openflow_13.Flows, error) {
+ logger.Debugw("ListLogicalDeviceFlows", log.Fields{"id": *id})
+ return handler.logicalDeviceMgr.ListLogicalDeviceFlows(ctx, id.Id)
+}
+
+// ListLogicalDeviceFlowGroups returns logical device flow groups
+func (handler *NBIHandler) ListLogicalDeviceFlowGroups(ctx context.Context, id *voltha.ID) (*openflow_13.FlowGroups, error) {
+ logger.Debugw("ListLogicalDeviceFlowGroups", log.Fields{"id": *id})
+ return handler.logicalDeviceMgr.ListLogicalDeviceFlowGroups(ctx, id.Id)
+}
+
+// ListLogicalDevicePorts returns ports of logical device
+func (handler *NBIHandler) ListLogicalDevicePorts(ctx context.Context, id *voltha.ID) (*voltha.LogicalPorts, error) {
+ logger.Debugw("ListLogicalDevicePorts", log.Fields{"logicaldeviceid": id})
+ return handler.logicalDeviceMgr.ListLogicalDevicePorts(ctx, id.Id)
+}
+
+// CreateDevice creates a new parent device in the data model
+func (handler *NBIHandler) CreateDevice(ctx context.Context, device *voltha.Device) (*voltha.Device, error) {
+ if device.MacAddress == "" && device.GetHostAndPort() == "" {
+ logger.Errorf("No Device Info Present")
+ return &voltha.Device{}, errors.New("no-device-info-present; MAC or HOSTIP&PORT")
+ }
+ logger.Debugw("create-device", log.Fields{"device": *device})
+
+ ch := make(chan interface{})
+ defer close(ch)
+ go handler.deviceMgr.CreateDevice(ctx, device, ch)
+ select {
+ case res := <-ch:
+ if res != nil {
+ if err, ok := res.(error); ok {
+ logger.Errorw("create-device-failed", log.Fields{"error": err})
+ return nil, err
+ }
+ if d, ok := res.(*voltha.Device); ok {
+ return d, nil
+ }
+ }
+ logger.Warnw("create-device-unexpected-return-type", log.Fields{"result": res})
+ err := status.Errorf(codes.Internal, "%s", res)
+ return &voltha.Device{}, err
+ case <-ctx.Done():
+ logger.Debug("createdevice-client-timeout")
+ return &voltha.Device{}, ctx.Err()
+ }
+}
+
+// EnableDevice activates a device by invoking the adopt_device API on the appropriate adapter
+func (handler *NBIHandler) EnableDevice(ctx context.Context, id *voltha.ID) (*empty.Empty, error) {
+ logger.Debugw("enabledevice", log.Fields{"id": id})
+
+ ch := make(chan interface{})
+ defer close(ch)
+ go handler.deviceMgr.EnableDevice(ctx, id, ch)
+ return waitForNilResponseOnSuccess(ctx, ch)
+}
+
+// DisableDevice disables a device along with any child device it may have
+func (handler *NBIHandler) DisableDevice(ctx context.Context, id *voltha.ID) (*empty.Empty, error) {
+ logger.Debugw("disabledevice-request", log.Fields{"id": id})
+
+ ch := make(chan interface{})
+ defer close(ch)
+ go handler.deviceMgr.DisableDevice(ctx, id, ch)
+ return waitForNilResponseOnSuccess(ctx, ch)
+}
+
+//RebootDevice invoked the reboot API to the corresponding adapter
+func (handler *NBIHandler) RebootDevice(ctx context.Context, id *voltha.ID) (*empty.Empty, error) {
+ logger.Debugw("rebootDevice-request", log.Fields{"id": id})
+
+ ch := make(chan interface{})
+ defer close(ch)
+ go handler.deviceMgr.RebootDevice(ctx, id, ch)
+ return waitForNilResponseOnSuccess(ctx, ch)
+}
+
+// DeleteDevice removes a device from the data model
+func (handler *NBIHandler) DeleteDevice(ctx context.Context, id *voltha.ID) (*empty.Empty, error) {
+ logger.Debugw("deletedevice-request", log.Fields{"id": id})
+
+ ch := make(chan interface{})
+ defer close(ch)
+ go handler.deviceMgr.DeleteDevice(ctx, id, ch)
+ return waitForNilResponseOnSuccess(ctx, ch)
+}
+
+// ListDevicePorts returns the ports details for a specific device entry
+func (handler *NBIHandler) ListDevicePorts(ctx context.Context, id *voltha.ID) (*voltha.Ports, error) {
+ logger.Debugw("listdeviceports-request", log.Fields{"id": id})
+ device, err := handler.deviceMgr.GetDevice(ctx, id.Id)
+ if err != nil {
+ return &voltha.Ports{}, err
+ }
+ ports := &voltha.Ports{}
+ ports.Items = append(ports.Items, device.Ports...)
+ return ports, nil
+}
+
+// ListDeviceFlows returns the flow details for a specific device entry
+func (handler *NBIHandler) ListDeviceFlows(ctx context.Context, id *voltha.ID) (*openflow_13.Flows, error) {
+ logger.Debugw("listdeviceflows-request", log.Fields{"id": id})
+
+ device, err := handler.deviceMgr.GetDevice(ctx, id.Id)
+ if err != nil {
+ return &openflow_13.Flows{}, err
+ }
+ flows := &openflow_13.Flows{}
+ flows.Items = append(flows.Items, device.Flows.Items...)
+ return flows, nil
+}
+
+// ListDeviceFlowGroups returns the flow group details for a specific device entry
+func (handler *NBIHandler) ListDeviceFlowGroups(ctx context.Context, id *voltha.ID) (*voltha.FlowGroups, error) {
+ logger.Debugw("ListDeviceFlowGroups", log.Fields{"deviceid": id})
+
+ if device, _ := handler.deviceMgr.GetDevice(ctx, id.Id); device != nil {
+ return device.GetFlowGroups(), nil
+ }
+ return &voltha.FlowGroups{}, status.Errorf(codes.NotFound, "device-%s", id.Id)
+}
+
+// ListDeviceGroups returns all the device groups known to the system
+func (handler *NBIHandler) ListDeviceGroups(ctx context.Context, empty *empty.Empty) (*voltha.DeviceGroups, error) {
+ logger.Debug("ListDeviceGroups")
+ return &voltha.DeviceGroups{}, errors.New("UnImplemented")
+}
+
+// GetDeviceGroup returns a specific device group entry
+func (handler *NBIHandler) GetDeviceGroup(ctx context.Context, id *voltha.ID) (*voltha.DeviceGroup, error) {
+ logger.Debug("GetDeviceGroup")
+ return &voltha.DeviceGroup{}, errors.New("UnImplemented")
+}
+
+// ListDeviceTypes returns all the device types known to the system
+func (handler *NBIHandler) ListDeviceTypes(ctx context.Context, _ *empty.Empty) (*voltha.DeviceTypes, error) {
+ logger.Debug("ListDeviceTypes")
+
+ return &voltha.DeviceTypes{Items: handler.adapterMgr.ListDeviceTypes()}, nil
+}
+
+// GetDeviceType returns the device type for a specific device entry
+func (handler *NBIHandler) GetDeviceType(ctx context.Context, id *voltha.ID) (*voltha.DeviceType, error) {
+ logger.Debugw("GetDeviceType", log.Fields{"typeid": id})
+
+ if deviceType := handler.adapterMgr.GetDeviceType(id.Id); deviceType != nil {
+ return deviceType, nil
+ }
+ return &voltha.DeviceType{}, status.Errorf(codes.NotFound, "device_type-%s", id.Id)
+}
+
+// GetVoltha returns the contents of all components (i.e. devices, logical_devices, ...)
+func (handler *NBIHandler) GetVoltha(ctx context.Context, empty *empty.Empty) (*voltha.Voltha, error) {
+
+ logger.Debug("GetVoltha")
+ /*
+ * For now, encode all the version information into a JSON object and
+ * pass that back as "version" so the client can get all the
+ * information associated with the version. Long term the API should
+ * better accomidate this, but for now this will work.
+ */
+ data, err := json.Marshal(&version.VersionInfo)
+ info := version.VersionInfo.Version
+ if err != nil {
+ logger.Warnf("Unable to encode version information as JSON: %s", err.Error())
+ } else {
+ info = string(data)
+ }
+
+ return &voltha.Voltha{
+ Version: info,
+ }, nil
+}
+
+// processImageRequest is a helper method to execute an image download request
+func (handler *NBIHandler) processImageRequest(ctx context.Context, img *voltha.ImageDownload, requestType int) (*common.OperationResp, error) {
+ logger.Debugw("processImageDownload", log.Fields{"img": *img, "requestType": requestType})
+
+ failedresponse := &common.OperationResp{Code: voltha.OperationResp_OPERATION_FAILURE}
+
+ ch := make(chan interface{})
+ defer close(ch)
+ switch requestType {
+ case ImageDownload:
+ go handler.deviceMgr.DownloadImage(ctx, img, ch)
+ case CancelImageDownload:
+ go handler.deviceMgr.CancelImageDownload(ctx, img, ch)
+ case ActivateImage:
+ go handler.deviceMgr.ActivateImage(ctx, img, ch)
+ case RevertImage:
+ go handler.deviceMgr.RevertImage(ctx, img, ch)
+ default:
+ logger.Warn("invalid-request-type", log.Fields{"requestType": requestType})
+ return failedresponse, status.Errorf(codes.InvalidArgument, "%d", requestType)
+ }
+ select {
+ case res := <-ch:
+ if res != nil {
+ if err, ok := res.(error); ok {
+ return failedresponse, err
+ }
+ if opResp, ok := res.(*common.OperationResp); ok {
+ return opResp, nil
+ }
+ }
+ logger.Warnw("download-image-unexpected-return-type", log.Fields{"result": res})
+ return failedresponse, status.Errorf(codes.Internal, "%s", res)
+ case <-ctx.Done():
+ logger.Debug("downloadImage-client-timeout")
+ return &common.OperationResp{}, ctx.Err()
+ }
+}
+
+// DownloadImage execute an image download request
+func (handler *NBIHandler) DownloadImage(ctx context.Context, img *voltha.ImageDownload) (*common.OperationResp, error) {
+ logger.Debugw("DownloadImage-request", log.Fields{"img": *img})
+
+ return handler.processImageRequest(ctx, img, ImageDownload)
+}
+
+// CancelImageDownload cancels image download request
+func (handler *NBIHandler) CancelImageDownload(ctx context.Context, img *voltha.ImageDownload) (*common.OperationResp, error) {
+ logger.Debugw("cancelImageDownload-request", log.Fields{"img": *img})
+ return handler.processImageRequest(ctx, img, CancelImageDownload)
+}
+
+// ActivateImageUpdate activates image update request
+func (handler *NBIHandler) ActivateImageUpdate(ctx context.Context, img *voltha.ImageDownload) (*common.OperationResp, error) {
+ logger.Debugw("activateImageUpdate-request", log.Fields{"img": *img})
+ return handler.processImageRequest(ctx, img, ActivateImage)
+}
+
+// RevertImageUpdate reverts image update
+func (handler *NBIHandler) RevertImageUpdate(ctx context.Context, img *voltha.ImageDownload) (*common.OperationResp, error) {
+ logger.Debugw("revertImageUpdate-request", log.Fields{"img": *img})
+ return handler.processImageRequest(ctx, img, RevertImage)
+}
+
+// GetImageDownloadStatus returns status of image download
+func (handler *NBIHandler) GetImageDownloadStatus(ctx context.Context, img *voltha.ImageDownload) (*voltha.ImageDownload, error) {
+ logger.Debugw("getImageDownloadStatus-request", log.Fields{"img": *img})
+
+ failedresponse := &voltha.ImageDownload{DownloadState: voltha.ImageDownload_DOWNLOAD_UNKNOWN}
+
+ ch := make(chan interface{})
+ defer close(ch)
+ go handler.deviceMgr.GetImageDownloadStatus(ctx, img, ch)
+
+ select {
+ case res := <-ch:
+ if res != nil {
+ if err, ok := res.(error); ok {
+ return failedresponse, err
+ }
+ if downloadResp, ok := res.(*voltha.ImageDownload); ok {
+ return downloadResp, nil
+ }
+ }
+ logger.Warnw("download-image-status", log.Fields{"result": res})
+ return failedresponse, status.Errorf(codes.Internal, "%s", res)
+ case <-ctx.Done():
+ logger.Debug("downloadImage-client-timeout")
+ return failedresponse, ctx.Err()
+ }
+}
+
+// GetImageDownload returns image download
+func (handler *NBIHandler) GetImageDownload(ctx context.Context, img *voltha.ImageDownload) (*voltha.ImageDownload, error) {
+ logger.Debugw("GetImageDownload-request", log.Fields{"img": *img})
+
+ download, err := handler.deviceMgr.GetImageDownload(ctx, img)
+ if err != nil {
+ return &voltha.ImageDownload{DownloadState: voltha.ImageDownload_DOWNLOAD_UNKNOWN}, err
+ }
+ return download, nil
+}
+
+// ListImageDownloads returns image downloads
+func (handler *NBIHandler) ListImageDownloads(ctx context.Context, id *voltha.ID) (*voltha.ImageDownloads, error) {
+ logger.Debugw("ListImageDownloads-request", log.Fields{"deviceId": id.Id})
+
+ downloads, err := handler.deviceMgr.ListImageDownloads(ctx, id.Id)
+ if err != nil {
+ failedResp := &voltha.ImageDownloads{
+ Items: []*voltha.ImageDownload{
+ {DownloadState: voltha.ImageDownload_DOWNLOAD_UNKNOWN},
+ },
+ }
+ return failedResp, err
+ }
+ return downloads, nil
+}
+
+// GetImages returns all images for a specific device entry
+func (handler *NBIHandler) GetImages(ctx context.Context, id *voltha.ID) (*voltha.Images, error) {
+ logger.Debugw("GetImages", log.Fields{"deviceid": id.Id})
+ device, err := handler.deviceMgr.GetDevice(ctx, id.Id)
+ if err != nil {
+ return &voltha.Images{}, err
+ }
+ return device.GetImages(), nil
+}
+
+// UpdateDevicePmConfigs updates the PM configs
+func (handler *NBIHandler) UpdateDevicePmConfigs(ctx context.Context, configs *voltha.PmConfigs) (*empty.Empty, error) {
+ logger.Debugw("UpdateDevicePmConfigs-request", log.Fields{"configs": *configs})
+
+ ch := make(chan interface{})
+ defer close(ch)
+ go handler.deviceMgr.UpdatePmConfigs(ctx, configs, ch)
+ return waitForNilResponseOnSuccess(ctx, ch)
+}
+
+// ListDevicePmConfigs returns pm configs of device
+func (handler *NBIHandler) ListDevicePmConfigs(ctx context.Context, id *voltha.ID) (*voltha.PmConfigs, error) {
+ logger.Debugw("ListDevicePmConfigs-request", log.Fields{"deviceId": *id})
+ return handler.deviceMgr.ListPmConfigs(ctx, id.Id)
+}
+
+func (handler *NBIHandler) CreateEventFilter(ctx context.Context, filter *voltha.EventFilter) (*voltha.EventFilter, error) {
+ logger.Debugw("CreateEventFilter-request", log.Fields{"filter": *filter})
+ return nil, errors.New("UnImplemented")
+}
+
+func (handler *NBIHandler) UpdateEventFilter(ctx context.Context, filter *voltha.EventFilter) (*voltha.EventFilter, error) {
+ logger.Debugw("UpdateEventFilter-request", log.Fields{"filter": *filter})
+ return nil, errors.New("UnImplemented")
+}
+
+func (handler *NBIHandler) DeleteEventFilter(ctx context.Context, filterInfo *voltha.EventFilter) (*empty.Empty, error) {
+ logger.Debugw("DeleteEventFilter-request", log.Fields{"device-id": filterInfo.DeviceId, "filter-id": filterInfo.Id})
+ return nil, errors.New("UnImplemented")
+}
+
+// GetEventFilter returns all the filters present for a device
+func (handler *NBIHandler) GetEventFilter(ctx context.Context, id *voltha.ID) (*voltha.EventFilters, error) {
+ logger.Debugw("GetEventFilter-request", log.Fields{"device-id": id})
+ return nil, errors.New("UnImplemented")
+}
+
+// ListEventFilters returns all the filters known to the system
+func (handler *NBIHandler) ListEventFilters(ctx context.Context, empty *empty.Empty) (*voltha.EventFilters, error) {
+ logger.Debug("ListEventFilter-request")
+ return nil, errors.New("UnImplemented")
+}
+
+func (handler *NBIHandler) SelfTest(ctx context.Context, id *voltha.ID) (*voltha.SelfTestResponse, error) {
+ logger.Debugw("SelfTest-request", log.Fields{"id": id})
+ return &voltha.SelfTestResponse{}, errors.New("UnImplemented")
+}
+
+// StreamPacketsOut sends packets to adapter
+func (handler *NBIHandler) StreamPacketsOut(packets voltha.VolthaService_StreamPacketsOutServer) error {
+ logger.Debugw("StreamPacketsOut-request", log.Fields{"packets": packets})
+loop:
+ for {
+ select {
+ case <-packets.Context().Done():
+ logger.Infow("StreamPacketsOut-context-done", log.Fields{"packets": packets, "error": packets.Context().Err()})
+ break loop
+ default:
+ }
+
+ packet, err := packets.Recv()
+
+ if err == io.EOF {
+ logger.Debugw("Received-EOF", log.Fields{"packets": packets})
+ break loop
+ }
+
+ if err != nil {
+ logger.Errorw("Failed to receive packet out", log.Fields{"error": err})
+ continue
+ }
+
+ handler.logicalDeviceMgr.PacketOut(packets.Context(), packet)
+ }
+
+ logger.Debugw("StreamPacketsOut-request-done", log.Fields{"packets": packets})
+ return nil
+}
+
+func (handler *NBIHandler) SendPacketIn(deviceID string, transationID string, packet *openflow_13.OfpPacketIn) {
+ // TODO: Augment the OF PacketIn to include the transactionId
+ packetIn := openflow_13.PacketIn{Id: deviceID, PacketIn: packet}
+ logger.Debugw("SendPacketIn", log.Fields{"packetIn": packetIn})
+ handler.packetInQueue <- packetIn
+}
+
+type callTracker struct {
+ failedPacket interface{}
+}
+type streamTracker struct {
+ calls map[string]*callTracker
+ sync.Mutex
+}
+
+var streamingTracker = &streamTracker{calls: make(map[string]*callTracker)}
+
+func (handler *NBIHandler) getStreamingTracker(method string, done chan<- bool) *callTracker {
+ streamingTracker.Lock()
+ defer streamingTracker.Unlock()
+ if _, ok := streamingTracker.calls[method]; ok {
+ // bail out the other packet in thread
+ logger.Debugf("%s streaming call already running. Exiting it", method)
+ done <- true
+ logger.Debugf("Last %s exited. Continuing ...", method)
+ } else {
+ streamingTracker.calls[method] = &callTracker{failedPacket: nil}
+ }
+ return streamingTracker.calls[method]
+}
+
+func (handler *NBIHandler) flushFailedPackets(tracker *callTracker) error {
+ if tracker.failedPacket != nil {
+ switch tracker.failedPacket.(type) {
+ case openflow_13.PacketIn:
+ logger.Debug("Enqueueing last failed packetIn")
+ handler.packetInQueue <- tracker.failedPacket.(openflow_13.PacketIn)
+ case openflow_13.ChangeEvent:
+ logger.Debug("Enqueueing last failed changeEvent")
+ handler.changeEventQueue <- tracker.failedPacket.(openflow_13.ChangeEvent)
+ }
+ }
+ return nil
+}
+
+// ReceivePacketsIn receives packets from adapter
+func (handler *NBIHandler) ReceivePacketsIn(empty *empty.Empty, packetsIn voltha.VolthaService_ReceivePacketsInServer) error {
+ var streamingTracker = handler.getStreamingTracker("ReceivePacketsIn", handler.packetInQueueDone)
+ logger.Debugw("ReceivePacketsIn-request", log.Fields{"packetsIn": packetsIn})
+
+ err := handler.flushFailedPackets(streamingTracker)
+ if err != nil {
+ logger.Errorw("unable-to-flush-failed-packets", log.Fields{"error": err})
+ }
+
+loop:
+ for {
+ select {
+ case packet := <-handler.packetInQueue:
+ logger.Debugw("sending-packet-in", log.Fields{
+ "packet": hex.EncodeToString(packet.PacketIn.Data),
+ })
+ if err := packetsIn.Send(&packet); err != nil {
+ logger.Errorw("failed-to-send-packet", log.Fields{"error": err})
+ // save the last failed packet in
+ streamingTracker.failedPacket = packet
+ } else {
+ if streamingTracker.failedPacket != nil {
+ // reset last failed packet saved to avoid flush
+ streamingTracker.failedPacket = nil
+ }
+ }
+ case <-handler.packetInQueueDone:
+ logger.Debug("Another ReceivePacketsIn running. Bailing out ...")
+ break loop
+ }
+ }
+
+ //TODO: Find an elegant way to get out of the above loop when the Core is stopped
+ return nil
+}
+
+func (handler *NBIHandler) SendChangeEvent(deviceID string, portStatus *openflow_13.OfpPortStatus) {
+ // TODO: validate the type of portStatus parameter
+ //if _, ok := portStatus.(*openflow_13.OfpPortStatus); ok {
+ //}
+ event := openflow_13.ChangeEvent{Id: deviceID, Event: &openflow_13.ChangeEvent_PortStatus{PortStatus: portStatus}}
+ logger.Debugw("SendChangeEvent", log.Fields{"event": event})
+ handler.changeEventQueue <- event
+}
+
+// ReceiveChangeEvents receives change in events
+func (handler *NBIHandler) ReceiveChangeEvents(empty *empty.Empty, changeEvents voltha.VolthaService_ReceiveChangeEventsServer) error {
+ var streamingTracker = handler.getStreamingTracker("ReceiveChangeEvents", handler.changeEventQueueDone)
+ logger.Debugw("ReceiveChangeEvents-request", log.Fields{"changeEvents": changeEvents})
+
+ err := handler.flushFailedPackets(streamingTracker)
+ if err != nil {
+ logger.Errorw("unable-to-flush-failed-packets", log.Fields{"error": err})
+ }
+
+loop:
+ for {
+ select {
+ // Dequeue a change event
+ case event := <-handler.changeEventQueue:
+ logger.Debugw("sending-change-event", log.Fields{"event": event})
+ if err := changeEvents.Send(&event); err != nil {
+ logger.Errorw("failed-to-send-change-event", log.Fields{"error": err})
+ // save last failed changeevent
+ streamingTracker.failedPacket = event
+ } else {
+ if streamingTracker.failedPacket != nil {
+ // reset last failed event saved on success to avoid flushing
+ streamingTracker.failedPacket = nil
+ }
+ }
+ case <-handler.changeEventQueueDone:
+ logger.Debug("Another ReceiveChangeEvents already running. Bailing out ...")
+ break loop
+ }
+ }
+
+ return nil
+}
+
+func (handler *NBIHandler) GetChangeEventsQueueForTest() <-chan openflow_13.ChangeEvent {
+ return handler.changeEventQueue
+}
+
+// Subscribe subscribing request of ofagent
+func (handler *NBIHandler) Subscribe(
+ ctx context.Context,
+ ofAgent *voltha.OfAgentSubscriber,
+) (*voltha.OfAgentSubscriber, error) {
+ logger.Debugw("Subscribe-request", log.Fields{"ofAgent": ofAgent})
+ return &voltha.OfAgentSubscriber{OfagentId: ofAgent.OfagentId, VolthaId: ofAgent.VolthaId}, nil
+}
+
+// GetAlarmDeviceData @TODO useless stub, what should this actually do?
+func (handler *NBIHandler) GetAlarmDeviceData(ctx context.Context, in *common.ID) (*omci.AlarmDeviceData, error) {
+ logger.Debug("GetAlarmDeviceData-stub")
+ return &omci.AlarmDeviceData{}, errors.New("UnImplemented")
+}
+
+// ListLogicalDeviceMeters returns logical device meters
+func (handler *NBIHandler) ListLogicalDeviceMeters(ctx context.Context, id *voltha.ID) (*openflow_13.Meters, error) {
+
+ logger.Debugw("ListLogicalDeviceMeters", log.Fields{"id": *id})
+ return handler.logicalDeviceMgr.ListLogicalDeviceMeters(ctx, id.Id)
+}
+
+// GetMeterStatsOfLogicalDevice @TODO useless stub, what should this actually do?
+func (handler *NBIHandler) GetMeterStatsOfLogicalDevice(ctx context.Context, in *common.ID) (*openflow_13.MeterStatsReply, error) {
+ logger.Debug("GetMeterStatsOfLogicalDevice")
+ return &openflow_13.MeterStatsReply{}, errors.New("UnImplemented")
+}
+
+// GetMibDeviceData @TODO useless stub, what should this actually do?
+func (handler *NBIHandler) GetMibDeviceData(ctx context.Context, in *common.ID) (*omci.MibDeviceData, error) {
+ logger.Debug("GetMibDeviceData")
+ return &omci.MibDeviceData{}, errors.New("UnImplemented")
+}
+
+// SimulateAlarm sends simulate alarm request
+func (handler *NBIHandler) SimulateAlarm(
+ ctx context.Context,
+ in *voltha.SimulateAlarmRequest,
+) (*common.OperationResp, error) {
+ logger.Debugw("SimulateAlarm-request", log.Fields{"id": in.Id})
+ successResp := &common.OperationResp{Code: common.OperationResp_OPERATION_SUCCESS}
+ ch := make(chan interface{})
+ defer close(ch)
+ go handler.deviceMgr.SimulateAlarm(ctx, in, ch)
+ return successResp, nil
+}
+
+// UpdateLogicalDeviceMeterTable - This function sends meter mod request to logical device manager and waits for response
+func (handler *NBIHandler) UpdateLogicalDeviceMeterTable(ctx context.Context, meter *openflow_13.MeterModUpdate) (*empty.Empty, error) {
+ logger.Debugw("UpdateLogicalDeviceMeterTable-request",
+ log.Fields{"meter": meter, "test": common.TestModeKeys_api_test.String()})
+ ch := make(chan interface{})
+ defer close(ch)
+ go handler.logicalDeviceMgr.UpdateMeterTable(ctx, meter.Id, meter.MeterMod, ch)
+ return waitForNilResponseOnSuccess(ctx, ch)
+}
+
+// GetMembership returns membership
+func (handler *NBIHandler) GetMembership(context.Context, *empty.Empty) (*voltha.Membership, error) {
+ return &voltha.Membership{}, errors.New("UnImplemented")
+}
+
+// UpdateMembership updates membership
+func (handler *NBIHandler) UpdateMembership(context.Context, *voltha.Membership) (*empty.Empty, error) {
+ return &empty.Empty{}, errors.New("UnImplemented")
+}
+
+func (handler *NBIHandler) EnablePort(ctx context.Context, port *voltha.Port) (*empty.Empty, error) {
+ logger.Debugw("EnablePort-request", log.Fields{"device-id": port.DeviceId, "port-no": port.PortNo})
+ ch := make(chan interface{})
+ defer close(ch)
+ go handler.deviceMgr.EnablePort(ctx, port, ch)
+ return waitForNilResponseOnSuccess(ctx, ch)
+}
+
+func (handler *NBIHandler) DisablePort(ctx context.Context, port *voltha.Port) (*empty.Empty, error) {
+
+ logger.Debugw("DisablePort-request", log.Fields{"device-id": port.DeviceId, "port-no": port.PortNo})
+ ch := make(chan interface{})
+ defer close(ch)
+ go handler.deviceMgr.DisablePort(ctx, port, ch)
+ return waitForNilResponseOnSuccess(ctx, ch)
+}
+
+func (handler *NBIHandler) StartOmciTestAction(ctx context.Context, omcitestrequest *voltha.OmciTestRequest) (*voltha.TestResponse, error) {
+ logger.Debugw("Omci_test_Request", log.Fields{"id": omcitestrequest.Id, "uuid": omcitestrequest.Uuid})
+ return handler.deviceMgr.StartOmciTest(ctx, omcitestrequest)
+}
diff --git a/rw_core/core/api/grpc_nbi_handler_test.go b/rw_core/core/api/grpc_nbi_handler_test.go
new file mode 100755
index 0000000..0579f94
--- /dev/null
+++ b/rw_core/core/api/grpc_nbi_handler_test.go
@@ -0,0 +1,1227 @@
+/*
+* Copyright 2019-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
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+ */
+package api
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "math/rand"
+ "os"
+ "runtime"
+ "runtime/pprof"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/golang/protobuf/ptypes/empty"
+ "github.com/opencord/voltha-go/db/model"
+ "github.com/opencord/voltha-go/rw_core/config"
+ "github.com/opencord/voltha-go/rw_core/core/adapter"
+ "github.com/opencord/voltha-go/rw_core/core/device"
+ cm "github.com/opencord/voltha-go/rw_core/mocks"
+ "github.com/opencord/voltha-lib-go/v3/pkg/db"
+ "github.com/opencord/voltha-lib-go/v3/pkg/flows"
+ "github.com/opencord/voltha-lib-go/v3/pkg/kafka"
+ "github.com/opencord/voltha-lib-go/v3/pkg/log"
+ mock_etcd "github.com/opencord/voltha-lib-go/v3/pkg/mocks/etcd"
+ mock_kafka "github.com/opencord/voltha-lib-go/v3/pkg/mocks/kafka"
+ "github.com/opencord/voltha-lib-go/v3/pkg/version"
+ ofp "github.com/opencord/voltha-protos/v3/go/openflow_13"
+ "github.com/opencord/voltha-protos/v3/go/voltha"
+ "github.com/phayes/freeport"
+ "github.com/stretchr/testify/assert"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/status"
+)
+
+const (
+ coreName = "rw_core"
+)
+
+type NBTest struct {
+ etcdServer *mock_etcd.EtcdServer
+ deviceMgr *device.Manager
+ logicalDeviceMgr *device.LogicalManager
+ adapterMgr *adapter.Manager
+ kmp kafka.InterContainerProxy
+ kClient kafka.Client
+ kvClientPort int
+ numONUPerOLT int
+ startingUNIPortNo int
+ oltAdapter *cm.OLTAdapter
+ onuAdapter *cm.ONUAdapter
+ oltAdapterName string
+ onuAdapterName string
+ coreInstanceID string
+ defaultTimeout time.Duration
+ maxTimeout time.Duration
+}
+
+func newNBTest() *NBTest {
+ test := &NBTest{}
+ // Start the embedded etcd server
+ var err error
+ test.etcdServer, test.kvClientPort, err = startEmbeddedEtcdServer("voltha.rwcore.nb.test", "voltha.rwcore.nb.etcd", "error")
+ if err != nil {
+ logger.Fatal(err)
+ }
+ // Create the kafka client
+ test.kClient = mock_kafka.NewKafkaClient()
+ test.oltAdapterName = "olt_adapter_mock"
+ test.onuAdapterName = "onu_adapter_mock"
+ test.coreInstanceID = "rw-nbi-test"
+ test.defaultTimeout = 10 * time.Second
+ test.maxTimeout = 20 * time.Second
+ return test
+}
+
+func (nb *NBTest) startCore(inCompeteMode bool) {
+ ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+ defer cancel()
+ cfg := config.NewRWCoreFlags()
+ cfg.CorePairTopic = "rw_core"
+ cfg.DefaultRequestTimeout = nb.defaultTimeout
+ cfg.DefaultCoreTimeout = nb.defaultTimeout
+ cfg.KVStorePort = nb.kvClientPort
+ cfg.InCompetingMode = inCompeteMode
+ grpcPort, err := freeport.GetFreePort()
+ if err != nil {
+ logger.Fatal("Cannot get a freeport for grpc")
+ }
+ cfg.GrpcPort = grpcPort
+ cfg.GrpcHost = "127.0.0.1"
+ setCoreCompeteMode(inCompeteMode)
+ client := setupKVClient(cfg, nb.coreInstanceID)
+ backend := &db.Backend{
+ Client: client,
+ StoreType: cfg.KVStoreType,
+ Host: cfg.KVStoreHost,
+ Port: cfg.KVStorePort,
+ Timeout: cfg.KVStoreTimeout,
+ LivenessChannelInterval: cfg.LiveProbeInterval / 2,
+ PathPrefix: cfg.KVStoreDataPrefix}
+ nb.kmp = kafka.NewInterContainerProxy(
+ kafka.InterContainerHost(cfg.KafkaAdapterHost),
+ kafka.InterContainerPort(cfg.KafkaAdapterPort),
+ kafka.MsgClient(nb.kClient),
+ kafka.DefaultTopic(&kafka.Topic{Name: cfg.CoreTopic}),
+ kafka.DeviceDiscoveryTopic(&kafka.Topic{Name: cfg.AffinityRouterTopic}))
+
+ endpointMgr := kafka.NewEndpointManager(backend)
+ proxy := model.NewProxy(backend, "/")
+ nb.adapterMgr = adapter.NewAdapterManager(proxy, nb.coreInstanceID, nb.kClient)
+ nb.deviceMgr, nb.logicalDeviceMgr = device.NewDeviceManagers(proxy, nb.adapterMgr, nb.kmp, endpointMgr, cfg.CorePairTopic, nb.coreInstanceID, cfg.DefaultCoreTimeout)
+ if err = nb.adapterMgr.Start(ctx); err != nil {
+ logger.Fatalf("Cannot start adapterMgr: %s", err)
+ }
+ nb.deviceMgr.Start(ctx)
+ nb.logicalDeviceMgr.Start(ctx)
+
+ if err = nb.kmp.Start(); err != nil {
+ logger.Fatalf("Cannot start InterContainerProxy: %s", err)
+ }
+ requestProxy := NewAdapterRequestHandlerProxy(nb.coreInstanceID, nb.deviceMgr, nb.adapterMgr, proxy, proxy, cfg.LongRunningRequestTimeout, cfg.DefaultRequestTimeout)
+ if err := nb.kmp.SubscribeWithRequestHandlerInterface(kafka.Topic{Name: cfg.CoreTopic}, requestProxy); err != nil {
+ logger.Fatalf("Cannot add request handler: %s", err)
+ }
+ if err := nb.kmp.SubscribeWithDefaultRequestHandler(kafka.Topic{Name: cfg.CorePairTopic}, kafka.OffsetNewest); err != nil {
+ logger.Fatalf("Cannot add default request handler: %s", err)
+ }
+}
+
+func (nb *NBTest) createAndregisterAdapters(t *testing.T) {
+ // Setup the mock OLT adapter
+ oltAdapter, err := createMockAdapter(OltAdapter, nb.kClient, nb.coreInstanceID, coreName, nb.oltAdapterName)
+ if err != nil {
+ logger.Fatalw("setting-mock-olt-adapter-failed", log.Fields{"error": err})
+ }
+ nb.oltAdapter = (oltAdapter).(*cm.OLTAdapter)
+ nb.numONUPerOLT = nb.oltAdapter.GetNumONUPerOLT()
+ nb.startingUNIPortNo = nb.oltAdapter.GetStartingUNIPortNo()
+
+ // Register the adapter
+ registrationData := &voltha.Adapter{
+ Id: nb.oltAdapterName,
+ Vendor: "Voltha-olt",
+ Version: version.VersionInfo.Version,
+ Type: nb.oltAdapterName,
+ CurrentReplica: 1,
+ TotalReplicas: 1,
+ Endpoint: nb.oltAdapterName,
+ }
+ types := []*voltha.DeviceType{{Id: nb.oltAdapterName, Adapter: nb.oltAdapterName, AcceptsAddRemoveFlowUpdates: true}}
+ deviceTypes := &voltha.DeviceTypes{Items: types}
+ if _, err := nb.adapterMgr.RegisterAdapter(registrationData, deviceTypes); err != nil {
+ logger.Errorw("failed-to-register-adapter", log.Fields{"error": err})
+ assert.NotNil(t, err)
+ }
+
+ // Setup the mock ONU adapter
+ onuAdapter, err := createMockAdapter(OnuAdapter, nb.kClient, nb.coreInstanceID, coreName, nb.onuAdapterName)
+ if err != nil {
+ logger.Fatalw("setting-mock-onu-adapter-failed", log.Fields{"error": err})
+ }
+ nb.onuAdapter = (onuAdapter).(*cm.ONUAdapter)
+
+ // Register the adapter
+ registrationData = &voltha.Adapter{
+ Id: nb.onuAdapterName,
+ Vendor: "Voltha-onu",
+ Version: version.VersionInfo.Version,
+ Type: nb.onuAdapterName,
+ CurrentReplica: 1,
+ TotalReplicas: 1,
+ Endpoint: nb.onuAdapterName,
+ }
+ types = []*voltha.DeviceType{{Id: nb.onuAdapterName, Adapter: nb.onuAdapterName, AcceptsAddRemoveFlowUpdates: true}}
+ deviceTypes = &voltha.DeviceTypes{Items: types}
+ if _, err := nb.adapterMgr.RegisterAdapter(registrationData, deviceTypes); err != nil {
+ logger.Errorw("failed-to-register-adapter", log.Fields{"error": err})
+ assert.NotNil(t, err)
+ }
+}
+
+func (nb *NBTest) stopAll() {
+ if nb.kClient != nil {
+ nb.kClient.Stop()
+ }
+ if nb.logicalDeviceMgr != nil {
+ nb.logicalDeviceMgr.Stop(context.Background())
+ }
+ if nb.deviceMgr != nil {
+ nb.deviceMgr.Stop(context.Background())
+ }
+ if nb.kmp != nil {
+ nb.kmp.Stop()
+ }
+ if nb.etcdServer != nil {
+ stopEmbeddedEtcdServer(nb.etcdServer)
+ }
+}
+
+func (nb *NBTest) verifyLogicalDevices(t *testing.T, oltDevice *voltha.Device, nbi *NBIHandler) {
+ // Get the latest set of logical devices
+ logicalDevices, err := nbi.ListLogicalDevices(getContext(), &empty.Empty{})
+ assert.Nil(t, err)
+ assert.NotNil(t, logicalDevices)
+ assert.Equal(t, 1, len(logicalDevices.Items))
+
+ ld := logicalDevices.Items[0]
+ assert.NotEqual(t, "", ld.Id)
+ assert.NotEqual(t, uint64(0), ld.DatapathId)
+ assert.Equal(t, "olt_adapter_mock", ld.Desc.HwDesc)
+ assert.Equal(t, "olt_adapter_mock", ld.Desc.SwDesc)
+ assert.NotEqual(t, "", ld.RootDeviceId)
+ assert.NotEqual(t, "", ld.Desc.SerialNum)
+ assert.Equal(t, uint32(256), ld.SwitchFeatures.NBuffers)
+ assert.Equal(t, uint32(2), ld.SwitchFeatures.NTables)
+ assert.Equal(t, uint32(15), ld.SwitchFeatures.Capabilities)
+ assert.Equal(t, 1+nb.numONUPerOLT, len(ld.Ports))
+ assert.Equal(t, oltDevice.ParentId, ld.Id)
+ //Expected port no
+ expectedPortNo := make(map[uint32]bool)
+ expectedPortNo[uint32(2)] = false
+ for i := 0; i < nb.numONUPerOLT; i++ {
+ expectedPortNo[uint32(i+100)] = false
+ }
+ for _, p := range ld.Ports {
+ assert.Equal(t, p.OfpPort.PortNo, p.DevicePortNo)
+ assert.Equal(t, uint32(4), p.OfpPort.State)
+ expectedPortNo[p.OfpPort.PortNo] = true
+ if strings.HasPrefix(p.Id, "nni") {
+ assert.Equal(t, true, p.RootPort)
+ //assert.Equal(t, uint32(2), p.OfpPort.PortNo)
+ assert.Equal(t, p.Id, fmt.Sprintf("nni-%d", p.DevicePortNo))
+ } else {
+ assert.Equal(t, p.Id, fmt.Sprintf("uni-%d", p.DevicePortNo))
+ assert.Equal(t, false, p.RootPort)
+ }
+ }
+}
+
+func (nb *NBTest) verifyDevices(t *testing.T, nbi *NBIHandler) {
+ // Get the latest set of devices
+ devices, err := nbi.ListDevices(getContext(), &empty.Empty{})
+ assert.Nil(t, err)
+ assert.NotNil(t, devices)
+
+ // A device is ready to be examined when its ADMIN state is ENABLED and OPERATIONAL state is ACTIVE
+ var vFunction isDeviceConditionSatisfied = func(device *voltha.Device) bool {
+ return device.AdminState == voltha.AdminState_ENABLED && device.OperStatus == voltha.OperStatus_ACTIVE
+ }
+
+ var wg sync.WaitGroup
+ for _, device := range devices.Items {
+ wg.Add(1)
+ go func(wg *sync.WaitGroup, device *voltha.Device) {
+ // Wait until the device is in the right state
+ err := waitUntilDeviceReadiness(device.Id, nb.maxTimeout, vFunction, nbi)
+ assert.Nil(t, err)
+
+ // Now, verify the details of the device. First get the latest update
+ d, err := nbi.GetDevice(getContext(), &voltha.ID{Id: device.Id})
+ assert.Nil(t, err)
+ assert.Equal(t, voltha.AdminState_ENABLED, d.AdminState)
+ assert.Equal(t, voltha.ConnectStatus_REACHABLE, d.ConnectStatus)
+ assert.Equal(t, voltha.OperStatus_ACTIVE, d.OperStatus)
+ assert.Equal(t, d.Type, d.Adapter)
+ assert.NotEqual(t, "", d.MacAddress)
+ assert.NotEqual(t, "", d.SerialNumber)
+
+ if d.Type == "olt_adapter_mock" {
+ assert.Equal(t, true, d.Root)
+ assert.NotEqual(t, "", d.Id)
+ assert.NotEqual(t, "", d.ParentId)
+ assert.Nil(t, d.ProxyAddress)
+ } else if d.Type == "onu_adapter_mock" {
+ assert.Equal(t, false, d.Root)
+ assert.NotEqual(t, uint32(0), d.Vlan)
+ assert.NotEqual(t, "", d.Id)
+ assert.NotEqual(t, "", d.ParentId)
+ assert.NotEqual(t, "", d.ProxyAddress.DeviceId)
+ assert.Equal(t, "olt_adapter_mock", d.ProxyAddress.DeviceType)
+ } else {
+ assert.Error(t, errors.New("invalid-device-type"))
+ }
+ assert.Equal(t, 2, len(d.Ports))
+ for _, p := range d.Ports {
+ assert.Equal(t, voltha.AdminState_ENABLED, p.AdminState)
+ assert.Equal(t, voltha.OperStatus_ACTIVE, p.OperStatus)
+ if p.Type == voltha.Port_ETHERNET_NNI || p.Type == voltha.Port_ETHERNET_UNI {
+ assert.Equal(t, 0, len(p.Peers))
+ } else if p.Type == voltha.Port_PON_OLT {
+ assert.Equal(t, nb.numONUPerOLT, len(p.Peers))
+ assert.Equal(t, uint32(1), p.PortNo)
+ } else if p.Type == voltha.Port_PON_ONU {
+ assert.Equal(t, 1, len(p.Peers))
+ assert.Equal(t, uint32(1), p.PortNo)
+ } else {
+ assert.Error(t, errors.New("invalid-port"))
+ }
+ }
+ wg.Done()
+ }(&wg, device)
+ }
+ wg.Wait()
+}
+
+func (nb *NBTest) getADevice(rootDevice bool, nbi *NBIHandler) (*voltha.Device, error) {
+ devices, err := nbi.ListDevices(getContext(), &empty.Empty{})
+ if err != nil {
+ return nil, err
+ }
+ for _, d := range devices.Items {
+ if d.Root == rootDevice {
+ return d, nil
+ }
+ }
+ return nil, status.Errorf(codes.NotFound, "%v device not found", rootDevice)
+}
+
+func (nb *NBTest) testCoreWithoutData(t *testing.T, nbi *NBIHandler) {
+ lds, err := nbi.ListLogicalDevices(getContext(), &empty.Empty{})
+ assert.Nil(t, err)
+ assert.NotNil(t, lds)
+ assert.Equal(t, 0, len(lds.Items))
+ devices, err := nbi.ListDevices(getContext(), &empty.Empty{})
+ assert.Nil(t, err)
+ assert.NotNil(t, devices)
+ assert.Equal(t, 0, len(devices.Items))
+ adapters, err := nbi.ListAdapters(getContext(), &empty.Empty{})
+ assert.Equal(t, 0, len(adapters.Items))
+ assert.Nil(t, err)
+ assert.NotNil(t, adapters)
+}
+
+func (nb *NBTest) testAdapterRegistration(t *testing.T, nbi *NBIHandler) {
+ adapters, err := nbi.ListAdapters(getContext(), &empty.Empty{})
+ assert.Nil(t, err)
+ assert.NotNil(t, adapters)
+ assert.Equal(t, 2, len(adapters.Items))
+ for _, a := range adapters.Items {
+ switch a.Id {
+ case nb.oltAdapterName:
+ assert.Equal(t, "Voltha-olt", a.Vendor)
+ case nb.onuAdapterName:
+ assert.Equal(t, "Voltha-onu", a.Vendor)
+ default:
+ logger.Fatal("unregistered-adapter", a.Id)
+ }
+ }
+ deviceTypes, err := nbi.ListDeviceTypes(getContext(), &empty.Empty{})
+ assert.Nil(t, err)
+ assert.NotNil(t, deviceTypes)
+ assert.Equal(t, 2, len(deviceTypes.Items))
+ for _, dt := range deviceTypes.Items {
+ switch dt.Id {
+ case nb.oltAdapterName:
+ assert.Equal(t, nb.oltAdapterName, dt.Adapter)
+ assert.Equal(t, false, dt.AcceptsBulkFlowUpdate)
+ assert.Equal(t, true, dt.AcceptsAddRemoveFlowUpdates)
+ case nb.onuAdapterName:
+ assert.Equal(t, nb.onuAdapterName, dt.Adapter)
+ assert.Equal(t, false, dt.AcceptsBulkFlowUpdate)
+ assert.Equal(t, true, dt.AcceptsAddRemoveFlowUpdates)
+ default:
+ logger.Fatal("invalid-device-type", dt.Id)
+ }
+ }
+}
+
+func (nb *NBTest) testCreateDevice(t *testing.T, nbi *NBIHandler) {
+ // Create a valid device
+ oltDevice, err := nbi.CreateDevice(getContext(), &voltha.Device{Type: nb.oltAdapterName, MacAddress: "aa:bb:cc:cc:ee:ee"})
+ assert.Nil(t, err)
+ assert.NotNil(t, oltDevice)
+ device, err := nbi.GetDevice(getContext(), &voltha.ID{Id: oltDevice.Id})
+ assert.Nil(t, err)
+ assert.NotNil(t, device)
+ assert.Equal(t, oltDevice.String(), device.String())
+
+ // Try to create the same device
+ _, err = nbi.CreateDevice(getContext(), &voltha.Device{Type: nb.oltAdapterName, MacAddress: "aa:bb:cc:cc:ee:ee"})
+ assert.NotNil(t, err)
+ assert.Equal(t, "Device is already pre-provisioned", err.Error())
+
+ // Try to create a device with invalid data
+ _, err = nbi.CreateDevice(getContext(), &voltha.Device{Type: nb.oltAdapterName})
+ assert.NotNil(t, err)
+ assert.Equal(t, "no-device-info-present; MAC or HOSTIP&PORT", err.Error())
+
+ // Ensure we only have 1 device in the Core
+ devices, err := nbi.ListDevices(getContext(), &empty.Empty{})
+ assert.Nil(t, err)
+ assert.NotNil(t, devices)
+ assert.Equal(t, 1, len(devices.Items))
+ assert.Equal(t, oltDevice.String(), devices.Items[0].String())
+
+ //Remove the device
+ _, err = nbi.DeleteDevice(getContext(), &voltha.ID{Id: oltDevice.Id})
+ assert.Nil(t, err)
+
+ //Ensure there are no devices in the Core now - wait until condition satisfied or timeout
+ var vFunction isDevicesConditionSatisfied = func(devices *voltha.Devices) bool {
+ return devices != nil && len(devices.Items) == 0
+ }
+ err = waitUntilConditionForDevices(nb.maxTimeout, nbi, vFunction)
+ assert.Nil(t, err)
+}
+
+func (nb *NBTest) testEnableDevice(t *testing.T, nbi *NBIHandler) {
+ // Create a device that has no adapter registered
+ oltDeviceNoAdapter, err := nbi.CreateDevice(getContext(), &voltha.Device{Type: "noAdapterRegistered", MacAddress: "aa:bb:cc:cc:ee:ff"})
+ assert.Nil(t, err)
+ assert.NotNil(t, oltDeviceNoAdapter)
+
+ // Try to enable the oltDevice and check the error message
+ _, err = nbi.EnableDevice(getContext(), &voltha.ID{Id: oltDeviceNoAdapter.Id})
+ assert.NotNil(t, err)
+ assert.Equal(t, "Adapter-not-registered-for-device-type noAdapterRegistered", err.Error())
+
+ //Remove the device
+ _, err = nbi.DeleteDevice(getContext(), &voltha.ID{Id: oltDeviceNoAdapter.Id})
+ assert.Nil(t, err)
+
+ //Ensure there are no devices in the Core now - wait until condition satisfied or timeout
+ var vdFunction isDevicesConditionSatisfied = func(devices *voltha.Devices) bool {
+ return devices != nil && len(devices.Items) == 0
+ }
+ err = waitUntilConditionForDevices(nb.maxTimeout, nbi, vdFunction)
+ assert.Nil(t, err)
+
+ // Create a logical device monitor will automatically send trap and eapol flows to the devices being enables
+ var wg sync.WaitGroup
+ wg.Add(1)
+ go nb.monitorLogicalDevice(t, nbi, 1, nb.numONUPerOLT, &wg)
+
+ // Create the device with valid data
+ oltDevice, err := nbi.CreateDevice(getContext(), &voltha.Device{Type: nb.oltAdapterName, MacAddress: "aa:bb:cc:cc:ee:ee"})
+ assert.Nil(t, err)
+ assert.NotNil(t, oltDevice)
+
+ // Verify oltDevice exist in the core
+ devices, err := nbi.ListDevices(getContext(), &empty.Empty{})
+ assert.Nil(t, err)
+ assert.Equal(t, 1, len(devices.Items))
+ assert.Equal(t, oltDevice.Id, devices.Items[0].Id)
+
+ // Enable the oltDevice
+ _, err = nbi.EnableDevice(getContext(), &voltha.ID{Id: oltDevice.Id})
+ assert.Nil(t, err)
+
+ // Wait for the logical device to be in the ready state
+ var vldFunction isLogicalDeviceConditionSatisfied = func(ld *voltha.LogicalDevice) bool {
+ return ld != nil && len(ld.Ports) == nb.numONUPerOLT+1
+ }
+ err = waitUntilLogicalDeviceReadiness(oltDevice.Id, nb.maxTimeout, nbi, vldFunction)
+ assert.Nil(t, err)
+
+ // Verify that the devices have been setup correctly
+ nb.verifyDevices(t, nbi)
+
+ // Get latest oltDevice data
+ oltDevice, err = nbi.GetDevice(getContext(), &voltha.ID{Id: oltDevice.Id})
+ assert.Nil(t, err)
+
+ // Verify that the logical device has been setup correctly
+ nb.verifyLogicalDevices(t, oltDevice, nbi)
+
+ // Wait until all flows has been sent to the devices successfully
+ wg.Wait()
+}
+
+func (nb *NBTest) testDisableAndReEnableRootDevice(t *testing.T, nbi *NBIHandler) {
+ //Get an OLT device
+ oltDevice, err := nb.getADevice(true, nbi)
+ assert.Nil(t, err)
+ assert.NotNil(t, oltDevice)
+
+ // Disable the oltDevice
+ _, err = nbi.DisableDevice(getContext(), &voltha.ID{Id: oltDevice.Id})
+ assert.Nil(t, err)
+
+ // Wait for the old device to be disabled
+ var vdFunction isDeviceConditionSatisfied = func(device *voltha.Device) bool {
+ return device.AdminState == voltha.AdminState_DISABLED && device.OperStatus == voltha.OperStatus_UNKNOWN
+ }
+ err = waitUntilDeviceReadiness(oltDevice.Id, nb.maxTimeout, vdFunction, nbi)
+ assert.Nil(t, err)
+
+ // Verify that all onu devices are disabled as well
+ onuDevices, err := nb.deviceMgr.GetAllChildDevices(getContext(), oltDevice.Id)
+ assert.Nil(t, err)
+ for _, onu := range onuDevices.Items {
+ err = waitUntilDeviceReadiness(onu.Id, nb.maxTimeout, vdFunction, nbi)
+ assert.Nil(t, err)
+ }
+
+ // Wait for the logical device to satisfy the expected condition
+ var vlFunction isLogicalDeviceConditionSatisfied = func(ld *voltha.LogicalDevice) bool {
+ if ld == nil {
+ return false
+ }
+ for _, lp := range ld.Ports {
+ if (lp.OfpPort.Config&uint32(ofp.OfpPortConfig_OFPPC_PORT_DOWN) != lp.OfpPort.Config) ||
+ lp.OfpPort.State != uint32(ofp.OfpPortState_OFPPS_LINK_DOWN) {
+ return false
+ }
+ }
+ return true
+ }
+ err = waitUntilLogicalDeviceReadiness(oltDevice.Id, nb.maxTimeout, nbi, vlFunction)
+ assert.Nil(t, err)
+
+ // Reenable the oltDevice
+ _, err = nbi.EnableDevice(getContext(), &voltha.ID{Id: oltDevice.Id})
+ assert.Nil(t, err)
+
+ // Wait for the old device to be enabled
+ vdFunction = func(device *voltha.Device) bool {
+ return device.AdminState == voltha.AdminState_ENABLED && device.OperStatus == voltha.OperStatus_ACTIVE
+ }
+ err = waitUntilDeviceReadiness(oltDevice.Id, nb.maxTimeout, vdFunction, nbi)
+ assert.Nil(t, err)
+
+ // Verify that all onu devices are enabled as well
+ onuDevices, err = nb.deviceMgr.GetAllChildDevices(getContext(), oltDevice.Id)
+ assert.Nil(t, err)
+ for _, onu := range onuDevices.Items {
+ err = waitUntilDeviceReadiness(onu.Id, nb.maxTimeout, vdFunction, nbi)
+ assert.Nil(t, err)
+ }
+
+ // Wait for the logical device to satisfy the expected condition
+ vlFunction = func(ld *voltha.LogicalDevice) bool {
+ if ld == nil {
+ return false
+ }
+ for _, lp := range ld.Ports {
+ if (lp.OfpPort.Config&^uint32(ofp.OfpPortConfig_OFPPC_PORT_DOWN) != lp.OfpPort.Config) ||
+ lp.OfpPort.State != uint32(ofp.OfpPortState_OFPPS_LIVE) {
+ return false
+ }
+ }
+ return true
+ }
+ err = waitUntilLogicalDeviceReadiness(oltDevice.Id, nb.maxTimeout, nbi, vlFunction)
+ assert.Nil(t, err)
+}
+
+func (nb *NBTest) testDisableAndDeleteAllDevice(t *testing.T, nbi *NBIHandler) {
+ //Get an OLT device
+ oltDevice, err := nb.getADevice(true, nbi)
+ assert.Nil(t, err)
+ assert.NotNil(t, oltDevice)
+
+ // Disable the oltDevice
+ _, err = nbi.DisableDevice(getContext(), &voltha.ID{Id: oltDevice.Id})
+ assert.Nil(t, err)
+
+ // Wait for the olt device to be disabled
+ var vdFunction isDeviceConditionSatisfied = func(device *voltha.Device) bool {
+ return device.AdminState == voltha.AdminState_DISABLED && device.OperStatus == voltha.OperStatus_UNKNOWN
+ }
+ err = waitUntilDeviceReadiness(oltDevice.Id, nb.maxTimeout, vdFunction, nbi)
+ assert.Nil(t, err)
+
+ // Verify that all onu devices are disabled as well
+ onuDevices, err := nb.deviceMgr.GetAllChildDevices(getContext(), oltDevice.Id)
+ assert.Nil(t, err)
+ for _, onu := range onuDevices.Items {
+ err = waitUntilDeviceReadiness(onu.Id, nb.maxTimeout, vdFunction, nbi)
+ assert.Nil(t, err)
+ }
+
+ // Delete the oltDevice
+ _, err = nbi.DeleteDevice(getContext(), &voltha.ID{Id: oltDevice.Id})
+ assert.Nil(t, err)
+
+ var vFunction isDevicesConditionSatisfied = func(devices *voltha.Devices) bool {
+ return devices != nil && len(devices.Items) == 0
+ }
+ err = waitUntilConditionForDevices(nb.maxTimeout, nbi, vFunction)
+ assert.Nil(t, err)
+
+ // Wait for absence of logical device
+ var vlFunction isLogicalDevicesConditionSatisfied = func(lds *voltha.LogicalDevices) bool {
+ return lds != nil && len(lds.Items) == 0
+ }
+
+ err = waitUntilConditionForLogicalDevices(nb.maxTimeout, nbi, vlFunction)
+ assert.Nil(t, err)
+}
+func (nb *NBTest) testEnableAndDeleteAllDevice(t *testing.T, nbi *NBIHandler) {
+ //Create the device with valid data
+ oltDevice, err := nbi.CreateDevice(getContext(), &voltha.Device{Type: nb.oltAdapterName, MacAddress: "aa:bb:cc:cc:ee:ee"})
+ assert.Nil(t, err)
+ assert.NotNil(t, oltDevice)
+
+ //Get an OLT device
+ oltDevice, err = nb.getADevice(true, nbi)
+ assert.Nil(t, err)
+ assert.NotNil(t, oltDevice)
+
+ // Enable the oltDevice
+ _, err = nbi.EnableDevice(getContext(), &voltha.ID{Id: oltDevice.Id})
+ assert.Nil(t, err)
+
+ // Wait for the logical device to be in the ready state
+ var vldFunction isLogicalDeviceConditionSatisfied = func(ld *voltha.LogicalDevice) bool {
+ return ld != nil && len(ld.Ports) == nb.numONUPerOLT+1
+ }
+ err = waitUntilLogicalDeviceReadiness(oltDevice.Id, nb.maxTimeout, nbi, vldFunction)
+ assert.Nil(t, err)
+
+ //Get all child devices
+ onuDevices, err := nb.deviceMgr.GetAllChildDevices(getContext(), oltDevice.Id)
+ assert.Nil(t, err)
+
+ // Wait for the all onu devices to be enabled
+ var vdFunction isDeviceConditionSatisfied = func(device *voltha.Device) bool {
+ return device.AdminState == voltha.AdminState_ENABLED
+ }
+ for _, onu := range onuDevices.Items {
+ err = waitUntilDeviceReadiness(onu.Id, nb.maxTimeout, vdFunction, nbi)
+ assert.Nil(t, err)
+ }
+ // Wait for each onu device to get deleted
+ var vdFunc isDeviceConditionSatisfied = func(device *voltha.Device) bool {
+ return device == nil
+ }
+
+ // Delete the onuDevice
+ for _, onu := range onuDevices.Items {
+ _, err = nbi.DeleteDevice(getContext(), &voltha.ID{Id: onu.Id})
+ assert.Nil(t, err)
+ err = waitUntilDeviceReadiness(onu.Id, nb.maxTimeout, vdFunc, nbi)
+ assert.Nil(t, err)
+ }
+
+ // Disable the oltDevice
+ _, err = nbi.DisableDevice(getContext(), &voltha.ID{Id: oltDevice.Id})
+ assert.Nil(t, err)
+
+ // Wait for the olt device to be disabled
+ var vFunction isDeviceConditionSatisfied = func(device *voltha.Device) bool {
+ return device.AdminState == voltha.AdminState_DISABLED && device.OperStatus == voltha.OperStatus_UNKNOWN
+ }
+ err = waitUntilDeviceReadiness(oltDevice.Id, nb.maxTimeout, vFunction, nbi)
+ assert.Nil(t, err)
+
+ // Delete the oltDevice
+ _, err = nbi.DeleteDevice(getContext(), &voltha.ID{Id: oltDevice.Id})
+ assert.Nil(t, err)
+
+ var vFunc isDevicesConditionSatisfied = func(devices *voltha.Devices) bool {
+ return devices != nil && len(devices.Items) == 0
+ }
+ err = waitUntilConditionForDevices(nb.maxTimeout, nbi, vFunc)
+ assert.Nil(t, err)
+}
+func (nb *NBTest) testDisableAndEnablePort(t *testing.T, nbi *NBIHandler) {
+ //Get an OLT device
+ var cp *voltha.Port
+ oltDevice, err := nb.getADevice(true, nbi)
+ assert.Nil(t, err)
+ assert.NotNil(t, oltDevice)
+
+ for _, cp = range oltDevice.Ports {
+ if cp.Type == voltha.Port_PON_OLT {
+ break
+ }
+
+ }
+ assert.NotNil(t, cp)
+ cp.DeviceId = oltDevice.Id
+
+ // Disable the NW Port of oltDevice
+ _, err = nbi.DisablePort(getContext(), cp)
+ assert.Nil(t, err)
+ // Wait for the olt device Port to be disabled
+ var vdFunction isDeviceConditionSatisfied = func(device *voltha.Device) bool {
+ for _, port := range device.Ports {
+ if port.PortNo == cp.PortNo {
+ return port.AdminState == voltha.AdminState_DISABLED
+ }
+ }
+ return false
+ }
+ err = waitUntilDeviceReadiness(oltDevice.Id, nb.maxTimeout, vdFunction, nbi)
+ assert.Nil(t, err)
+ // Wait for the logical device to satisfy the expected condition
+ var vlFunction = func(ld *voltha.LogicalDevice) bool {
+ if ld == nil {
+ return false
+ }
+ for _, lp := range ld.Ports {
+ if (lp.OfpPort.Config&^uint32(ofp.OfpPortConfig_OFPPC_PORT_DOWN) != lp.OfpPort.Config) ||
+ lp.OfpPort.State != uint32(ofp.OfpPortState_OFPPS_LIVE) {
+ return false
+ }
+ }
+ return true
+ }
+ err = waitUntilLogicalDeviceReadiness(oltDevice.Id, nb.maxTimeout, nbi, vlFunction)
+ assert.Nil(t, err)
+
+ // Enable the NW Port of oltDevice
+ _, err = nbi.EnablePort(getContext(), cp)
+ assert.Nil(t, err)
+
+ // Wait for the olt device Port to be enabled
+ vdFunction = func(device *voltha.Device) bool {
+ for _, port := range device.Ports {
+ if port.PortNo == cp.PortNo {
+ return port.AdminState == voltha.AdminState_ENABLED
+ }
+ }
+ return false
+ }
+ err = waitUntilDeviceReadiness(oltDevice.Id, nb.maxTimeout, vdFunction, nbi)
+ assert.Nil(t, err)
+ // Wait for the logical device to satisfy the expected condition
+ vlFunction = func(ld *voltha.LogicalDevice) bool {
+ if ld == nil {
+ return false
+ }
+ for _, lp := range ld.Ports {
+ if (lp.OfpPort.Config&^uint32(ofp.OfpPortConfig_OFPPC_PORT_DOWN) != lp.OfpPort.Config) ||
+ lp.OfpPort.State != uint32(ofp.OfpPortState_OFPPS_LIVE) {
+ return false
+ }
+ }
+ return true
+ }
+ err = waitUntilLogicalDeviceReadiness(oltDevice.Id, nb.maxTimeout, nbi, vlFunction)
+ assert.Nil(t, err)
+
+ // Disable a non-PON port
+ for _, cp = range oltDevice.Ports {
+ if cp.Type != voltha.Port_PON_OLT {
+ break
+ }
+
+ }
+ assert.NotNil(t, cp)
+ cp.DeviceId = oltDevice.Id
+
+ // Disable the NW Port of oltDevice
+ _, err = nbi.DisablePort(getContext(), cp)
+ assert.NotNil(t, err)
+
+}
+
+func (nb *NBTest) testDeviceRebootWhenOltIsEnabled(t *testing.T, nbi *NBIHandler) {
+ //Get an OLT device
+ oltDevice, err := nb.getADevice(true, nbi)
+ assert.Nil(t, err)
+ assert.NotNil(t, oltDevice)
+ assert.Equal(t, oltDevice.ConnectStatus, voltha.ConnectStatus_REACHABLE)
+ assert.Equal(t, oltDevice.AdminState, voltha.AdminState_ENABLED)
+
+ // Verify that we have one or more ONUs to start with
+ onuDevices, err := nb.deviceMgr.GetAllChildDevices(getContext(), oltDevice.Id)
+ assert.Nil(t, err)
+ assert.NotNil(t, onuDevices)
+ assert.Greater(t, len(onuDevices.Items), 0)
+
+ // Reboot the OLT and very that Connection Status goes to UNREACHABLE and operation status to UNKNOWN
+ _, err = nbi.RebootDevice(getContext(), &voltha.ID{Id: oltDevice.Id})
+ assert.Nil(t, err)
+
+ var vlFunction0 = func(d *voltha.Device) bool {
+ return d.ConnectStatus == voltha.ConnectStatus_UNREACHABLE && d.OperStatus == voltha.OperStatus_UNKNOWN
+ }
+
+ err = waitUntilDeviceReadiness(oltDevice.Id, nb.maxTimeout, vlFunction0, nbi)
+ assert.Nil(t, err)
+
+ // Wait for the logical device to satisfy the expected condition
+ var vlFunction1 = func(ld *voltha.LogicalDevice) bool {
+ return ld == nil
+ }
+
+ err = waitUntilLogicalDeviceReadiness(oltDevice.Id, nb.maxTimeout, nbi, vlFunction1)
+ assert.Nil(t, err)
+
+ // Wait for the device to satisfy the expected condition (device does not have flows)
+ var vlFunction2 = func(d *voltha.Device) bool {
+ var deviceFlows *ofp.Flows
+ var err error
+ if deviceFlows, err = nbi.ListDeviceFlows(getContext(), &voltha.ID{Id: d.Id}); err != nil {
+ return false
+ }
+ return len(deviceFlows.Items) == 0
+ }
+
+ err = waitUntilDeviceReadiness(oltDevice.Id, nb.maxTimeout, vlFunction2, nbi)
+ assert.Nil(t, err)
+
+ // Wait for the device to satisfy the expected condition (there are no child devices)
+ var vlFunction3 = func(d *voltha.Device) bool {
+ var devices *voltha.Devices
+ var err error
+ if devices, err = nbi.ListDevices(getContext(), nil); err != nil {
+ return false
+ }
+ for _, device := range devices.Items {
+ if device.ParentId == d.Id {
+ // We have a child device still left
+ return false
+ }
+ }
+ return true
+ }
+
+ err = waitUntilDeviceReadiness(oltDevice.Id, nb.maxTimeout, vlFunction3, nbi)
+ assert.Nil(t, err)
+
+ // Update the OLT Connection Status to REACHABLE and operation status to ACTIVE
+ // Normally, in a real adapter this happens after connection regain via a heartbeat mechanism with real hardware
+ err = nbi.deviceMgr.UpdateDeviceStatus(getContext(), oltDevice.Id, voltha.OperStatus_ACTIVE, voltha.ConnectStatus_REACHABLE)
+ assert.Nil(t, err)
+
+ // Verify the device connection and operation states
+ oltDevice, err = nb.getADevice(true, nbi)
+ assert.Nil(t, err)
+ assert.NotNil(t, oltDevice)
+ assert.Equal(t, oltDevice.ConnectStatus, voltha.ConnectStatus_REACHABLE)
+ assert.Equal(t, oltDevice.AdminState, voltha.AdminState_ENABLED)
+
+ // Wait for the logical device to satisfy the expected condition
+ var vlFunction4 = func(ld *voltha.LogicalDevice) bool {
+ return ld != nil
+ }
+ err = waitUntilLogicalDeviceReadiness(oltDevice.Id, nb.maxTimeout, nbi, vlFunction4)
+ assert.Nil(t, err)
+
+ // Verify that logical device is created again
+ logicalDevices, err := nbi.ListLogicalDevices(getContext(), &empty.Empty{})
+ assert.Nil(t, err)
+ assert.NotNil(t, logicalDevices)
+ assert.Equal(t, 1, len(logicalDevices.Items))
+
+ // Verify that we have no ONUs left
+ onuDevices, err = nb.deviceMgr.GetAllChildDevices(getContext(), oltDevice.Id)
+ assert.Nil(t, err)
+ assert.NotNil(t, onuDevices)
+ assert.Equal(t, 0, len(onuDevices.Items))
+}
+
+func (nb *NBTest) testStartOmciTestAction(t *testing.T, nbi *NBIHandler) {
+ // -----------------------------------------------------------------------
+ // SubTest 1: Omci test action should fail due to nonexistent device id
+
+ request := &voltha.OmciTestRequest{Id: "123", Uuid: "456"}
+ _, err := nbi.StartOmciTestAction(getContext(), request)
+ assert.NotNil(t, err)
+ assert.Equal(t, "rpc error: code = NotFound desc = 123", err.Error())
+
+ // -----------------------------------------------------------------------
+ // SubTest 2: Error should be returned for device with no adapter registered
+
+ // Create a device that has no adapter registered
+ deviceNoAdapter, err := nbi.CreateDevice(getContext(), &voltha.Device{Type: "noAdapterRegisteredOmciTest", MacAddress: "aa:bb:cc:cc:ee:01"})
+ assert.Nil(t, err)
+ assert.NotNil(t, deviceNoAdapter)
+
+ // Omci test action should fail due to nonexistent adapter
+ request = &voltha.OmciTestRequest{Id: deviceNoAdapter.Id, Uuid: "456"}
+ _, err = nbi.StartOmciTestAction(getContext(), request)
+ assert.NotNil(t, err)
+ assert.Equal(t, "Adapter-not-registered-for-device-type noAdapterRegisteredOmciTest", err.Error())
+
+ //Remove the device
+ _, err = nbi.DeleteDevice(getContext(), &voltha.ID{Id: deviceNoAdapter.Id})
+ assert.Nil(t, err)
+
+ //Ensure there are no devices in the Core now - wait until condition satisfied or timeout
+ var vFunction isDevicesConditionSatisfied = func(devices *voltha.Devices) bool {
+ return devices != nil && len(devices.Items) == 0
+ }
+ err = waitUntilConditionForDevices(nb.maxTimeout, nbi, vFunction)
+ assert.Nil(t, err)
+
+ // -----------------------------------------------------------------------
+ // SubTest 3: Omci test action should succeed on valid ONU
+
+ // Create the device with valid data
+ oltDevice, err := nbi.CreateDevice(getContext(), &voltha.Device{Type: nb.oltAdapterName, MacAddress: "aa:bb:cc:cc:ee:ee"})
+ assert.Nil(t, err)
+ assert.NotNil(t, oltDevice)
+
+ // Verify oltDevice exist in the core
+ devices, err := nbi.ListDevices(getContext(), &empty.Empty{})
+ assert.Nil(t, err)
+ assert.Equal(t, 1, len(devices.Items))
+ assert.Equal(t, oltDevice.Id, devices.Items[0].Id)
+
+ // Enable the oltDevice
+ _, err = nbi.EnableDevice(getContext(), &voltha.ID{Id: oltDevice.Id})
+ assert.Nil(t, err)
+
+ // Wait for the logical device to be in the ready state
+ var vldFunction isLogicalDeviceConditionSatisfied = func(ld *voltha.LogicalDevice) bool {
+ return ld != nil && len(ld.Ports) == nb.numONUPerOLT+1
+ }
+ err = waitUntilLogicalDeviceReadiness(oltDevice.Id, nb.maxTimeout, nbi, vldFunction)
+ assert.Nil(t, err)
+
+ // Wait for the olt device to be enabled
+ vdFunction := func(device *voltha.Device) bool {
+ return device.AdminState == voltha.AdminState_ENABLED && device.OperStatus == voltha.OperStatus_ACTIVE
+ }
+ err = waitUntilDeviceReadiness(oltDevice.Id, nb.maxTimeout, vdFunction, nbi)
+ assert.Nil(t, err)
+
+ onuDevices, err := nb.deviceMgr.GetAllChildDevices(getContext(), oltDevice.Id)
+ assert.Nil(t, err)
+ assert.Greater(t, len(onuDevices.Items), 0)
+
+ onuDevice := onuDevices.Items[0]
+
+ // Omci test action should succeed
+ request = &voltha.OmciTestRequest{Id: onuDevice.Id, Uuid: "456"}
+ resp, err := nbi.StartOmciTestAction(getContext(), request)
+ assert.Nil(t, err)
+ assert.Equal(t, resp.Result, voltha.TestResponse_SUCCESS)
+
+ //Remove the device
+ _, err = nbi.DeleteDevice(getContext(), &voltha.ID{Id: oltDevice.Id})
+ assert.Nil(t, err)
+ //Ensure there are no devices in the Core now - wait until condition satisfied or timeout
+ err = waitUntilConditionForDevices(nb.maxTimeout, nbi, vFunction)
+ assert.Nil(t, err)
+}
+
+func makeSimpleFlowMod(fa *flows.FlowArgs) *ofp.OfpFlowMod {
+ matchFields := make([]*ofp.OfpOxmField, 0)
+ for _, val := range fa.MatchFields {
+ matchFields = append(matchFields, &ofp.OfpOxmField{Field: &ofp.OfpOxmField_OfbField{OfbField: val}})
+ }
+ return flows.MkSimpleFlowMod(matchFields, fa.Actions, fa.Command, fa.KV)
+}
+
+func createMetadata(cTag int, techProfile int, port int) uint64 {
+ md := 0
+ md = (md | (cTag & 0xFFFF)) << 16
+ md = (md | (techProfile & 0xFFFF)) << 32
+ return uint64(md | (port & 0xFFFFFFFF))
+}
+
+func (nb *NBTest) verifyLogicalDeviceFlowCount(t *testing.T, nbi *NBIHandler, numNNIPorts int, numUNIPorts int) {
+ expectedNumFlows := numNNIPorts*3 + numNNIPorts*numUNIPorts
+ // Wait for logical device to have all the flows
+ var vlFunction isLogicalDevicesConditionSatisfied = func(lds *voltha.LogicalDevices) bool {
+ return lds != nil && len(lds.Items) == 1 && len(lds.Items[0].Flows.Items) == expectedNumFlows
+ }
+ // No timeout implies a success
+ err := waitUntilConditionForLogicalDevices(nb.maxTimeout, nbi, vlFunction)
+ assert.Nil(t, err)
+}
+
+func (nb *NBTest) sendTrapFlows(t *testing.T, nbi *NBIHandler, logicalDevice *voltha.LogicalDevice, meterID uint64, startingVlan int) (numNNIPorts, numUNIPorts int) {
+ // Send flows for the parent device
+ var nniPorts []*voltha.LogicalPort
+ var uniPorts []*voltha.LogicalPort
+ for _, p := range logicalDevice.Ports {
+ if p.RootPort {
+ nniPorts = append(nniPorts, p)
+ } else {
+ uniPorts = append(uniPorts, p)
+ }
+ }
+ assert.Equal(t, 1, len(nniPorts))
+ //assert.Greater(t, len(uniPorts), 1 )
+ nniPort := nniPorts[0].OfpPort.PortNo
+ maxInt32 := uint64(0xFFFFFFFF)
+ controllerPortMask := uint32(4294967293) // will result in 4294967293&0x7fffffff => 2147483645 which is the actual controller port
+ var fa *flows.FlowArgs
+ fa = &flows.FlowArgs{
+ KV: flows.OfpFlowModArgs{"priority": 10000, "buffer_id": maxInt32, "out_port": maxInt32, "out_group": maxInt32, "flags": 1},
+ MatchFields: []*ofp.OfpOxmOfbField{
+ flows.InPort(nniPort),
+ flows.EthType(35020),
+ },
+ Actions: []*ofp.OfpAction{
+ flows.Output(controllerPortMask),
+ },
+ }
+ flowLLDP := ofp.FlowTableUpdate{FlowMod: makeSimpleFlowMod(fa), Id: logicalDevice.Id}
+ _, err := nbi.UpdateLogicalDeviceFlowTable(getContext(), &flowLLDP)
+ assert.Nil(t, err)
+
+ fa = &flows.FlowArgs{
+ KV: flows.OfpFlowModArgs{"priority": 10000, "buffer_id": maxInt32, "out_port": maxInt32, "out_group": maxInt32, "flags": 1},
+ MatchFields: []*ofp.OfpOxmOfbField{
+ flows.InPort(nniPort),
+ flows.EthType(2048),
+ flows.IpProto(17),
+ flows.UdpSrc(67),
+ flows.UdpDst(68),
+ },
+ Actions: []*ofp.OfpAction{
+ flows.Output(controllerPortMask),
+ },
+ }
+ flowIPV4 := ofp.FlowTableUpdate{FlowMod: makeSimpleFlowMod(fa), Id: logicalDevice.Id}
+ _, err = nbi.UpdateLogicalDeviceFlowTable(getContext(), &flowIPV4)
+ assert.Nil(t, err)
+
+ fa = &flows.FlowArgs{
+ KV: flows.OfpFlowModArgs{"priority": 10000, "buffer_id": maxInt32, "out_port": maxInt32, "out_group": maxInt32, "flags": 1},
+ MatchFields: []*ofp.OfpOxmOfbField{
+ flows.InPort(nniPort),
+ flows.EthType(34525),
+ flows.IpProto(17),
+ flows.UdpSrc(546),
+ flows.UdpDst(547),
+ },
+ Actions: []*ofp.OfpAction{
+ flows.Output(controllerPortMask),
+ },
+ }
+ flowIPV6 := ofp.FlowTableUpdate{FlowMod: makeSimpleFlowMod(fa), Id: logicalDevice.Id}
+ _, err = nbi.UpdateLogicalDeviceFlowTable(getContext(), &flowIPV6)
+ assert.Nil(t, err)
+
+ return len(nniPorts), len(uniPorts)
+}
+
+func (nb *NBTest) sendEAPFlows(t *testing.T, nbi *NBIHandler, logicalDeviceID string, port *ofp.OfpPort, vlan int, meterID uint64) {
+ maxInt32 := uint64(0xFFFFFFFF)
+ controllerPortMask := uint32(4294967293) // will result in 4294967293&0x7fffffff => 2147483645 which is the actual controller port
+ fa := &flows.FlowArgs{
+ KV: flows.OfpFlowModArgs{"priority": 10000, "buffer_id": maxInt32, "out_port": maxInt32, "out_group": maxInt32, "flags": 1, "write_metadata": createMetadata(vlan, 64, 0), "meter_id": meterID},
+ MatchFields: []*ofp.OfpOxmOfbField{
+ flows.InPort(port.PortNo),
+ flows.EthType(34958),
+ flows.VlanVid(8187),
+ },
+ Actions: []*ofp.OfpAction{
+ flows.Output(controllerPortMask),
+ },
+ }
+ flowEAP := ofp.FlowTableUpdate{FlowMod: makeSimpleFlowMod(fa), Id: logicalDeviceID}
+ _, err := nbi.UpdateLogicalDeviceFlowTable(getContext(), &flowEAP)
+ assert.Nil(t, err)
+}
+
+func (nb *NBTest) monitorLogicalDevice(t *testing.T, nbi *NBIHandler, numNNIPorts int, numUNIPorts int, wg *sync.WaitGroup) {
+ defer wg.Done()
+ nb.logicalDeviceMgr.SetEventCallbacks(nbi)
+
+ // Clear any existing flows on the adapters
+ nb.oltAdapter.ClearFlows()
+ nb.onuAdapter.ClearFlows()
+
+ // Wait until a logical device is ready
+ var vlFunction isLogicalDevicesConditionSatisfied = func(lds *voltha.LogicalDevices) bool {
+ if lds == nil || len(lds.Items) != 1 {
+ return false
+ }
+ // Ensure there are both NNI ports and at least one UNI port on the logical device
+ ld := lds.Items[0]
+ nniPort := false
+ uniPort := false
+ for _, p := range ld.Ports {
+ nniPort = nniPort || p.RootPort == true
+ uniPort = uniPort || p.RootPort == false
+ if nniPort && uniPort {
+ return true
+ }
+ }
+ return false
+ }
+ err := waitUntilConditionForLogicalDevices(nb.maxTimeout, nbi, vlFunction)
+ assert.Nil(t, err)
+
+ logicalDevices, err := nbi.ListLogicalDevices(getContext(), &empty.Empty{})
+ assert.Nil(t, err)
+ assert.NotNil(t, logicalDevices)
+ assert.Equal(t, 1, len(logicalDevices.Items))
+
+ logicalDevice := logicalDevices.Items[0]
+ meterID := rand.Uint32()
+
+ // Add a meter to the logical device
+ meterMod := &ofp.OfpMeterMod{
+ Command: ofp.OfpMeterModCommand_OFPMC_ADD,
+ Flags: rand.Uint32(),
+ MeterId: meterID,
+ Bands: []*ofp.OfpMeterBandHeader{
+ {Type: ofp.OfpMeterBandType_OFPMBT_EXPERIMENTER,
+ Rate: rand.Uint32(),
+ BurstSize: rand.Uint32(),
+ Data: nil,
+ },
+ },
+ }
+ _, err = nbi.UpdateLogicalDeviceMeterTable(getContext(), &ofp.MeterModUpdate{Id: logicalDevice.Id, MeterMod: meterMod})
+ assert.Nil(t, err)
+
+ // Send initial set of Trap flows
+ startingVlan := 4091
+ nb.sendTrapFlows(t, nbi, logicalDevice, uint64(meterID), startingVlan)
+
+ // Listen for port events
+ start := time.Now()
+ processedNniLogicalPorts := 0
+ processedUniLogicalPorts := 0
+
+ for event := range nbi.changeEventQueue {
+ startingVlan++
+ if portStatus, ok := (event.Event).(*ofp.ChangeEvent_PortStatus); ok {
+ ps := portStatus.PortStatus
+ if ps.Reason == ofp.OfpPortReason_OFPPR_ADD {
+ if ps.Desc.PortNo >= uint32(nb.startingUNIPortNo) {
+ processedUniLogicalPorts++
+ nb.sendEAPFlows(t, nbi, logicalDevice.Id, ps.Desc, startingVlan, uint64(meterID))
+ } else {
+ processedNniLogicalPorts++
+ }
+ }
+ }
+
+ if processedNniLogicalPorts >= numNNIPorts && processedUniLogicalPorts >= numUNIPorts {
+ fmt.Println("Total time to send all flows:", time.Since(start))
+ break
+ }
+ }
+ //Verify the flow count on the logical device
+ nb.verifyLogicalDeviceFlowCount(t, nbi, numNNIPorts, numUNIPorts)
+
+ // Wait until all flows have been sent to the OLT adapters
+ var oltVFunc isConditionSatisfied = func() bool {
+ return nb.oltAdapter.GetFlowCount() >= (numNNIPorts*3)+numNNIPorts*numUNIPorts
+ }
+ err = waitUntilCondition(nb.maxTimeout, nbi, oltVFunc)
+ assert.Nil(t, err)
+
+ // Wait until all flows have been sent to the ONU adapters
+ var onuVFunc isConditionSatisfied = func() bool {
+ return nb.onuAdapter.GetFlowCount() == numUNIPorts
+ }
+ err = waitUntilCondition(nb.maxTimeout, nbi, onuVFunc)
+ assert.Nil(t, err)
+}
+
+func TestSuiteNbiApiHandler(t *testing.T) {
+ f, err := os.Create("../../../tests/results/profile.cpu")
+ if err != nil {
+ logger.Fatalf("could not create CPU profile: %v\n ", err)
+ }
+ defer f.Close()
+ runtime.SetBlockProfileRate(1)
+ runtime.SetMutexProfileFraction(-1)
+ if err := pprof.StartCPUProfile(f); err != nil {
+ logger.Fatalf("could not start CPU profile: %v\n", err)
+ }
+ defer pprof.StopCPUProfile()
+
+ //log.SetPackageLogLevel("github.com/opencord/voltha-go/rw_core/core", log.DebugLevel)
+
+ nb := newNBTest()
+ assert.NotNil(t, nb)
+
+ defer nb.stopAll()
+
+ // Start the Core
+ nb.startCore(false)
+
+ // Set the grpc API interface - no grpc server is running in unit test
+ nbi := NewAPIHandler(nb.deviceMgr, nb.logicalDeviceMgr, nb.adapterMgr)
+
+ // 1. Basic test with no data in Core
+ nb.testCoreWithoutData(t, nbi)
+
+ // Create/register the adapters
+ nb.createAndregisterAdapters(t)
+
+ // 2. Test adapter registration
+ nb.testAdapterRegistration(t, nbi)
+
+ numberOfDeviceTestRuns := 2
+ for i := 1; i <= numberOfDeviceTestRuns; i++ {
+ //3. Test create device
+ nb.testCreateDevice(t, nbi)
+
+ // 4. Test Enable a device
+ nb.testEnableDevice(t, nbi)
+
+ // 5. Test disable and ReEnable a root device
+ nb.testDisableAndReEnableRootDevice(t, nbi)
+
+ // 6. Test disable and Enable pon port of OLT device
+ nb.testDisableAndEnablePort(t, nbi)
+
+ // 7.Test Device unreachable when OLT is enabled
+ nb.testDeviceRebootWhenOltIsEnabled(t, nbi)
+
+ // 8. Test disable and delete all devices
+ nb.testDisableAndDeleteAllDevice(t, nbi)
+
+ // 9. Test enable and delete all devices
+ nb.testEnableAndDeleteAllDevice(t, nbi)
+
+ // 10. Test omci test
+ nb.testStartOmciTestAction(t, nbi)
+ }
+
+ //x. TODO - More tests to come
+}