[VOL-2318] - Fix for jenkins NBI Failure
This commit adds the latest devices and logical devices to the
device agents and logical device agents respectively. Any GET
is returned directly from these agents instead from the model.
And any create/update results in the data being sent to the KV
store via the model and also results in the latest data in the
agent being updated. If the Core dies and restart then the
latest data will be pulled from KV. These changes assumes
that a device or a logical device is always owned by one Core
only which is the case.
Change-Id: Ie671cd70b38a58a3b32fa476eced5f218aeadad9
diff --git a/go.mod b/go.mod
index 4233a21..90a586a 100644
--- a/go.mod
+++ b/go.mod
@@ -8,7 +8,7 @@
github.com/golang/protobuf v1.3.2
github.com/google/uuid v1.1.1
github.com/gyuho/goraph v0.0.0-20160328020532-d460590d53a9
- github.com/opencord/voltha-lib-go/v2 v2.2.18
+ github.com/opencord/voltha-lib-go/v2 v2.2.21
github.com/opencord/voltha-protos/v2 v2.1.0
github.com/phayes/freeport v0.0.0-20180830031419-95f893ade6f2
github.com/stretchr/testify v1.4.0
diff --git a/go.sum b/go.sum
index ae79e02..00a1590 100644
--- a/go.sum
+++ b/go.sum
@@ -192,8 +192,8 @@
github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
github.com/onsi/gomega v1.4.2 h1:3mYCb7aPxS/RU7TI1y4rkEn1oKmPRjNJLNEXgw7MH2I=
github.com/onsi/gomega v1.4.2/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
-github.com/opencord/voltha-lib-go/v2 v2.2.18 h1:rie20b5QNhRjh9GL4THHdGiHSUqrx4VMEJw44hZI1Mk=
-github.com/opencord/voltha-lib-go/v2 v2.2.18/go.mod h1:vjd6PeC3vOYDE/ztMjraWFB2Wv1oaPcum1Xz4RJbSoE=
+github.com/opencord/voltha-lib-go/v2 v2.2.21 h1:QACE9X0h+ZPf1F3Sb+lNmvzWrWizCcRMiYcq7Wm2rTQ=
+github.com/opencord/voltha-lib-go/v2 v2.2.21/go.mod h1:CoY2amUEsbO2grCbJRk7G+Fl1Xb7vQLw3/uGLbTz0Ms=
github.com/opencord/voltha-protos/v2 v2.1.0 h1:Ppl4/3OBwgGuLk0ob9vIEwMGGRC2sqe7WWoxh0Uq/n0=
github.com/opencord/voltha-protos/v2 v2.1.0/go.mod h1:6kOcfYi1CadWowFxI2SH5wLfHrsRECZLZlD2MFK6WDI=
github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
diff --git a/rw_core/core/device_agent.go b/rw_core/core/device_agent.go
index 1cc1673..06ba307 100755
--- a/rw_core/core/device_agent.go
+++ b/rw_core/core/device_agent.go
@@ -49,6 +49,7 @@
deviceProxy *model.Proxy
exitChannel chan int
lockDevice sync.RWMutex
+ device *voltha.Device
defaultTimeout int64
}
@@ -71,6 +72,7 @@
agent.clusterDataProxy = cdProxy
agent.lockDevice = sync.RWMutex{}
agent.defaultTimeout = timeout
+ agent.device = proto.Clone(device).(*voltha.Device)
return &agent
}
@@ -90,6 +92,7 @@
var ok bool
if device, ok = loadedDevice.(*voltha.Device); ok {
agent.deviceType = device.Adapter
+ agent.device = proto.Clone(device).(*voltha.Device)
} else {
log.Errorw("failed-to-convert-device", log.Fields{"deviceId": agent.deviceID})
return nil, status.Errorf(codes.NotFound, "device-%s", agent.deviceID)
@@ -120,6 +123,7 @@
log.Errorw("failed-to-add-device", log.Fields{"deviceId": agent.deviceID})
return nil, status.Errorf(codes.Aborted, "failed-adding-device-%s", agent.deviceID)
}
+ agent.device = proto.Clone(device).(*voltha.Device)
}
agent.deviceProxy = agent.clusterDataProxy.CreateProxy(ctx, "/devices/"+agent.deviceID, false)
@@ -134,6 +138,10 @@
agent.lockDevice.Lock()
defer agent.lockDevice.Unlock()
log.Debug("stopping-device-agent")
+
+ // First unregister any callbacks
+ agent.deviceProxy.UnregisterCallback(model.POST_UPDATE, agent.processUpdate)
+
// Remove the device from the KV store
if removed := agent.clusterDataProxy.Remove(ctx, "/devices/"+agent.deviceID, ""); removed == nil {
log.Debugw("device-already-removed", log.Fields{"id": agent.deviceID})
@@ -152,34 +160,22 @@
if device := agent.clusterDataProxy.Get(context.Background(), "/devices/"+agent.deviceID, 1, true, ""); device != nil {
if d, ok := device.(*voltha.Device); ok {
agent.deviceType = d.Adapter
+ agent.device = proto.Clone(d).(*voltha.Device)
log.Debugw("reconciled-device-agent-devicetype", log.Fields{"Id": agent.deviceID, "type": agent.deviceType})
}
}
}
-// GetDevice retrieves the latest device information from the data model
-func (agent *DeviceAgent) getDevice() (*voltha.Device, error) {
+// getDevice returns the device data from cache
+func (agent *DeviceAgent) getDevice() *voltha.Device {
agent.lockDevice.RLock()
defer agent.lockDevice.RUnlock()
- if device := agent.clusterDataProxy.Get(context.Background(), "/devices/"+agent.deviceID, 0, false, ""); device != nil {
- if d, ok := device.(*voltha.Device); ok {
- cloned := proto.Clone(d).(*voltha.Device)
- return cloned, nil
- }
- }
- return nil, status.Errorf(codes.NotFound, "device-%s", agent.deviceID)
+ return proto.Clone(agent.device).(*voltha.Device)
}
// getDeviceWithoutLock is a helper function to be used ONLY by any device agent function AFTER it has acquired the device lock.
-// This function is meant so that we do not have duplicate code all over the device agent functions
-func (agent *DeviceAgent) getDeviceWithoutLock() (*voltha.Device, error) {
- if device := agent.clusterDataProxy.Get(context.Background(), "/devices/"+agent.deviceID, 0, false, ""); device != nil {
- if d, ok := device.(*voltha.Device); ok {
- cloned := proto.Clone(d).(*voltha.Device)
- return cloned, nil
- }
- }
- return nil, status.Errorf(codes.NotFound, "device-%s", agent.deviceID)
+func (agent *DeviceAgent) getDeviceWithoutLock() *voltha.Device {
+ return proto.Clone(agent.device).(*voltha.Device)
}
// enableDevice activates a preprovisioned or a disable device
@@ -188,37 +184,34 @@
defer agent.lockDevice.Unlock()
log.Debugw("enableDevice", log.Fields{"id": agent.deviceID})
- device, err := agent.getDeviceWithoutLock()
- if err != nil {
- return status.Errorf(codes.NotFound, "%s", agent.deviceID)
- }
+ cloned := agent.getDeviceWithoutLock()
+
// First figure out which adapter will handle this device type. We do it at this stage as allow devices to be
// pre-provisionned with the required adapter not registered. At this stage, since we need to communicate
// with the adapter then we need to know the adapter that will handle this request
- adapterName, err := agent.adapterMgr.getAdapterName(device.Type)
+ adapterName, err := agent.adapterMgr.getAdapterName(cloned.Type)
if err != nil {
- log.Warnw("no-adapter-registered-for-device-type", log.Fields{"deviceType": device.Type, "deviceAdapter": device.Adapter})
+ log.Warnw("no-adapter-registered-for-device-type", log.Fields{"deviceType": cloned.Type, "deviceAdapter": cloned.Adapter})
return err
}
- device.Adapter = adapterName
+ cloned.Adapter = adapterName
- if device.AdminState == voltha.AdminState_ENABLED {
+ if cloned.AdminState == voltha.AdminState_ENABLED {
log.Debugw("device-already-enabled", log.Fields{"id": agent.deviceID})
return nil
}
- if device.AdminState == voltha.AdminState_DELETED {
+ if cloned.AdminState == voltha.AdminState_DELETED {
// This is a temporary state when a device is deleted before it gets removed from the model.
- err = status.Error(codes.FailedPrecondition, fmt.Sprintf("cannot-enable-a-deleted-device: %s ", device.Id))
- log.Warnw("invalid-state", log.Fields{"id": agent.deviceID, "state": device.AdminState, "error": err})
+ err = status.Error(codes.FailedPrecondition, fmt.Sprintf("cannot-enable-a-deleted-device: %s ", cloned.Id))
+ log.Warnw("invalid-state", log.Fields{"id": agent.deviceID, "state": cloned.AdminState, "error": err})
return err
}
- previousAdminState := device.AdminState
+ previousAdminState := cloned.AdminState
// Update the Admin State and set the operational state to activating before sending the request to the
// Adapters
- cloned := proto.Clone(device).(*voltha.Device)
cloned.AdminState = voltha.AdminState_ENABLED
cloned.OperStatus = voltha.OperStatus_ACTIVATING
@@ -227,6 +220,7 @@
}
// Adopt the device if it was in preprovision state. In all other cases, try to reenable it.
+ device := proto.Clone(cloned).(*voltha.Device)
if previousAdminState == voltha.AdminState_PREPROVISIONED {
if err := agent.adapterProxy.AdoptDevice(ctx, device); err != nil {
log.Debugw("adoptDevice-error", log.Fields{"id": agent.deviceID, "error": err})
@@ -268,12 +262,7 @@
agent.lockDevice.Lock()
defer agent.lockDevice.Unlock()
- var device *voltha.Device
- var err error
- if device, err = agent.getDeviceWithoutLock(); err != nil {
- return coreutils.DoneResponse(), status.Errorf(codes.NotFound, "%s", agent.deviceID)
- }
-
+ device := agent.getDeviceWithoutLock()
existingFlows := proto.Clone(device.Flows).(*voltha.Flows)
existingGroups := proto.Clone(device.FlowGroups).(*ofp.FlowGroups)
@@ -371,12 +360,7 @@
agent.lockDevice.Lock()
defer agent.lockDevice.Unlock()
- var device *voltha.Device
- var err error
-
- if device, err = agent.getDeviceWithoutLock(); err != nil {
- return coreutils.DoneResponse(), status.Errorf(codes.NotFound, "%s", agent.deviceID)
- }
+ device := agent.getDeviceWithoutLock()
existingFlows := proto.Clone(device.Flows).(*voltha.Flows)
existingGroups := proto.Clone(device.FlowGroups).(*ofp.FlowGroups)
@@ -469,11 +453,9 @@
agent.lockDevice.Lock()
defer agent.lockDevice.Unlock()
- var device *voltha.Device
- var err error
- if device, err = agent.getDeviceWithoutLock(); err != nil {
- return coreutils.DoneResponse(), status.Errorf(codes.NotFound, "%s", agent.deviceID)
- }
+
+ device := agent.getDeviceWithoutLock()
+
existingFlows := proto.Clone(device.Flows).(*voltha.Flows)
existingGroups := proto.Clone(device.FlowGroups).(*ofp.FlowGroups)
@@ -580,30 +562,26 @@
agent.lockDevice.Lock()
defer agent.lockDevice.Unlock()
log.Debugw("disableDevice", log.Fields{"id": agent.deviceID})
- // Get the most up to date the device info
- device, err := agent.getDeviceWithoutLock()
- if err != nil {
- return status.Errorf(codes.NotFound, "%s", agent.deviceID)
- }
- if device.AdminState == voltha.AdminState_DISABLED {
+
+ cloned := agent.getDeviceWithoutLock()
+
+ if cloned.AdminState == voltha.AdminState_DISABLED {
log.Debugw("device-already-disabled", log.Fields{"id": agent.deviceID})
return nil
}
- if device.AdminState == voltha.AdminState_PREPROVISIONED ||
- device.AdminState == voltha.AdminState_DELETED {
+ if cloned.AdminState == voltha.AdminState_PREPROVISIONED ||
+ cloned.AdminState == voltha.AdminState_DELETED {
log.Debugw("device-not-enabled", log.Fields{"id": agent.deviceID})
- return status.Errorf(codes.FailedPrecondition, "deviceId:%s, invalid-admin-state:%s", agent.deviceID, device.AdminState)
+ return status.Errorf(codes.FailedPrecondition, "deviceId:%s, invalid-admin-state:%s", agent.deviceID, cloned.AdminState)
}
// Update the Admin State and operational state before sending the request out
- cloned := proto.Clone(device).(*voltha.Device)
cloned.AdminState = voltha.AdminState_DISABLED
cloned.OperStatus = voltha.OperStatus_UNKNOWN
if err := agent.updateDeviceInStoreWithoutLock(cloned, false, ""); err != nil {
return err
}
-
- if err := agent.adapterProxy.DisableDevice(ctx, device); err != nil {
+ if err := agent.adapterProxy.DisableDevice(ctx, proto.Clone(cloned).(*voltha.Device)); err != nil {
log.Debugw("disableDevice-error", log.Fields{"id": agent.deviceID, "error": err})
return err
}
@@ -614,17 +592,14 @@
agent.lockDevice.Lock()
defer agent.lockDevice.Unlock()
log.Debugw("updateAdminState", log.Fields{"id": agent.deviceID})
- // Get the most up to date the device info
- device, err := agent.getDeviceWithoutLock()
- if err != nil {
- return status.Errorf(codes.NotFound, "%s", agent.deviceID)
- }
- if device.AdminState == adminState {
+
+ cloned := agent.getDeviceWithoutLock()
+
+ if cloned.AdminState == adminState {
log.Debugw("no-change-needed", log.Fields{"id": agent.deviceID, "state": adminState})
return nil
}
// Received an Ack (no error found above). Now update the device in the model to the expected state
- cloned := proto.Clone(device).(*voltha.Device)
cloned.AdminState = adminState
if err := agent.updateDeviceInStoreWithoutLock(cloned, false, ""); err != nil {
return err
@@ -636,11 +611,8 @@
agent.lockDevice.Lock()
defer agent.lockDevice.Unlock()
log.Debugw("rebootDevice", log.Fields{"id": agent.deviceID})
- // Get the most up to date the device info
- device, err := agent.getDeviceWithoutLock()
- if err != nil {
- return status.Errorf(codes.NotFound, "%s", agent.deviceID)
- }
+
+ device := agent.getDeviceWithoutLock()
if err := agent.adapterProxy.RebootDevice(ctx, device); err != nil {
log.Debugw("rebootDevice-error", log.Fields{"id": agent.deviceID, "error": err})
return err
@@ -652,39 +624,35 @@
agent.lockDevice.Lock()
defer agent.lockDevice.Unlock()
log.Debugw("deleteDevice", log.Fields{"id": agent.deviceID})
- // Get the most up to date the device info
- device, err := agent.getDeviceWithoutLock()
- if err != nil {
- return status.Errorf(codes.NotFound, "%s", agent.deviceID)
- }
- if device.AdminState == voltha.AdminState_DELETED {
+
+ cloned := agent.getDeviceWithoutLock()
+ if cloned.AdminState == voltha.AdminState_DELETED {
log.Debugw("device-already-in-deleted-state", log.Fields{"id": agent.deviceID})
return nil
}
- if (device.AdminState != voltha.AdminState_DISABLED) &&
- (device.AdminState != voltha.AdminState_PREPROVISIONED) {
+ if (cloned.AdminState != voltha.AdminState_DISABLED) &&
+ (cloned.AdminState != voltha.AdminState_PREPROVISIONED) {
log.Debugw("device-not-disabled", log.Fields{"id": agent.deviceID})
//TODO: Needs customized error message
return status.Errorf(codes.FailedPrecondition, "deviceId:%s, expected-admin-state:%s", agent.deviceID, voltha.AdminState_DISABLED)
}
- if device.AdminState != voltha.AdminState_PREPROVISIONED {
+ if cloned.AdminState != voltha.AdminState_PREPROVISIONED {
// Send the request to an Adapter only if the device is not in poreporovision state and wait for a response
- if err := agent.adapterProxy.DeleteDevice(ctx, device); err != nil {
+ if err := agent.adapterProxy.DeleteDevice(ctx, cloned); err != nil {
log.Debugw("deleteDevice-error", log.Fields{"id": agent.deviceID, "error": err})
return err
}
}
// Set the state to deleted after we receive an Ack - this will trigger some background process to clean up
// the device as well as its association with the logical device
- cloned := proto.Clone(device).(*voltha.Device)
cloned.AdminState = voltha.AdminState_DELETED
if err := agent.updateDeviceInStoreWithoutLock(cloned, false, ""); err != nil {
return err
}
// If this is a child device then remove the associated peer ports on the parent device
- if !device.Root {
+ if !cloned.Root {
go func() {
- err := agent.deviceMgr.deletePeerPorts(device.ParentId, device.Id)
+ err := agent.deviceMgr.deletePeerPorts(cloned.ParentId, cloned.Id)
if err != nil {
log.Errorw("unable-to-delete-peer-ports", log.Fields{"error": err})
}
@@ -697,12 +665,8 @@
agent.lockDevice.Lock()
defer agent.lockDevice.Unlock()
log.Debugw("setParentId", log.Fields{"deviceId": device.Id, "parentId": parentID})
- storeDevice, err := agent.getDeviceWithoutLock()
- if err != nil {
- return status.Errorf(codes.NotFound, "%s", agent.deviceID)
- }
- // clone the device
- cloned := proto.Clone(storeDevice).(*voltha.Device)
+
+ cloned := agent.getDeviceWithoutLock()
cloned.ParentId = parentID
// Store the device
if err := agent.updateDeviceInStoreWithoutLock(cloned, false, ""); err != nil {
@@ -715,13 +679,8 @@
agent.lockDevice.Lock()
defer agent.lockDevice.Unlock()
log.Debugw("updatePmConfigs", log.Fields{"id": pmConfigs.Id})
- // Work only on latest data
- storeDevice, err := agent.getDeviceWithoutLock()
- if err != nil {
- return status.Errorf(codes.NotFound, "%s", agent.deviceID)
- }
- // clone the device
- cloned := proto.Clone(storeDevice).(*voltha.Device)
+
+ cloned := agent.getDeviceWithoutLock()
cloned.PmConfigs = proto.Clone(pmConfigs).(*voltha.PmConfigs)
// Store the device
if err := agent.updateDeviceInStoreWithoutLock(cloned, false, ""); err != nil {
@@ -739,13 +698,8 @@
agent.lockDevice.Lock()
defer agent.lockDevice.Unlock()
log.Debugw("initPmConfigs", log.Fields{"id": pmConfigs.Id})
- // Work only on latest data
- storeDevice, err := agent.getDeviceWithoutLock()
- if err != nil {
- return status.Errorf(codes.NotFound, "%s", agent.deviceID)
- }
- // clone the device
- cloned := proto.Clone(storeDevice).(*voltha.Device)
+
+ cloned := agent.getDeviceWithoutLock()
cloned.PmConfigs = proto.Clone(pmConfigs).(*voltha.PmConfigs)
// Store the device
updateCtx := context.WithValue(context.Background(), model.RequestTimestamp, time.Now().UnixNano())
@@ -760,24 +714,17 @@
agent.lockDevice.RLock()
defer agent.lockDevice.RUnlock()
log.Debugw("listPmConfigs", log.Fields{"id": agent.deviceID})
- // Get the most up to date the device info
- device, err := agent.getDeviceWithoutLock()
- if err != nil {
- return nil, status.Errorf(codes.NotFound, "%s", agent.deviceID)
- }
- cloned := proto.Clone(device).(*voltha.Device)
- return cloned.PmConfigs, nil
+
+ return agent.getDeviceWithoutLock().PmConfigs, nil
}
func (agent *DeviceAgent) downloadImage(ctx context.Context, img *voltha.ImageDownload) (*voltha.OperationResp, error) {
agent.lockDevice.Lock()
defer agent.lockDevice.Unlock()
log.Debugw("downloadImage", log.Fields{"id": agent.deviceID})
- // Get the most up to date the device info
- device, err := agent.getDeviceWithoutLock()
- if err != nil {
- return nil, status.Errorf(codes.NotFound, "%s", agent.deviceID)
- }
+
+ device := agent.getDeviceWithoutLock()
+
if device.AdminState != voltha.AdminState_ENABLED {
log.Debugw("device-not-enabled", log.Fields{"id": agent.deviceID})
return nil, status.Errorf(codes.FailedPrecondition, "deviceId:%s, expected-admin-state:%s", agent.deviceID, voltha.AdminState_ENABLED)
@@ -829,11 +776,9 @@
agent.lockDevice.Lock()
defer agent.lockDevice.Unlock()
log.Debugw("cancelImageDownload", log.Fields{"id": agent.deviceID})
- // Get the most up to date the device info
- device, err := agent.getDeviceWithoutLock()
- if err != nil {
- return nil, status.Errorf(codes.NotFound, "%s", agent.deviceID)
- }
+
+ device := agent.getDeviceWithoutLock()
+
// Verify whether the Image is in the list of image being downloaded
if !isImageRegistered(img, device) {
return nil, status.Errorf(codes.FailedPrecondition, "deviceId:%s, image-not-registered:%s", agent.deviceID, img.Name)
@@ -866,21 +811,17 @@
agent.lockDevice.Lock()
defer agent.lockDevice.Unlock()
log.Debugw("activateImage", log.Fields{"id": agent.deviceID})
- // Get the most up to date the device info
- device, err := agent.getDeviceWithoutLock()
- if err != nil {
- return nil, status.Errorf(codes.NotFound, "%s", agent.deviceID)
- }
+ cloned := agent.getDeviceWithoutLock()
+
// Verify whether the Image is in the list of image being downloaded
- if !isImageRegistered(img, device) {
+ if !isImageRegistered(img, cloned) {
return nil, status.Errorf(codes.FailedPrecondition, "deviceId:%s, image-not-registered:%s", agent.deviceID, img.Name)
}
- if device.AdminState == voltha.AdminState_DOWNLOADING_IMAGE {
+ if cloned.AdminState == voltha.AdminState_DOWNLOADING_IMAGE {
return nil, status.Errorf(codes.FailedPrecondition, "deviceId:%s, device-in-downloading-state:%s", agent.deviceID, img.Name)
}
// Update image download state
- cloned := proto.Clone(device).(*voltha.Device)
for _, image := range cloned.ImageDownloads {
if image.Id == img.Id && image.Name == img.Name {
image.ImageState = voltha.ImageDownload_IMAGE_ACTIVATING
@@ -892,7 +833,7 @@
return nil, err
}
- if err := agent.adapterProxy.ActivateImageUpdate(ctx, device, img); err != nil {
+ if err := agent.adapterProxy.ActivateImageUpdate(ctx, proto.Clone(cloned).(*voltha.Device), img); err != nil {
log.Debugw("activateImage-error", log.Fields{"id": agent.deviceID, "error": err, "image": img.Name})
return nil, err
}
@@ -905,21 +846,18 @@
agent.lockDevice.Lock()
defer agent.lockDevice.Unlock()
log.Debugw("revertImage", log.Fields{"id": agent.deviceID})
- // Get the most up to date the device info
- device, err := agent.getDeviceWithoutLock()
- if err != nil {
- return nil, status.Errorf(codes.NotFound, "%s", agent.deviceID)
- }
+
+ cloned := agent.getDeviceWithoutLock()
+
// Verify whether the Image is in the list of image being downloaded
- if !isImageRegistered(img, device) {
+ if !isImageRegistered(img, cloned) {
return nil, status.Errorf(codes.FailedPrecondition, "deviceId:%s, image-not-registered:%s", agent.deviceID, img.Name)
}
- if device.AdminState != voltha.AdminState_ENABLED {
+ if cloned.AdminState != voltha.AdminState_ENABLED {
return nil, status.Errorf(codes.FailedPrecondition, "deviceId:%s, device-not-enabled-state:%s", agent.deviceID, img.Name)
}
// Update image download state
- cloned := proto.Clone(device).(*voltha.Device)
for _, image := range cloned.ImageDownloads {
if image.Id == img.Id && image.Name == img.Name {
image.ImageState = voltha.ImageDownload_IMAGE_REVERTING
@@ -930,7 +868,7 @@
return nil, err
}
- if err := agent.adapterProxy.RevertImageUpdate(ctx, device, img); err != nil {
+ if err := agent.adapterProxy.RevertImageUpdate(ctx, proto.Clone(cloned).(*voltha.Device), img); err != nil {
log.Debugw("revertImage-error", log.Fields{"id": agent.deviceID, "error": err, "image": img.Name})
return nil, err
}
@@ -941,12 +879,9 @@
agent.lockDevice.Lock()
defer agent.lockDevice.Unlock()
log.Debugw("getImageDownloadStatus", log.Fields{"id": agent.deviceID})
- // Get the most up to date the device info
- device, err := agent.getDeviceWithoutLock()
- if err != nil {
- return nil, status.Errorf(codes.NotFound, "%s", agent.deviceID)
- }
- resp, err := agent.adapterProxy.GetImageDownloadStatus(ctx, device, img)
+
+ cloned := agent.getDeviceWithoutLock()
+ resp, err := agent.adapterProxy.GetImageDownloadStatus(ctx, cloned, img)
if err != nil {
log.Debugw("getImageDownloadStatus-error", log.Fields{"id": agent.deviceID, "error": err, "image": img.Name})
return nil, err
@@ -958,13 +893,10 @@
agent.lockDevice.Lock()
defer agent.lockDevice.Unlock()
log.Debugw("updateImageDownload", log.Fields{"id": agent.deviceID})
- // Get the most up to date the device info
- device, err := agent.getDeviceWithoutLock()
- if err != nil {
- return status.Errorf(codes.NotFound, "%s", agent.deviceID)
- }
+
+ cloned := agent.getDeviceWithoutLock()
+
// Update the image as well as remove it if the download was cancelled
- cloned := proto.Clone(device).(*voltha.Device)
clonedImages := make([]*voltha.ImageDownload, len(cloned.ImageDownloads))
for _, image := range cloned.ImageDownloads {
if image.Id == img.Id && image.Name == img.Name {
@@ -991,12 +923,9 @@
agent.lockDevice.RLock()
defer agent.lockDevice.RUnlock()
log.Debugw("getImageDownload", log.Fields{"id": agent.deviceID})
- // Get the most up to date the device info
- device, err := agent.getDeviceWithoutLock()
- if err != nil {
- return nil, status.Errorf(codes.NotFound, "%s", agent.deviceID)
- }
- for _, image := range device.ImageDownloads {
+
+ cloned := agent.getDeviceWithoutLock()
+ for _, image := range cloned.ImageDownloads {
if image.Id == img.Id && image.Name == img.Name {
return image, nil
}
@@ -1008,12 +937,8 @@
agent.lockDevice.RLock()
defer agent.lockDevice.RUnlock()
log.Debugw("listImageDownloads", log.Fields{"id": agent.deviceID})
- // Get the most up to date the device info
- device, err := agent.getDeviceWithoutLock()
- if err != nil {
- return nil, status.Errorf(codes.NotFound, "%s", agent.deviceID)
- }
- return &voltha.ImageDownloads{Items: device.ImageDownloads}, nil
+
+ return &voltha.ImageDownloads{Items: agent.getDeviceWithoutLock().ImageDownloads}, nil
}
// getPorts retrieves the ports information of the device based on the port type.
@@ -1114,13 +1039,7 @@
// updatePartialDeviceData updates a subset of a device that an Adapter can update.
// TODO: May need a specific proto to handle only a subset of a device that can be changed by an adapter
func (agent *DeviceAgent) mergeDeviceInfoFromAdapter(device *voltha.Device) (*voltha.Device, error) {
- // First retrieve the most up to date device info
- var currentDevice *voltha.Device
- var err error
- if currentDevice, err = agent.getDeviceWithoutLock(); err != nil {
- return nil, err
- }
- cloned := proto.Clone(currentDevice).(*voltha.Device)
+ cloned := agent.getDeviceWithoutLock()
cloned.Root = device.Root
cloned.Vendor = device.Vendor
cloned.Model = device.Model
@@ -1152,13 +1071,9 @@
func (agent *DeviceAgent) updateDeviceStatus(operStatus voltha.OperStatus_OperStatus, connStatus voltha.ConnectStatus_ConnectStatus) error {
agent.lockDevice.Lock()
defer agent.lockDevice.Unlock()
- // Work only on latest data
- storeDevice, err := agent.getDeviceWithoutLock()
- if err != nil {
- return status.Errorf(codes.NotFound, "%s", agent.deviceID)
- }
- // clone the device
- cloned := proto.Clone(storeDevice).(*voltha.Device)
+
+ cloned := agent.getDeviceWithoutLock()
+
// Ensure the enums passed in are valid - they will be invalid if they are not set when this function is invoked
if s, ok := voltha.ConnectStatus_ConnectStatus_value[connStatus.String()]; ok {
log.Debugw("updateDeviceStatus-conn", log.Fields{"ok": ok, "val": s})
@@ -1176,12 +1091,9 @@
func (agent *DeviceAgent) enablePorts() error {
agent.lockDevice.Lock()
defer agent.lockDevice.Unlock()
- storeDevice, err := agent.getDeviceWithoutLock()
- if err != nil {
- return status.Errorf(codes.NotFound, "%s", agent.deviceID)
- }
- // clone the device
- cloned := proto.Clone(storeDevice).(*voltha.Device)
+
+ cloned := agent.getDeviceWithoutLock()
+
for _, port := range cloned.Ports {
port.AdminState = voltha.AdminState_ENABLED
port.OperStatus = voltha.OperStatus_ACTIVE
@@ -1194,12 +1106,7 @@
log.Debugw("disablePorts", log.Fields{"deviceid": agent.deviceID})
agent.lockDevice.Lock()
defer agent.lockDevice.Unlock()
- storeDevice, err := agent.getDeviceWithoutLock()
- if err != nil {
- return status.Errorf(codes.NotFound, "%s", agent.deviceID)
- }
- // clone the device
- cloned := proto.Clone(storeDevice).(*voltha.Device)
+ cloned := agent.getDeviceWithoutLock()
for _, port := range cloned.Ports {
port.AdminState = voltha.AdminState_DISABLED
port.OperStatus = voltha.OperStatus_UNKNOWN
@@ -1213,12 +1120,8 @@
defer agent.lockDevice.Unlock()
// Work only on latest data
// TODO: Get list of ports from device directly instead of the entire device
- storeDevice, err := agent.getDeviceWithoutLock()
- if err != nil {
- return status.Errorf(codes.NotFound, "%s", agent.deviceID)
- }
- // clone the device
- cloned := proto.Clone(storeDevice).(*voltha.Device)
+ cloned := agent.getDeviceWithoutLock()
+
// Ensure the enums passed in are valid - they will be invalid if they are not set when this function is invoked
if _, ok := voltha.Port_PortType_value[portType.String()]; !ok {
return status.Errorf(codes.InvalidArgument, "%s", portType)
@@ -1243,22 +1146,19 @@
log.Debugw("deleteAllPorts", log.Fields{"deviceId": agent.deviceID})
agent.lockDevice.Lock()
defer agent.lockDevice.Unlock()
- // Work only on latest data
- storeDevice, err := agent.getDeviceWithoutLock()
- if err != nil {
- return status.Errorf(codes.NotFound, "%s", agent.deviceID)
- }
- if storeDevice.AdminState != voltha.AdminState_DISABLED && storeDevice.AdminState != voltha.AdminState_DELETED {
- err = status.Error(codes.FailedPrecondition, fmt.Sprintf("invalid-state-%v", storeDevice.AdminState))
- log.Warnw("invalid-state-removing-ports", log.Fields{"state": storeDevice.AdminState, "error": err})
+
+ cloned := agent.getDeviceWithoutLock()
+
+ if cloned.AdminState != voltha.AdminState_DISABLED && cloned.AdminState != voltha.AdminState_DELETED {
+ err := status.Error(codes.FailedPrecondition, fmt.Sprintf("invalid-state-%v", cloned.AdminState))
+ log.Warnw("invalid-state-removing-ports", log.Fields{"state": cloned.AdminState, "error": err})
return err
}
- if len(storeDevice.Ports) == 0 {
+ if len(cloned.Ports) == 0 {
log.Debugw("no-ports-present", log.Fields{"deviceId": agent.deviceID})
return nil
}
- // clone the device & set the fields to empty
- cloned := proto.Clone(storeDevice).(*voltha.Device)
+
cloned.Ports = []*voltha.Port{}
log.Debugw("portStatusUpdate", log.Fields{"deviceId": cloned.Id})
// Store the device
@@ -1269,13 +1169,8 @@
agent.lockDevice.Lock()
defer agent.lockDevice.Unlock()
log.Debugw("addPort", log.Fields{"deviceId": agent.deviceID})
- // Work only on latest data
- storeDevice, err := agent.getDeviceWithoutLock()
- if err != nil {
- return status.Errorf(codes.NotFound, "%s", agent.deviceID)
- }
- // clone the device
- cloned := proto.Clone(storeDevice).(*voltha.Device)
+
+ cloned := agent.getDeviceWithoutLock()
if cloned.Ports == nil {
// First port
log.Debugw("addPort-first-port-to-add", log.Fields{"deviceId": agent.deviceID})
@@ -1303,13 +1198,9 @@
agent.lockDevice.Lock()
defer agent.lockDevice.Unlock()
log.Debug("addPeerPort")
- // Work only on latest data
- storeDevice, err := agent.getDeviceWithoutLock()
- if err != nil {
- return status.Errorf(codes.NotFound, "%s", agent.deviceID)
- }
- // clone the device
- cloned := proto.Clone(storeDevice).(*voltha.Device)
+
+ cloned := agent.getDeviceWithoutLock()
+
// Get the peer port on the device based on the port no
for _, peerPort := range cloned.Ports {
if peerPort.PortNo == port.PortNo { // found port
@@ -1327,13 +1218,9 @@
agent.lockDevice.Lock()
defer agent.lockDevice.Unlock()
log.Debug("deletePeerPorts")
- // Work only on latest data
- storeDevice, err := agent.getDeviceWithoutLock()
- if err != nil {
- return status.Errorf(codes.NotFound, "%s", agent.deviceID)
- }
- // clone the device
- cloned := proto.Clone(storeDevice).(*voltha.Device)
+
+ cloned := agent.getDeviceWithoutLock()
+
var updatedPeers []*voltha.Port_PeerPort
for _, port := range cloned.Ports {
updatedPeers = make([]*voltha.Port_PeerPort, 0)
@@ -1356,13 +1243,10 @@
if value == nil {
return
}
- var storeDevice *voltha.Device
- var err error
- if storeDevice, err = agent.getDeviceWithoutLock(); err != nil {
- return
- }
+
+ cloned := agent.getDeviceWithoutLock()
updated := false
- s := reflect.ValueOf(storeDevice).Elem()
+ s := reflect.ValueOf(cloned).Elem()
if s.Kind() == reflect.Struct {
// exported field
f := s.FieldByName(name)
@@ -1380,10 +1264,10 @@
}
}
}
- log.Debugw("update-field-status", log.Fields{"deviceId": storeDevice.Id, "name": name, "updated": updated})
+ log.Debugw("update-field-status", log.Fields{"deviceId": cloned.Id, "name": name, "updated": updated})
// Save the data
- cloned := proto.Clone(storeDevice).(*voltha.Device)
- if err = agent.updateDeviceInStoreWithoutLock(cloned, false, ""); err != nil {
+
+ if err := agent.updateDeviceInStoreWithoutLock(cloned, false, ""); err != nil {
log.Warnw("attribute-update-failed", log.Fields{"attribute": name, "value": value})
}
}
@@ -1392,13 +1276,11 @@
agent.lockDevice.Lock()
defer agent.lockDevice.Unlock()
log.Debugw("simulateAlarm", log.Fields{"id": agent.deviceID})
- // Get the most up to date the device info
- device, err := agent.getDeviceWithoutLock()
- if err != nil {
- return status.Errorf(codes.NotFound, "%s", agent.deviceID)
- }
+
+ cloned := agent.getDeviceWithoutLock()
+
// First send the request to an Adapter and wait for a response
- if err := agent.adapterProxy.SimulateAlarm(ctx, device, simulatereq); err != nil {
+ if err := agent.adapterProxy.SimulateAlarm(ctx, cloned, simulatereq); err != nil {
log.Debugw("simulateAlarm-error", log.Fields{"id": agent.deviceID, "error": err})
return err
}
@@ -1414,19 +1296,16 @@
}
log.Debugw("updated-device-in-store", log.Fields{"deviceId: ": agent.deviceID})
+ agent.device = proto.Clone(device).(*voltha.Device)
+
return nil
}
func (agent *DeviceAgent) updateDeviceReason(reason string) error {
agent.lockDevice.Lock()
defer agent.lockDevice.Unlock()
- // Work only on latest data
- storeDevice, err := agent.getDeviceWithoutLock()
- if err != nil {
- return status.Errorf(codes.NotFound, "%s", agent.deviceID)
- }
- // clone the device
- cloned := proto.Clone(storeDevice).(*voltha.Device)
+
+ cloned := agent.getDeviceWithoutLock()
cloned.Reason = reason
log.Debugw("updateDeviceReason", log.Fields{"deviceId": cloned.Id, "reason": cloned.Reason})
// Store the device
diff --git a/rw_core/core/device_agent_test.go b/rw_core/core/device_agent_test.go
new file mode 100755
index 0000000..3f22002
--- /dev/null
+++ b/rw_core/core/device_agent_test.go
@@ -0,0 +1,227 @@
+/*
+* 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 core
+
+import (
+ "context"
+ "github.com/gogo/protobuf/proto"
+ "github.com/opencord/voltha-go/rw_core/config"
+ com "github.com/opencord/voltha-lib-go/v2/pkg/adapters/common"
+ "github.com/opencord/voltha-lib-go/v2/pkg/kafka"
+ "github.com/opencord/voltha-lib-go/v2/pkg/log"
+ lm "github.com/opencord/voltha-lib-go/v2/pkg/mocks"
+ "github.com/opencord/voltha-protos/v2/go/voltha"
+ "github.com/phayes/freeport"
+ "github.com/stretchr/testify/assert"
+ "math/rand"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+)
+
+type DATest struct {
+ etcdServer *lm.EtcdServer
+ core *Core
+ kClient kafka.Client
+ kvClientPort int
+ oltAdapterName string
+ onuAdapterName string
+ coreInstanceID string
+ defaultTimeout time.Duration
+ maxTimeout time.Duration
+ device *voltha.Device
+ done chan int
+}
+
+func newDATest() *DATest {
+ test := &DATest{}
+ // Start the embedded etcd server
+ var err error
+ test.etcdServer, test.kvClientPort, err = startEmbeddedEtcdServer("voltha.rwcore.da.test", "voltha.rwcore.da.etcd", "error")
+ if err != nil {
+ log.Fatal(err)
+ }
+ // Create the kafka client
+ test.kClient = lm.NewKafkaClient()
+ test.oltAdapterName = "olt_adapter_mock"
+ test.onuAdapterName = "onu_adapter_mock"
+ test.coreInstanceID = "rw-da-test"
+ test.defaultTimeout = 5 * time.Second
+ test.maxTimeout = 20 * time.Second
+ test.done = make(chan int)
+ parentID := com.GetRandomString(10)
+ test.device = &voltha.Device{
+ Type: "onu_adapter_mock",
+ ParentId: parentID,
+ ParentPortNo: 1,
+ VendorId: "onu_adapter_mock",
+ Adapter: "onu_adapter_mock",
+ Vlan: 100,
+ Address: nil,
+ ProxyAddress: &voltha.Device_ProxyAddress{
+ DeviceId: parentID,
+ DeviceType: "olt_adapter_mock",
+ ChannelId: 100,
+ ChannelGroupId: 0,
+ ChannelTermination: "",
+ OnuId: 2,
+ },
+ AdminState: voltha.AdminState_PREPROVISIONED,
+ OperStatus: voltha.OperStatus_UNKNOWN,
+ Reason: "All good",
+ ConnectStatus: voltha.ConnectStatus_UNKNOWN,
+ Custom: nil,
+ Ports: []*voltha.Port{
+ {PortNo: 1, Label: "pon-1", Type: voltha.Port_PON_ONU, AdminState: voltha.AdminState_ENABLED,
+ OperStatus: voltha.OperStatus_ACTIVE, Peers: []*voltha.Port_PeerPort{{DeviceId: parentID, PortNo: 1}}},
+ {PortNo: 100, Label: "uni-100", Type: voltha.Port_ETHERNET_UNI, AdminState: voltha.AdminState_ENABLED,
+ OperStatus: voltha.OperStatus_ACTIVE},
+ },
+ }
+
+ return test
+}
+
+func (dat *DATest) startCore(inCompeteMode bool) {
+ cfg := config.NewRWCoreFlags()
+ cfg.CorePairTopic = "rw_core"
+ cfg.DefaultRequestTimeout = dat.defaultTimeout.Nanoseconds() / 1000000 //TODO: change when Core changes to Duration
+ cfg.KVStorePort = dat.kvClientPort
+ cfg.InCompetingMode = inCompeteMode
+ grpcPort, err := freeport.GetFreePort()
+ if err != nil {
+ log.Fatal("Cannot get a freeport for grpc")
+ }
+ cfg.GrpcPort = grpcPort
+ cfg.GrpcHost = "127.0.0.1"
+ setCoreCompeteMode(inCompeteMode)
+ client := setupKVClient(cfg, dat.coreInstanceID)
+ dat.core = NewCore(dat.coreInstanceID, cfg, client, dat.kClient)
+ dat.core.Start(context.Background())
+}
+
+func (dat *DATest) stopAll() {
+ if dat.kClient != nil {
+ dat.kClient.Stop()
+ }
+ if dat.core != nil {
+ dat.core.Stop(context.Background())
+ }
+ if dat.etcdServer != nil {
+ stopEmbeddedEtcdServer(dat.etcdServer)
+ }
+}
+
+func (dat *DATest) createDeviceAgent(t *testing.T) *DeviceAgent {
+ deviceMgr := dat.core.deviceMgr
+ clonedDevice := proto.Clone(dat.device).(*voltha.Device)
+ deviceAgent := newDeviceAgent(deviceMgr.adapterProxy, clonedDevice, deviceMgr, deviceMgr.clusterDataProxy, deviceMgr.defaultTimeout)
+ d, err := deviceAgent.start(context.TODO(), clonedDevice)
+ assert.Nil(t, err)
+ assert.NotNil(t, d)
+ deviceMgr.addDeviceAgentToMap(deviceAgent)
+ return deviceAgent
+}
+
+func (dat *DATest) updateDeviceConcurrently(t *testing.T, da *DeviceAgent, globalWG *sync.WaitGroup) {
+ originalDevice := da.getDevice()
+ assert.NotNil(t, originalDevice)
+ var localWG sync.WaitGroup
+
+ // Update device routine
+ var (
+ root = false
+ vendor = "onu_adapter_mock"
+ model = "go-mock"
+ serialNumber = com.GetRandomSerialNumber()
+ macAddress = strings.ToUpper(com.GetRandomMacAddress())
+ vlan = rand.Uint32()
+ reason = "testing concurrent device update"
+ portToAdd = &voltha.Port{PortNo: 101, Label: "uni-101", Type: voltha.Port_ETHERNET_UNI, AdminState: voltha.AdminState_ENABLED,
+ OperStatus: voltha.OperStatus_ACTIVE}
+ )
+ localWG.Add(1)
+ go func() {
+ deviceToUpdate := proto.Clone(originalDevice).(*voltha.Device)
+ deviceToUpdate.Root = root
+ deviceToUpdate.Vendor = vendor
+ deviceToUpdate.Model = model
+ deviceToUpdate.SerialNumber = serialNumber
+ deviceToUpdate.MacAddress = macAddress
+ deviceToUpdate.Vlan = vlan
+ deviceToUpdate.Reason = reason
+ err := da.updateDeviceUsingAdapterData(deviceToUpdate)
+ assert.Nil(t, err)
+ localWG.Done()
+ }()
+
+ // Update the device status routine
+ localWG.Add(1)
+ go func() {
+ err := da.updateDeviceStatus(voltha.OperStatus_ACTIVE, voltha.ConnectStatus_REACHABLE)
+ assert.Nil(t, err)
+ localWG.Done()
+ }()
+
+ // Add a port routine
+ localWG.Add(1)
+ go func() {
+ err := da.addPort(portToAdd)
+ assert.Nil(t, err)
+ localWG.Done()
+ }()
+
+ // wait for go routines to be done
+ localWG.Wait()
+
+ expectedChange := proto.Clone(originalDevice).(*voltha.Device)
+ expectedChange.OperStatus = voltha.OperStatus_ACTIVE
+ expectedChange.ConnectStatus = voltha.ConnectStatus_REACHABLE
+ expectedChange.Ports = append(expectedChange.Ports, portToAdd)
+ expectedChange.Root = root
+ expectedChange.Vendor = vendor
+ expectedChange.Model = model
+ expectedChange.SerialNumber = serialNumber
+ expectedChange.MacAddress = macAddress
+ expectedChange.Vlan = vlan
+ expectedChange.Reason = reason
+
+ updatedDevice := da.getDevice()
+ assert.NotNil(t, updatedDevice)
+ assert.True(t, proto.Equal(expectedChange, updatedDevice))
+
+ globalWG.Done()
+}
+
+func TestConcurrentDevices(t *testing.T) {
+ da := newDATest()
+ assert.NotNil(t, da)
+ defer da.stopAll()
+
+ // Start the Core
+ da.startCore(false)
+
+ var wg sync.WaitGroup
+ numConCurrentDeviceAgents := 20
+ for i := 0; i < numConCurrentDeviceAgents; i++ {
+ wg.Add(1)
+ a := da.createDeviceAgent(t)
+ go da.updateDeviceConcurrently(t, a, &wg)
+ }
+
+ wg.Wait()
+}
diff --git a/rw_core/core/device_manager.go b/rw_core/core/device_manager.go
index 9e4956e..a876bbe 100755
--- a/rw_core/core/device_manager.go
+++ b/rw_core/core/device_manager.go
@@ -269,7 +269,7 @@
func (dMgr *DeviceManager) GetDevice(id string) (*voltha.Device, error) {
log.Debugw("GetDevice", log.Fields{"deviceid": id})
if agent := dMgr.getDeviceAgent(id); agent != nil {
- return agent.getDevice()
+ return agent.getDevice(), nil
}
return nil, status.Errorf(codes.NotFound, "%s", id)
}
@@ -535,10 +535,7 @@
return err
}
// Get the loaded device details
- var device *voltha.Device
- if device, err = dAgent.getDevice(); err != nil {
- return err
- }
+ device := dAgent.getDevice()
// If the device is in Pre-provisioning or deleted state stop here
if device.AdminState == voltha.AdminState_PREPROVISIONED || device.AdminState == voltha.AdminState_DELETED {
diff --git a/rw_core/core/logical_device_agent.go b/rw_core/core/logical_device_agent.go
index ca07454..e19d111 100644
--- a/rw_core/core/logical_device_agent.go
+++ b/rw_core/core/logical_device_agent.go
@@ -60,6 +60,7 @@
lockLogicalPortsNo sync.RWMutex
flowDecomposer *fd.FlowDecomposer
defaultTimeout int64
+ logicalDevice *voltha.LogicalDevice
}
func newLogicalDeviceAgent(id string, deviceID string, ldeviceMgr *LogicalDeviceManager,
@@ -117,6 +118,8 @@
} else {
log.Debugw("logicaldevice-created", log.Fields{"logicaldeviceId": agent.logicalDeviceID})
}
+
+ agent.logicalDevice = proto.Clone(ld).(*voltha.LogicalDevice)
agent.lockLogicalDevice.Unlock()
// TODO: Set the logical ports in a separate call once the port update issue is fixed.
@@ -130,18 +133,23 @@
} else {
// load from dB - the logical may not exist at this time. On error, just return and the calling function
// will destroy this agent.
- var err error
- if ld, err = agent.GetLogicalDevice(); err != nil {
- log.Warnw("failed-to-load-logical-device", log.Fields{"logicaldeviceId": agent.logicalDeviceID})
- return err
+ agent.lockLogicalDevice.Lock()
+ logicalDevice := agent.clusterDataProxy.Get(context.Background(), "/logical_devices/"+agent.logicalDeviceID, 0, true, "")
+ ld, ok := logicalDevice.(*voltha.LogicalDevice)
+ if !ok {
+ agent.lockLogicalDevice.Unlock()
+ return status.Errorf(codes.NotFound, "logical_device-%s", agent.logicalDeviceID)
}
-
// Update the root device Id
agent.rootDeviceID = ld.RootDeviceId
+ // Update the last data
+ agent.logicalDevice = proto.Clone(ld).(*voltha.LogicalDevice)
+
+ agent.lockLogicalDevice.Unlock()
+
// Setup the local list of logical ports
agent.addLogicalPortsToMap(ld.Ports)
-
}
agent.lockLogicalDevice.Lock()
defer agent.lockLogicalDevice.Unlock()
@@ -194,85 +202,65 @@
log.Info("logical_device-agent-stopped")
}
-// GetLogicalDevice locks the logical device model and then retrieves the latest logical device information
-func (agent *LogicalDeviceAgent) GetLogicalDevice() (*voltha.LogicalDevice, error) {
- log.Debug("GetLogicalDevice")
+// GetLogicalDevice returns the latest logical device data
+func (agent *LogicalDeviceAgent) GetLogicalDevice() *voltha.LogicalDevice {
agent.lockLogicalDevice.RLock()
defer agent.lockLogicalDevice.RUnlock()
- logicalDevice := agent.clusterDataProxy.Get(context.Background(), "/logical_devices/"+agent.logicalDeviceID, 0, false, "")
- if lDevice, ok := logicalDevice.(*voltha.LogicalDevice); ok {
- return lDevice, nil
- }
- return nil, status.Errorf(codes.NotFound, "logical_device-%s", agent.logicalDeviceID)
+
+ return proto.Clone(agent.logicalDevice).(*voltha.LogicalDevice)
}
// ListLogicalDeviceFlows returns logical device flows
-func (agent *LogicalDeviceAgent) ListLogicalDeviceFlows() (*ofp.Flows, error) {
+func (agent *LogicalDeviceAgent) ListLogicalDeviceFlows() *ofp.Flows {
log.Debug("ListLogicalDeviceFlows")
- agent.lockLogicalDevice.RLock()
- defer agent.lockLogicalDevice.RUnlock()
- logicalDevice := agent.clusterDataProxy.Get(context.Background(), "/logical_devices/"+agent.logicalDeviceID, 0, false, "")
- if lDevice, ok := logicalDevice.(*voltha.LogicalDevice); ok {
- cFlows := (proto.Clone(lDevice.Flows)).(*ofp.Flows)
- return cFlows, nil
+
+ logicalDevice := agent.GetLogicalDevice()
+ if logicalDevice.Flows == nil {
+ return &ofp.Flows{}
}
- return nil, status.Errorf(codes.NotFound, "logical_device-%s", agent.logicalDeviceID)
+ return (proto.Clone(logicalDevice.Flows)).(*ofp.Flows)
}
// ListLogicalDeviceMeters returns logical device meters
-func (agent *LogicalDeviceAgent) ListLogicalDeviceMeters() (*ofp.Meters, error) {
+func (agent *LogicalDeviceAgent) ListLogicalDeviceMeters() *ofp.Meters {
log.Debug("ListLogicalDeviceMeters")
- agent.lockLogicalDevice.RLock()
- defer agent.lockLogicalDevice.RUnlock()
- logicalDevice := agent.clusterDataProxy.Get(context.Background(), "/logical_devices/"+agent.logicalDeviceID, 0, false, "")
- if lDevice, ok := logicalDevice.(*voltha.LogicalDevice); ok {
- if lDevice.Meters == nil {
- return &ofp.Meters{}, nil
- }
- cMeters := (proto.Clone(lDevice.Meters)).(*ofp.Meters)
- return cMeters, nil
+
+ logicalDevice := agent.GetLogicalDevice()
+ if logicalDevice.Meters == nil {
+ return &ofp.Meters{}
}
- return nil, status.Errorf(codes.NotFound, "logical_device-%s", agent.logicalDeviceID)
+ return (proto.Clone(logicalDevice.Meters)).(*ofp.Meters)
}
// ListLogicalDeviceFlowGroups returns logical device flow groups
-func (agent *LogicalDeviceAgent) ListLogicalDeviceFlowGroups() (*ofp.FlowGroups, error) {
+func (agent *LogicalDeviceAgent) ListLogicalDeviceFlowGroups() *ofp.FlowGroups {
log.Debug("ListLogicalDeviceFlowGroups")
- agent.lockLogicalDevice.RLock()
- defer agent.lockLogicalDevice.RUnlock()
- logicalDevice := agent.clusterDataProxy.Get(context.Background(), "/logical_devices/"+agent.logicalDeviceID, 0, false, "")
- if lDevice, ok := logicalDevice.(*voltha.LogicalDevice); ok {
- cFlowGroups := (proto.Clone(lDevice.FlowGroups)).(*ofp.FlowGroups)
- return cFlowGroups, nil
+
+ logicalDevice := agent.GetLogicalDevice()
+ if logicalDevice.FlowGroups == nil {
+ return &ofp.FlowGroups{}
}
- return nil, status.Errorf(codes.NotFound, "logical_device-%s", agent.logicalDeviceID)
+ return (proto.Clone(logicalDevice.FlowGroups)).(*ofp.FlowGroups)
}
// ListLogicalDevicePorts returns logical device ports
-func (agent *LogicalDeviceAgent) ListLogicalDevicePorts() (*voltha.LogicalPorts, error) {
+func (agent *LogicalDeviceAgent) ListLogicalDevicePorts() *voltha.LogicalPorts {
log.Debug("ListLogicalDevicePorts")
- agent.lockLogicalDevice.RLock()
- defer agent.lockLogicalDevice.RUnlock()
- logicalDevice := agent.clusterDataProxy.Get(context.Background(), "/logical_devices/"+agent.logicalDeviceID, 0, false, "")
- if lDevice, ok := logicalDevice.(*voltha.LogicalDevice); ok {
- lPorts := make([]*voltha.LogicalPort, 0)
- lPorts = append(lPorts, lDevice.Ports...)
- return &voltha.LogicalPorts{Items: lPorts}, nil
- }
- return nil, status.Errorf(codes.NotFound, "logical_device-%s", agent.logicalDeviceID)
+
+ logicalDevice := agent.GetLogicalDevice()
+ lPorts := make([]*voltha.LogicalPort, 0)
+ lPorts = append(lPorts, logicalDevice.Ports...)
+ return &voltha.LogicalPorts{Items: lPorts}
}
//updateLogicalDeviceFlowsWithoutLock updates the logical device with the latest flows in the model.
func (agent *LogicalDeviceAgent) updateLogicalDeviceFlowsWithoutLock(flows *ofp.Flows) error {
- ld, err := agent.getLogicalDeviceWithoutLock()
- if err != nil {
- return status.Errorf(codes.Internal, "logical-device-absent:%s", agent.logicalDeviceID)
- }
- log.Debugw("logical-device-before", log.Fields{"lports": len(ld.Ports)})
- cloned := (proto.Clone(ld)).(*voltha.LogicalDevice)
- cloned.Flows = flows
+ ld := agent.getLogicalDeviceWithoutLock()
- if err = agent.updateLogicalDeviceWithoutLock(cloned); err != nil {
+ log.Debugw("logical-device-before", log.Fields{"lports": len(ld.Ports)})
+ ld.Flows = flows
+
+ if err := agent.updateLogicalDeviceWithoutLock(ld); err != nil {
log.Errorw("error-updating-logical-device-with-flows", log.Fields{"error": err})
return err
}
@@ -281,15 +269,12 @@
//updateLogicalDeviceMetersWithoutLock updates the logical device with the meters info
func (agent *LogicalDeviceAgent) updateLogicalDeviceMetersWithoutLock(meters *ofp.Meters) error {
- ld, err := agent.getLogicalDeviceWithoutLock()
- if err != nil {
- return status.Errorf(codes.Internal, "logical-device-absent:%s", agent.logicalDeviceID)
- }
- log.Debugw("logical-device-before", log.Fields{"lports": len(ld.Ports)})
- cloned := (proto.Clone(ld)).(*voltha.LogicalDevice)
- cloned.Meters = meters
+ ld := agent.getLogicalDeviceWithoutLock()
- if err = agent.updateLogicalDeviceWithoutLock(cloned); err != nil {
+ log.Debugw("logical-device-before", log.Fields{"lports": len(ld.Ports)})
+ ld.Meters = meters
+
+ if err := agent.updateLogicalDeviceWithoutLock(ld); err != nil {
log.Errorw("error-updating-logical-device-with-meters", log.Fields{"error": err})
return err
}
@@ -298,31 +283,22 @@
//updateLogicalDeviceFlowGroupsWithoutLock updates the logical device with the flow groups
func (agent *LogicalDeviceAgent) updateLogicalDeviceFlowGroupsWithoutLock(flowGroups *ofp.FlowGroups) error {
- ld, err := agent.getLogicalDeviceWithoutLock()
- if err != nil {
- return status.Errorf(codes.Internal, "logical-device-absent:%s", agent.logicalDeviceID)
- }
- log.Debugw("logical-device-before", log.Fields{"lports": len(ld.Ports)})
- cloned := (proto.Clone(ld)).(*voltha.LogicalDevice)
- cloned.FlowGroups = flowGroups
+ ld := agent.getLogicalDeviceWithoutLock()
- if err = agent.updateLogicalDeviceWithoutLock(cloned); err != nil {
+ log.Debugw("logical-device-before", log.Fields{"lports": len(ld.Ports)})
+ ld.FlowGroups = flowGroups
+
+ if err := agent.updateLogicalDeviceWithoutLock(ld); err != nil {
log.Errorw("error-updating-logical-device-with-flowgroups", log.Fields{"error": err})
return err
}
return nil
}
-// getLogicalDeviceWithoutLock retrieves a logical device from the model without locking it. This is used only by
-// functions that have already acquired the logical device lock to the model
-func (agent *LogicalDeviceAgent) getLogicalDeviceWithoutLock() (*voltha.LogicalDevice, error) {
+// getLogicalDeviceWithoutLock returns a cloned logical device to a function that already holds the agent lock.
+func (agent *LogicalDeviceAgent) getLogicalDeviceWithoutLock() *voltha.LogicalDevice {
log.Debug("getLogicalDeviceWithoutLock")
- logicalDevice := agent.clusterDataProxy.Get(context.Background(), "/logical_devices/"+agent.logicalDeviceID, 0, false, "")
- if lDevice, ok := logicalDevice.(*voltha.LogicalDevice); ok {
- //log.Debug("getLogicalDeviceWithoutLock", log.Fields{"ldevice": lDevice})
- return lDevice, nil
- }
- return nil, status.Errorf(codes.NotFound, "logical_device-%s", agent.logicalDeviceID)
+ return proto.Clone(agent.logicalDevice).(*voltha.LogicalDevice)
}
func (agent *LogicalDeviceAgent) updateLogicalPort(device *voltha.Device, port *voltha.Port) error {
@@ -414,14 +390,9 @@
agent.lockLogicalDevice.Lock()
defer agent.lockLogicalDevice.Unlock()
// Get the latest logical device info
- ld, err := agent.getLogicalDeviceWithoutLock()
- if err != nil {
- log.Warnw("logical-device-unknown", log.Fields{"logicalDeviceId": agent.logicalDeviceID, "error": err})
- return err
- }
- for idx, lPort := range ld.Ports {
+ cloned := agent.getLogicalDeviceWithoutLock()
+ for idx, lPort := range cloned.Ports {
if lPort.DeviceId == deviceID && lPort.DevicePortNo == portNo {
- cloned := (proto.Clone(ld)).(*voltha.LogicalDevice)
if operStatus == voltha.OperStatus_ACTIVE {
cloned.Ports[idx].OfpPort.Config = cloned.Ports[idx].OfpPort.Config & ^uint32(ofp.OfpPortConfig_OFPPC_PORT_DOWN)
cloned.Ports[idx].OfpPort.State = uint32(ofp.OfpPortState_OFPPS_LIVE)
@@ -446,12 +417,7 @@
agent.lockLogicalDevice.Lock()
defer agent.lockLogicalDevice.Unlock()
// Get the latest logical device info
- ld, err := agent.getLogicalDeviceWithoutLock()
- if err != nil {
- log.Warnw("logical-device-unknown", log.Fields{"ldeviceId": agent.logicalDeviceID, "error": err})
- return err
- }
- cloned := (proto.Clone(ld)).(*voltha.LogicalDevice)
+ cloned := agent.getLogicalDeviceWithoutLock()
for _, lport := range cloned.Ports {
if lport.DeviceId == device.Id {
switch state {
@@ -500,11 +466,8 @@
agent.lockLogicalDevice.Lock()
defer agent.lockLogicalDevice.Unlock()
// Get the latest logical device info
- ld, err := agent.getLogicalDeviceWithoutLock()
- if err != nil {
- log.Warnw("logical-device-unknown", log.Fields{"ldeviceId": agent.logicalDeviceID, "error": err})
- return err
- }
+ ld := agent.getLogicalDeviceWithoutLock()
+
cloned := (proto.Clone(ld)).(*voltha.LogicalDevice)
updateLogicalPorts := []*voltha.LogicalPort{}
for _, lport := range cloned.Ports {
@@ -532,17 +495,14 @@
if afterUpdate == nil {
return status.Errorf(codes.Internal, "failed-updating-logical-device:%s", agent.logicalDeviceID)
}
+ agent.logicalDevice = (proto.Clone(logicalDevice)).(*voltha.LogicalDevice)
return nil
}
//generateDeviceGraphIfNeeded generates the device graph if the logical device has been updated since the last time
//that device graph was generated.
func (agent *LogicalDeviceAgent) generateDeviceGraphIfNeeded() error {
- ld, err := agent.GetLogicalDevice()
- if err != nil {
- log.Errorw("get-logical-device-error", log.Fields{"error": err})
- return err
- }
+ ld := agent.GetLogicalDevice()
agent.lockDeviceGraph.Lock()
defer agent.lockDeviceGraph.Unlock()
if agent.deviceGraph != nil && agent.deviceGraph.IsUpToDate(ld) {
@@ -630,12 +590,7 @@
agent.lockLogicalDevice.Lock()
defer agent.lockLogicalDevice.Unlock()
log.Debug("Acquired logical device lock")
- var lDevice *voltha.LogicalDevice
- var err error
- if lDevice, err = agent.getLogicalDeviceWithoutLock(); err != nil {
- log.Errorw("no-logical-device-present", log.Fields{"logicalDeviceId": agent.logicalDeviceID})
- return fmt.Errorf("no-logical-device-present:%s", agent.logicalDeviceID)
- }
+ lDevice := agent.getLogicalDeviceWithoutLock()
var meters []*ofp.OfpMeterEntry
if lDevice.Meters != nil && lDevice.Meters.Items != nil {
@@ -669,12 +624,7 @@
agent.lockLogicalDevice.Lock()
defer agent.lockLogicalDevice.Unlock()
- var lDevice *voltha.LogicalDevice
- var err error
- if lDevice, err = agent.getLogicalDeviceWithoutLock(); err != nil {
- log.Errorw("no-logical-device-present", log.Fields{"logicalDeviceId": agent.logicalDeviceID})
- return fmt.Errorf("no-logical-device-present:%s", agent.logicalDeviceID)
- }
+ lDevice := agent.getLogicalDeviceWithoutLock()
var meters []*ofp.OfpMeterEntry
var flows []*ofp.OfpFlowStats
@@ -730,12 +680,7 @@
agent.lockLogicalDevice.Lock()
defer agent.lockLogicalDevice.Unlock()
- var lDevice *voltha.LogicalDevice
- var err error
- if lDevice, err = agent.getLogicalDeviceWithoutLock(); err != nil {
- log.Errorw("no-logical-device-present", log.Fields{"logicalDeviceId": agent.logicalDeviceID})
- return fmt.Errorf("no-logical-device-present:%s", agent.logicalDeviceID)
- }
+ lDevice := agent.getLogicalDeviceWithoutLock()
var meters []*ofp.OfpMeterEntry
if lDevice.Meters != nil && lDevice.Meters.Items != nil {
@@ -824,12 +769,7 @@
agent.lockLogicalDevice.Lock()
defer agent.lockLogicalDevice.Unlock()
- var lDevice *voltha.LogicalDevice
- var err error
- if lDevice, err = agent.getLogicalDeviceWithoutLock(); err != nil {
- log.Errorw("no-logical-device-present", log.Fields{"logicalDeviceId": agent.logicalDeviceID})
- return fmt.Errorf("no-logical-device-present:%s", agent.logicalDeviceID)
- }
+ lDevice := agent.getLogicalDeviceWithoutLock()
var flows []*ofp.OfpFlowStats
var meters []*ofp.OfpMeterEntry
@@ -958,12 +898,7 @@
agent.lockLogicalDevice.Lock()
defer agent.lockLogicalDevice.Unlock()
- var lDevice *voltha.LogicalDevice
- var err error
- if lDevice, err = agent.getLogicalDeviceWithoutLock(); err != nil {
- log.Errorw("no-logical-device-present", log.Fields{"logicalDeviceId": agent.logicalDeviceID})
- return fmt.Errorf("no-logical-device-present:%s", agent.logicalDeviceID)
- }
+ lDevice := agent.getLogicalDeviceWithoutLock()
var meters []*ofp.OfpMeterEntry
var flows []*ofp.OfpFlowStats
@@ -1094,12 +1029,8 @@
agent.lockLogicalDevice.Lock()
defer agent.lockLogicalDevice.Unlock()
- var lDevice *voltha.LogicalDevice
- var err error
- if lDevice, err = agent.getLogicalDeviceWithoutLock(); err != nil {
- log.Errorw("no-logical-device-present", log.Fields{"logicalDeviceId": agent.logicalDeviceID})
- return fmt.Errorf("no-logical-device-present:%s", agent.logicalDeviceID)
- }
+ lDevice := agent.getLogicalDeviceWithoutLock()
+
var meters []*ofp.OfpMeterEntry
var flows []*ofp.OfpFlowStats
if lDevice.Meters != nil && lDevice.Meters.Items != nil {
@@ -1174,12 +1105,8 @@
agent.lockLogicalDevice.Lock()
defer agent.lockLogicalDevice.Unlock()
- var lDevice *voltha.LogicalDevice
- var err error
- if lDevice, err = agent.getLogicalDeviceWithoutLock(); err != nil {
- log.Errorw("no-logical-device-present", log.Fields{"logicalDeviceId": agent.logicalDeviceID})
- return fmt.Errorf("no-logical-device-present:%s", agent.logicalDeviceID)
- }
+ lDevice := agent.getLogicalDeviceWithoutLock()
+
groups := lDevice.FlowGroups.Items
if fu.FindGroup(groups, groupMod.GroupId) == -1 {
groups = append(groups, fu.GroupEntryFromGroupMod(groupMod))
@@ -1209,12 +1136,7 @@
agent.lockLogicalDevice.Lock()
defer agent.lockLogicalDevice.Unlock()
- var lDevice *voltha.LogicalDevice
- var err error
- if lDevice, err = agent.getLogicalDeviceWithoutLock(); err != nil {
- log.Errorw("no-logical-device-present", log.Fields{"logicalDeviceId": agent.logicalDeviceID})
- return fmt.Errorf("no-logical-device-present:%s", agent.logicalDeviceID)
- }
+ lDevice := agent.getLogicalDeviceWithoutLock()
groups := lDevice.FlowGroups.Items
flows := lDevice.Flows.Items
var groupsChanged bool
@@ -1267,12 +1189,7 @@
agent.lockLogicalDevice.Lock()
defer agent.lockLogicalDevice.Unlock()
- var lDevice *voltha.LogicalDevice
- var err error
- if lDevice, err = agent.getLogicalDeviceWithoutLock(); err != nil {
- log.Errorw("no-logical-device-present", log.Fields{"logicalDeviceId": agent.logicalDeviceID})
- return fmt.Errorf("no-logical-device-present:%s", agent.logicalDeviceID)
- }
+ lDevice := agent.getLogicalDeviceWithoutLock()
groups := lDevice.FlowGroups.Items
var groupsChanged bool
groupID := groupMod.GroupId
@@ -1307,25 +1224,21 @@
agent.lockLogicalDevice.Lock()
defer agent.lockLogicalDevice.Unlock()
- // Get the most up to date logical device
- var logicaldevice *voltha.LogicalDevice
- if logicaldevice, _ = agent.getLogicalDeviceWithoutLock(); logicaldevice == nil {
- log.Debugw("no-logical-device", log.Fields{"logicalDeviceId": agent.logicalDeviceID, "logicalPortId": lPort.Id})
- return nil
- }
+ logicalDevice := agent.getLogicalDeviceWithoutLock()
+
index := -1
- for i, logicalPort := range logicaldevice.Ports {
+ for i, logicalPort := range logicalDevice.Ports {
if logicalPort.Id == lPort.Id {
index = i
break
}
}
if index >= 0 {
- copy(logicaldevice.Ports[index:], logicaldevice.Ports[index+1:])
- logicaldevice.Ports[len(logicaldevice.Ports)-1] = nil
- logicaldevice.Ports = logicaldevice.Ports[:len(logicaldevice.Ports)-1]
+ copy(logicalDevice.Ports[index:], logicalDevice.Ports[index+1:])
+ logicalDevice.Ports[len(logicalDevice.Ports)-1] = nil
+ logicalDevice.Ports = logicalDevice.Ports[:len(logicalDevice.Ports)-1]
log.Debugw("logical-port-deleted", log.Fields{"logicalDeviceId": agent.logicalDeviceID})
- if err := agent.updateLogicalDeviceWithoutLock(logicaldevice); err != nil {
+ if err := agent.updateLogicalDeviceWithoutLock(logicalDevice); err != nil {
log.Errorw("logical-device-update-failed", log.Fields{"logicalDeviceId": agent.logicalDeviceID})
return err
}
@@ -1340,21 +1253,16 @@
agent.lockLogicalDevice.Lock()
defer agent.lockLogicalDevice.Unlock()
- // Get the most up to date logical device
- var logicaldevice *voltha.LogicalDevice
- if logicaldevice, _ = agent.getLogicalDeviceWithoutLock(); logicaldevice == nil {
- log.Debugw("no-logical-device", log.Fields{"logicalDeviceId": agent.logicalDeviceID})
- return nil
- }
+ logicalDevice := agent.getLogicalDeviceWithoutLock()
updatedLPorts := []*voltha.LogicalPort{}
- for _, logicalPort := range logicaldevice.Ports {
+ for _, logicalPort := range logicalDevice.Ports {
if logicalPort.DeviceId != deviceID {
updatedLPorts = append(updatedLPorts, logicalPort)
}
}
- logicaldevice.Ports = updatedLPorts
+ logicalDevice.Ports = updatedLPorts
log.Debugw("updated-logical-ports", log.Fields{"ports": updatedLPorts})
- if err := agent.updateLogicalDeviceWithoutLock(logicaldevice); err != nil {
+ if err := agent.updateLogicalDeviceWithoutLock(logicalDevice); err != nil {
log.Errorw("logical-device-update-failed", log.Fields{"logicalDeviceId": agent.logicalDeviceID})
return err
}
@@ -1369,22 +1277,18 @@
agent.lockLogicalDevice.Lock()
defer agent.lockLogicalDevice.Unlock()
- // Get the most up to date logical device
- var logicaldevice *voltha.LogicalDevice
- if logicaldevice, _ = agent.getLogicalDeviceWithoutLock(); logicaldevice == nil {
- log.Debugw("no-logical-device", log.Fields{"logicalDeviceId": agent.logicalDeviceID, "logicalPortId": lPortID})
- return nil
- }
+ logicalDevice := agent.getLogicalDeviceWithoutLock()
+
index := -1
- for i, logicalPort := range logicaldevice.Ports {
+ for i, logicalPort := range logicalDevice.Ports {
if logicalPort.Id == lPortID {
index = i
break
}
}
if index >= 0 {
- logicaldevice.Ports[index].OfpPort.Config = logicaldevice.Ports[index].OfpPort.Config & ^uint32(ofp.OfpPortConfig_OFPPC_PORT_DOWN)
- return agent.updateLogicalDeviceWithoutLock(logicaldevice)
+ logicalDevice.Ports[index].OfpPort.Config = logicalDevice.Ports[index].OfpPort.Config & ^uint32(ofp.OfpPortConfig_OFPPC_PORT_DOWN)
+ return agent.updateLogicalDeviceWithoutLock(logicalDevice)
}
return status.Errorf(codes.NotFound, "Port %s on Logical Device %s", lPortID, agent.logicalDeviceID)
}
@@ -1395,21 +1299,17 @@
defer agent.lockLogicalDevice.Unlock()
// Get the most up to date logical device
- var logicaldevice *voltha.LogicalDevice
- if logicaldevice, _ = agent.getLogicalDeviceWithoutLock(); logicaldevice == nil {
- log.Debugw("no-logical-device", log.Fields{"logicalDeviceId": agent.logicalDeviceID, "logicalPortId": lPortID})
- return nil
- }
+ logicalDevice := agent.getLogicalDeviceWithoutLock()
index := -1
- for i, logicalPort := range logicaldevice.Ports {
+ for i, logicalPort := range logicalDevice.Ports {
if logicalPort.Id == lPortID {
index = i
break
}
}
if index >= 0 {
- logicaldevice.Ports[index].OfpPort.Config = (logicaldevice.Ports[index].OfpPort.Config & ^uint32(ofp.OfpPortConfig_OFPPC_PORT_DOWN)) | uint32(ofp.OfpPortConfig_OFPPC_PORT_DOWN)
- return agent.updateLogicalDeviceWithoutLock(logicaldevice)
+ logicalDevice.Ports[index].OfpPort.Config = (logicalDevice.Ports[index].OfpPort.Config & ^uint32(ofp.OfpPortConfig_OFPPC_PORT_DOWN)) | uint32(ofp.OfpPortConfig_OFPPC_PORT_DOWN)
+ return agent.updateLogicalDeviceWithoutLock(logicalDevice)
}
return status.Errorf(codes.NotFound, "Port %s on Logical Device %s", lPortID, agent.logicalDeviceID)
}
@@ -1505,11 +1405,10 @@
if len(excludePort) == 1 {
exclPort = excludePort[0]
}
- if lDevice, _ := agent.getLogicalDeviceWithoutLock(); lDevice != nil {
- for _, port := range lDevice.Ports {
- if port.OfpPort.PortNo != exclPort {
- lPorts = append(lPorts, port.OfpPort.PortNo)
- }
+ lDevice := agent.getLogicalDeviceWithoutLock()
+ for _, port := range lDevice.Ports {
+ if port.OfpPort.PortNo != exclPort {
+ lPorts = append(lPorts, port.OfpPort.PortNo)
}
}
return lPorts
@@ -1529,11 +1428,8 @@
agent.deviceGraph = graph.NewDeviceGraph(agent.logicalDeviceID, agent.deviceMgr.GetDevice)
}
// Get all the logical ports on that logical device
- lDevice, err := agent.getLogicalDeviceWithoutLock()
- if err != nil {
- log.Errorw("unknown-logical-device", log.Fields{"error": err, "logicalDeviceId": agent.logicalDeviceID})
- return err
- }
+ lDevice := agent.getLogicalDeviceWithoutLock()
+
//TODO: Find a better way to refresh only missing routes
agent.deviceGraph.ComputeRoutes(lDevice.Ports)
agent.deviceGraph.Print()
@@ -1558,16 +1454,13 @@
agent.lockLogicalDevice.Lock()
defer agent.lockLogicalDevice.Unlock()
// Get the latest logical device
- if ld, err := agent.getLogicalDeviceWithoutLock(); err != nil {
- log.Errorw("logical-device-not-present", log.Fields{"logicalDeviceId": agent.logicalDeviceID, "error": err})
- } else {
- log.Debugw("generating-graph", log.Fields{"lDeviceId": agent.logicalDeviceID, "lPorts": len(ld.Ports)})
- if agent.deviceGraph == nil {
- agent.deviceGraph = graph.NewDeviceGraph(agent.logicalDeviceID, agent.deviceMgr.GetDevice)
- }
- agent.deviceGraph.ComputeRoutes(ld.Ports)
- agent.deviceGraph.Print()
+ ld := agent.getLogicalDeviceWithoutLock()
+ log.Debugw("generating-graph", log.Fields{"lDeviceId": agent.logicalDeviceID, "lPorts": len(ld.Ports)})
+ if agent.deviceGraph == nil {
+ agent.deviceGraph = graph.NewDeviceGraph(agent.logicalDeviceID, agent.deviceMgr.GetDevice)
}
+ agent.deviceGraph.ComputeRoutes(ld.Ports)
+ agent.deviceGraph.Print()
}
// diff go over two lists of logical ports and return what's new, what's changed and what's removed.
@@ -1693,11 +1586,8 @@
lp.OfpPort.Name = lp.Id
lp.DevicePortNo = port.PortNo
- var ld *voltha.LogicalDevice
- if ld, err = agent.getLogicalDeviceWithoutLock(); err != nil {
- log.Errorw("error-retrieving-logical-device", log.Fields{"error": err})
- return false, err
- }
+ ld := agent.getLogicalDeviceWithoutLock()
+
cloned := (proto.Clone(ld)).(*voltha.LogicalDevice)
if cloned.Ports == nil {
cloned.Ports = make([]*voltha.LogicalPort, 0)
@@ -1717,11 +1607,10 @@
}
func (agent *LogicalDeviceAgent) portExist(device *voltha.Device, port *voltha.Port) bool {
- if ldevice, _ := agent.getLogicalDeviceWithoutLock(); ldevice != nil {
- for _, lPort := range ldevice.Ports {
- if lPort.DeviceId == device.Id && lPort.DevicePortNo == port.PortNo && lPort.Id == port.Label {
- return true
- }
+ ldevice := agent.getLogicalDeviceWithoutLock()
+ for _, lPort := range ldevice.Ports {
+ if lPort.DeviceId == device.Id && lPort.DevicePortNo == port.PortNo && lPort.Id == port.Label {
+ return true
}
}
return false
@@ -1759,10 +1648,8 @@
return false, nil
}
// Get stored logical device
- ldevice, err := agent.getLogicalDeviceWithoutLock()
- if err != nil {
- return false, status.Error(codes.NotFound, agent.logicalDeviceID)
- }
+ ldevice := agent.getLogicalDeviceWithoutLock()
+
log.Debugw("adding-uni", log.Fields{"deviceId": childDevice.Id})
portCap.Port.RootPort = false
portCap.Port.Id = port.Label
diff --git a/rw_core/core/logical_device_agent_test.go b/rw_core/core/logical_device_agent_test.go
index 381e2a5..b5789cd 100644
--- a/rw_core/core/logical_device_agent_test.go
+++ b/rw_core/core/logical_device_agent_test.go
@@ -16,10 +16,22 @@
package core
import (
+ "context"
+ "github.com/gogo/protobuf/proto"
+ "github.com/opencord/voltha-go/rw_core/config"
+ com "github.com/opencord/voltha-lib-go/v2/pkg/adapters/common"
+ fu "github.com/opencord/voltha-lib-go/v2/pkg/flows"
+ "github.com/opencord/voltha-lib-go/v2/pkg/kafka"
+ "github.com/opencord/voltha-lib-go/v2/pkg/log"
+ lm "github.com/opencord/voltha-lib-go/v2/pkg/mocks"
ofp "github.com/opencord/voltha-protos/v2/go/openflow_13"
"github.com/opencord/voltha-protos/v2/go/voltha"
+ "github.com/phayes/freeport"
"github.com/stretchr/testify/assert"
+ "math/rand"
+ "sync"
"testing"
+ "time"
)
func TestLogicalDeviceAgent_diff_nochange_1(t *testing.T) {
@@ -343,3 +355,224 @@
assert.Equal(t, updatedLogicalPorts[1], changedPorts[1])
assert.Equal(t, currentLogicalPorts[2], deletedPorts[0])
}
+
+type LDATest struct {
+ etcdServer *lm.EtcdServer
+ core *Core
+ kClient kafka.Client
+ kvClientPort int
+ oltAdapterName string
+ onuAdapterName string
+ coreInstanceID string
+ defaultTimeout time.Duration
+ maxTimeout time.Duration
+ logicalDevice *voltha.LogicalDevice
+ deviceIds []string
+ done chan int
+}
+
+func newLDATest() *LDATest {
+ test := &LDATest{}
+ // Start the embedded etcd server
+ var err error
+ test.etcdServer, test.kvClientPort, err = startEmbeddedEtcdServer("voltha.rwcore.lda.test", "voltha.rwcore.lda.etcd", "error")
+ if err != nil {
+ log.Fatal(err)
+ }
+ // Create the kafka client
+ test.kClient = lm.NewKafkaClient()
+ test.oltAdapterName = "olt_adapter_mock"
+ test.onuAdapterName = "onu_adapter_mock"
+ test.coreInstanceID = "rw-da-test"
+ test.defaultTimeout = 5 * time.Second
+ test.maxTimeout = 20 * time.Second
+ test.done = make(chan int)
+ test.deviceIds = []string{com.GetRandomString(10), com.GetRandomString(10), com.GetRandomString(10)}
+ test.logicalDevice = &voltha.LogicalDevice{
+ Desc: &ofp.OfpDesc{
+ HwDesc: "olt_adapter_mock",
+ SwDesc: "olt_adapter_mock",
+ SerialNum: com.GetRandomSerialNumber(),
+ },
+ SwitchFeatures: &ofp.OfpSwitchFeatures{
+ NBuffers: 256,
+ NTables: 2,
+ Capabilities: uint32(ofp.OfpCapabilities_OFPC_FLOW_STATS |
+ ofp.OfpCapabilities_OFPC_TABLE_STATS |
+ ofp.OfpCapabilities_OFPC_PORT_STATS |
+ ofp.OfpCapabilities_OFPC_GROUP_STATS),
+ },
+ RootDeviceId: test.deviceIds[0],
+ Ports: []*voltha.LogicalPort{
+ {
+ Id: "1001",
+ DeviceId: test.deviceIds[0],
+ DevicePortNo: 1,
+ RootPort: true,
+ OfpPort: &ofp.OfpPort{
+ PortNo: 1,
+ Name: "port1",
+ Config: 4,
+ State: 4,
+ },
+ },
+ {
+ Id: "1002",
+ DeviceId: test.deviceIds[1],
+ DevicePortNo: 2,
+ RootPort: false,
+ OfpPort: &ofp.OfpPort{
+ PortNo: 2,
+ Name: "port2",
+ Config: 4,
+ State: 4,
+ },
+ },
+ {
+ Id: "1003",
+ DeviceId: test.deviceIds[2],
+ DevicePortNo: 3,
+ RootPort: false,
+ OfpPort: &ofp.OfpPort{
+ PortNo: 4,
+ Name: "port3",
+ Config: 4,
+ State: 4,
+ },
+ },
+ },
+ }
+ return test
+}
+
+func (lda *LDATest) startCore(inCompeteMode bool) {
+ cfg := config.NewRWCoreFlags()
+ cfg.CorePairTopic = "rw_core"
+ cfg.DefaultRequestTimeout = lda.defaultTimeout.Nanoseconds() / 1000000 //TODO: change when Core changes to Duration
+ cfg.KVStorePort = lda.kvClientPort
+ cfg.InCompetingMode = inCompeteMode
+ grpcPort, err := freeport.GetFreePort()
+ if err != nil {
+ log.Fatal("Cannot get a freeport for grpc")
+ }
+ cfg.GrpcPort = grpcPort
+ cfg.GrpcHost = "127.0.0.1"
+ setCoreCompeteMode(inCompeteMode)
+ client := setupKVClient(cfg, lda.coreInstanceID)
+ lda.core = NewCore(lda.coreInstanceID, cfg, client, lda.kClient)
+ lda.core.Start(context.Background())
+}
+
+func (lda *LDATest) stopAll() {
+ if lda.kClient != nil {
+ lda.kClient.Stop()
+ }
+ if lda.core != nil {
+ lda.core.Stop(context.Background())
+ }
+ if lda.etcdServer != nil {
+ stopEmbeddedEtcdServer(lda.etcdServer)
+ }
+}
+
+func (lda *LDATest) createLogicalDeviceAgent(t *testing.T) *LogicalDeviceAgent {
+ lDeviceMgr := lda.core.logicalDeviceMgr
+ deviceMgr := lda.core.deviceMgr
+ clonedLD := proto.Clone(lda.logicalDevice).(*voltha.LogicalDevice)
+ clonedLD.Id = com.GetRandomString(10)
+ clonedLD.DatapathId = rand.Uint64()
+ lDeviceAgent := newLogicalDeviceAgent(clonedLD.Id, clonedLD.RootDeviceId, lDeviceMgr, deviceMgr, lDeviceMgr.clusterDataProxy, lDeviceMgr.defaultTimeout)
+ lDeviceAgent.logicalDevice = clonedLD
+ added := lDeviceAgent.clusterDataProxy.AddWithID(context.Background(), "/logical_devices", clonedLD.Id, clonedLD, "")
+ assert.NotNil(t, added)
+ lDeviceMgr.addLogicalDeviceAgentToMap(lDeviceAgent)
+ return lDeviceAgent
+}
+
+func (lda *LDATest) updateLogicalDeviceConcurrently(t *testing.T, ldAgent *LogicalDeviceAgent, globalWG *sync.WaitGroup) {
+ originalLogicalDevice := ldAgent.GetLogicalDevice()
+ assert.NotNil(t, originalLogicalDevice)
+ var localWG sync.WaitGroup
+
+ // Change the state of the first port to FAILED
+ localWG.Add(1)
+ go func() {
+ err := ldAgent.updatePortState(lda.logicalDevice.Ports[0].DeviceId, lda.logicalDevice.Ports[0].DevicePortNo, voltha.OperStatus_FAILED)
+ assert.Nil(t, err)
+ localWG.Done()
+ }()
+
+ // Change the state of the second port to TESTING
+ localWG.Add(1)
+ go func() {
+ err := ldAgent.updatePortState(lda.logicalDevice.Ports[1].DeviceId, lda.logicalDevice.Ports[1].DevicePortNo, voltha.OperStatus_TESTING)
+ assert.Nil(t, err)
+ localWG.Done()
+ }()
+
+ // Change the state of the third port to UNKNOWN and then back to ACTIVE
+ localWG.Add(1)
+ go func() {
+ err := ldAgent.updatePortState(lda.logicalDevice.Ports[2].DeviceId, lda.logicalDevice.Ports[2].DevicePortNo, voltha.OperStatus_UNKNOWN)
+ assert.Nil(t, err)
+ err = ldAgent.updatePortState(lda.logicalDevice.Ports[2].DeviceId, lda.logicalDevice.Ports[2].DevicePortNo, voltha.OperStatus_ACTIVE)
+ assert.Nil(t, err)
+ localWG.Done()
+ }()
+
+ // Add a meter to the logical device
+ meterMod := &ofp.OfpMeterMod{
+ Command: ofp.OfpMeterModCommand_OFPMC_ADD,
+ Flags: rand.Uint32(),
+ MeterId: rand.Uint32(),
+ Bands: []*ofp.OfpMeterBandHeader{
+ {Type: ofp.OfpMeterBandType_OFPMBT_EXPERIMENTER,
+ Rate: rand.Uint32(),
+ BurstSize: rand.Uint32(),
+ Data: nil,
+ },
+ },
+ }
+ localWG.Add(1)
+ go func() {
+ err := ldAgent.meterAdd(meterMod)
+ assert.Nil(t, err)
+ localWG.Done()
+ }()
+
+ // wait for go routines to be done
+ localWG.Wait()
+
+ expectedChange := proto.Clone(originalLogicalDevice).(*voltha.LogicalDevice)
+ expectedChange.Ports[0].OfpPort.Config = originalLogicalDevice.Ports[0].OfpPort.Config | uint32(ofp.OfpPortConfig_OFPPC_PORT_DOWN)
+ expectedChange.Ports[0].OfpPort.State = uint32(ofp.OfpPortState_OFPPS_LINK_DOWN)
+ expectedChange.Ports[1].OfpPort.Config = originalLogicalDevice.Ports[0].OfpPort.Config | uint32(ofp.OfpPortConfig_OFPPC_PORT_DOWN)
+ expectedChange.Ports[1].OfpPort.State = uint32(ofp.OfpPortState_OFPPS_LINK_DOWN)
+ expectedChange.Ports[2].OfpPort.Config = originalLogicalDevice.Ports[0].OfpPort.Config & ^uint32(ofp.OfpPortConfig_OFPPC_PORT_DOWN)
+ expectedChange.Ports[2].OfpPort.State = uint32(ofp.OfpPortState_OFPPS_LIVE)
+ expectedChange.Meters = &voltha.Meters{Items: nil}
+ expectedChange.Meters.Items = append(expectedChange.Meters.Items, fu.MeterEntryFromMeterMod(meterMod))
+ updatedLogicalDevice := ldAgent.GetLogicalDevice()
+ assert.NotNil(t, updatedLogicalDevice)
+ assert.True(t, proto.Equal(expectedChange, updatedLogicalDevice))
+ globalWG.Done()
+}
+
+func TestConcurrentLogicalDeviceUpdate(t *testing.T) {
+ lda := newLDATest()
+ assert.NotNil(t, lda)
+ defer lda.stopAll()
+
+ // Start the Core
+ lda.startCore(false)
+
+ var wg sync.WaitGroup
+ numConCurrentLogicalDeviceAgents := 20
+ for i := 0; i < numConCurrentLogicalDeviceAgents; i++ {
+ wg.Add(1)
+ a := lda.createLogicalDeviceAgent(t)
+ go lda.updateLogicalDeviceConcurrently(t, a, &wg)
+ }
+
+ wg.Wait()
+}
diff --git a/rw_core/core/logical_device_manager.go b/rw_core/core/logical_device_manager.go
index 8c326a8..53711d7 100644
--- a/rw_core/core/logical_device_manager.go
+++ b/rw_core/core/logical_device_manager.go
@@ -119,7 +119,7 @@
func (ldMgr *LogicalDeviceManager) getLogicalDevice(id string) (*voltha.LogicalDevice, error) {
log.Debugw("getlogicalDevice", log.Fields{"logicaldeviceid": id})
if agent := ldMgr.getLogicalDeviceAgent(id); agent != nil {
- return agent.GetLogicalDevice()
+ return agent.GetLogicalDevice(), nil
}
return nil, status.Errorf(codes.NotFound, "%s", id)
}
@@ -129,7 +129,7 @@
result := &voltha.LogicalDevices{}
ldMgr.logicalDeviceAgents.Range(func(key, value interface{}) bool {
agent := value.(*LogicalDeviceAgent)
- if ld, _ := agent.GetLogicalDevice(); ld != nil {
+ if ld := agent.GetLogicalDevice(); ld != nil {
result.Items = append(result.Items, ld)
}
return true
@@ -336,7 +336,7 @@
func (ldMgr *LogicalDeviceManager) ListLogicalDeviceFlows(ctx context.Context, id string) (*openflow_13.Flows, error) {
log.Debugw("ListLogicalDeviceFlows", log.Fields{"logicaldeviceid": id})
if agent := ldMgr.getLogicalDeviceAgent(id); agent != nil {
- return agent.ListLogicalDeviceFlows()
+ return agent.ListLogicalDeviceFlows(), nil
}
return nil, status.Errorf(codes.NotFound, "%s", id)
}
@@ -345,7 +345,7 @@
func (ldMgr *LogicalDeviceManager) ListLogicalDeviceFlowGroups(ctx context.Context, id string) (*openflow_13.FlowGroups, error) {
log.Debugw("ListLogicalDeviceFlowGroups", log.Fields{"logicaldeviceid": id})
if agent := ldMgr.getLogicalDeviceAgent(id); agent != nil {
- return agent.ListLogicalDeviceFlowGroups()
+ return agent.ListLogicalDeviceFlowGroups(), nil
}
return nil, status.Errorf(codes.NotFound, "%s", id)
}
@@ -354,7 +354,7 @@
func (ldMgr *LogicalDeviceManager) ListLogicalDevicePorts(ctx context.Context, id string) (*voltha.LogicalPorts, error) {
log.Debugw("ListLogicalDevicePorts", log.Fields{"logicaldeviceid": id})
if agent := ldMgr.getLogicalDeviceAgent(id); agent != nil {
- return agent.ListLogicalDevicePorts()
+ return agent.ListLogicalDevicePorts(), nil
}
return nil, status.Errorf(codes.NotFound, "%s", id)
}
@@ -540,7 +540,7 @@
func (ldMgr *LogicalDeviceManager) ListLogicalDeviceMeters(ctx context.Context, id string) (*openflow_13.Meters, error) {
log.Debugw("ListLogicalDeviceMeters", log.Fields{"logicalDeviceId": id})
if agent := ldMgr.getLogicalDeviceAgent(id); agent != nil {
- return agent.ListLogicalDeviceMeters()
+ return agent.ListLogicalDeviceMeters(), nil
}
return nil, status.Errorf(codes.NotFound, "%s", id)
}
diff --git a/rw_core/coreif/logical_device_agent_if.go b/rw_core/coreif/logical_device_agent_if.go
index 34a71e5..167d309 100644
--- a/rw_core/coreif/logical_device_agent_if.go
+++ b/rw_core/coreif/logical_device_agent_if.go
@@ -27,7 +27,7 @@
// LogicalDeviceAgent represents a generic agent
type LogicalDeviceAgent interface {
- GetLogicalDevice() (*voltha.LogicalDevice, error)
+ GetLogicalDevice() *voltha.LogicalDevice
GetDeviceGraph() *graph.DeviceGraph
GetWildcardInputPorts(excludePort ...uint32) []uint32
GetRoute(ingressPortNo uint32, egressPortNo uint32) []graph.RouteHop
diff --git a/rw_core/flowdecomposition/flow_decomposer_test.go b/rw_core/flowdecomposition/flow_decomposer_test.go
index cc89331..cf05549 100644
--- a/rw_core/flowdecomposition/flow_decomposer_test.go
+++ b/rw_core/flowdecomposition/flow_decomposer_test.go
@@ -401,8 +401,8 @@
return ""
}
-func (tfd *testFlowDecomposer) GetLogicalDevice() (*voltha.LogicalDevice, error) {
- return nil, nil
+func (tfd *testFlowDecomposer) GetLogicalDevice() *voltha.LogicalDevice {
+ return nil
}
func (tfd *testFlowDecomposer) GetDeviceGraph() *graph.DeviceGraph {
diff --git a/rw_core/mocks/adapter_onu.go b/rw_core/mocks/adapter_onu.go
index bb6f58a..6c7e35b 100644
--- a/rw_core/mocks/adapter_onu.go
+++ b/rw_core/mocks/adapter_onu.go
@@ -92,7 +92,7 @@
Type: voltha.Port_PON_ONU,
OperStatus: voltha.OperStatus_ACTIVE,
Peers: []*voltha.Port_PeerPort{{DeviceId: d.ParentId, // Peer device is OLT
- PortNo: uniPortNo}}, // Peer port is UNI port
+ PortNo: device.ParentPortNo}}, // Peer port is parent's port number
}
if err = onuA.coreProxy.PortCreated(context.TODO(), d.Id, ponPort); err != nil {
log.Fatalf("PortCreated-failed-%s", err)
diff --git a/vendor/github.com/opencord/voltha-lib-go/v2/pkg/db/common.go b/vendor/github.com/opencord/voltha-lib-go/v2/pkg/db/common.go
new file mode 100644
index 0000000..0851ede
--- /dev/null
+++ b/vendor/github.com/opencord/voltha-lib-go/v2/pkg/db/common.go
@@ -0,0 +1,34 @@
+/*
+ * 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 db
+
+import (
+ "github.com/opencord/voltha-lib-go/v2/pkg/log"
+)
+
+const (
+ logLevel = log.FatalLevel
+)
+
+// Unit test initialization. This init() function handles all unit tests in
+// the current directory.
+func init() {
+ // Setup this package so that it's log level can be modified at run time
+ _, err := log.AddPackage(log.JSON, logLevel, log.Fields{"pkg": "db"})
+ if err != nil {
+ panic(err)
+ }
+}
diff --git a/vendor/github.com/opencord/voltha-lib-go/v2/pkg/kafka/sarama_client.go b/vendor/github.com/opencord/voltha-lib-go/v2/pkg/kafka/sarama_client.go
index 73025d9..ff521a7 100644
--- a/vendor/github.com/opencord/voltha-lib-go/v2/pkg/kafka/sarama_client.go
+++ b/vendor/github.com/opencord/voltha-lib-go/v2/pkg/kafka/sarama_client.go
@@ -907,7 +907,10 @@
select {
case err, ok := <-consumer.Errors():
if ok {
- log.Warnw("partition-consumers-error", log.Fields{"error": err})
+ if sc.isLivenessError(err) {
+ sc.updateLiveness(false)
+ log.Warnw("partition-consumers-error", log.Fields{"error": err})
+ }
} else {
// Channel is closed
break startloop
@@ -919,6 +922,8 @@
break startloop
}
msgBody := msg.Value
+ sc.updateLiveness(true)
+ log.Debugw("message-received", log.Fields{"timestamp": msg.Timestamp, "receivedTopic": msg.Topic})
icm := &ic.InterContainerMessage{}
if err := proto.Unmarshal(msgBody, icm); err != nil {
log.Warnw("partition-invalid-message", log.Fields{"error": err})
diff --git a/vendor/github.com/opencord/voltha-lib-go/v2/pkg/mocks/common.go b/vendor/github.com/opencord/voltha-lib-go/v2/pkg/mocks/common.go
new file mode 100644
index 0000000..c20dab1
--- /dev/null
+++ b/vendor/github.com/opencord/voltha-lib-go/v2/pkg/mocks/common.go
@@ -0,0 +1,34 @@
+/*
+ * 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 mocks
+
+import (
+ "github.com/opencord/voltha-lib-go/v2/pkg/log"
+)
+
+const (
+ logLevel = log.FatalLevel
+)
+
+// Unit test initialization. This init() function handles all unit tests in
+// the current directory.
+func init() {
+ // Setup this package so that it's log level can be modified at run time
+ _, err := log.AddPackage(log.JSON, logLevel, log.Fields{"pkg": "mocks"})
+ if err != nil {
+ panic(err)
+ }
+}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 35803f5..7286932 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -100,7 +100,7 @@
github.com/modern-go/concurrent
# github.com/modern-go/reflect2 v1.0.1
github.com/modern-go/reflect2
-# github.com/opencord/voltha-lib-go/v2 v2.2.18
+# github.com/opencord/voltha-lib-go/v2 v2.2.21
github.com/opencord/voltha-lib-go/v2/pkg/log
github.com/opencord/voltha-lib-go/v2/pkg/db
github.com/opencord/voltha-lib-go/v2/pkg/db/kvstore