[VOL-4291] Rw-core updates for gRPC migration
Change-Id: I8d5a554409115b29318089671ca4e1ab3fa98810
diff --git a/rw_core/core/device/manager.go b/rw_core/core/device/manager.go
index 0b34bfb..5519cab 100755
--- a/rw_core/core/device/manager.go
+++ b/rw_core/core/device/manager.go
@@ -18,27 +18,24 @@
import (
"context"
- "errors"
- "github.com/opencord/voltha-go/rw_core/config"
- "github.com/opencord/voltha-lib-go/v5/pkg/probe"
+ "fmt"
"sync"
"time"
- "github.com/golang/protobuf/ptypes/empty"
+ "github.com/opencord/voltha-go/rw_core/config"
+ "github.com/opencord/voltha-lib-go/v7/pkg/probe"
+ "github.com/opencord/voltha-protos/v5/go/core"
+
"github.com/opencord/voltha-go/db/model"
"github.com/opencord/voltha-go/rw_core/core/adapter"
"github.com/opencord/voltha-go/rw_core/core/device/event"
- "github.com/opencord/voltha-go/rw_core/core/device/remote"
"github.com/opencord/voltha-go/rw_core/core/device/state"
"github.com/opencord/voltha-go/rw_core/utils"
- "github.com/opencord/voltha-lib-go/v5/pkg/events"
- "github.com/opencord/voltha-lib-go/v5/pkg/kafka"
- "github.com/opencord/voltha-lib-go/v5/pkg/log"
- "github.com/opencord/voltha-protos/v4/go/common"
- ic "github.com/opencord/voltha-protos/v4/go/inter_container"
- "github.com/opencord/voltha-protos/v4/go/openflow_13"
- ofp "github.com/opencord/voltha-protos/v4/go/openflow_13"
- "github.com/opencord/voltha-protos/v4/go/voltha"
+ "github.com/opencord/voltha-lib-go/v7/pkg/events"
+ "github.com/opencord/voltha-lib-go/v7/pkg/log"
+ ic "github.com/opencord/voltha-protos/v5/go/inter_container"
+ ofp "github.com/opencord/voltha-protos/v5/go/openflow_13"
+ "github.com/opencord/voltha-protos/v5/go/voltha"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)
@@ -48,32 +45,30 @@
deviceAgents sync.Map
rootDevices map[string]bool
lockRootDeviceMap sync.RWMutex
- adapterProxy *remote.AdapterProxy
*event.Agent
adapterMgr *adapter.Manager
logicalDeviceMgr *LogicalManager
- kafkaICProxy kafka.InterContainerProxy
stateTransitions *state.TransitionMap
dbPath *model.Path
dProxy *model.Proxy
coreInstanceID string
- defaultTimeout time.Duration
+ internalTimeout time.Duration
+ rpcTimeout time.Duration
devicesLoadingLock sync.RWMutex
deviceLoadingInProgress map[string][]chan int
config *config.RWCoreFlags
}
//NewManagers creates the Manager and the Logical Manager.
-func NewManagers(dbPath *model.Path, adapterMgr *adapter.Manager, kmp kafka.InterContainerProxy, endpointMgr kafka.EndpointManager, cf *config.RWCoreFlags, coreInstanceID string, eventProxy *events.EventProxy) (*Manager, *LogicalManager) {
+func NewManagers(dbPath *model.Path, adapterMgr *adapter.Manager, cf *config.RWCoreFlags, coreInstanceID string, eventProxy *events.EventProxy) (*Manager, *LogicalManager) {
deviceMgr := &Manager{
rootDevices: make(map[string]bool),
- kafkaICProxy: kmp,
- adapterProxy: remote.NewAdapterProxy(kmp, cf.CoreTopic, endpointMgr),
coreInstanceID: coreInstanceID,
dbPath: dbPath,
dProxy: dbPath.Proxy("devices"),
adapterMgr: adapterMgr,
- defaultTimeout: cf.DefaultCoreTimeout,
+ internalTimeout: cf.InternalTimeout,
+ rpcTimeout: cf.RPCTimeout,
Agent: event.NewAgent(eventProxy, coreInstanceID, cf.VolthaStackID),
deviceLoadingInProgress: make(map[string][]chan int),
config: cf,
@@ -83,33 +78,40 @@
logicalDeviceMgr := &LogicalManager{
Manager: event.NewManager(eventProxy, coreInstanceID, cf.VolthaStackID),
deviceMgr: deviceMgr,
- kafkaICProxy: kmp,
dbPath: dbPath,
ldProxy: dbPath.Proxy("logical_devices"),
- defaultTimeout: cf.DefaultCoreTimeout,
+ internalTimeout: cf.InternalTimeout,
logicalDeviceLoadingInProgress: make(map[string][]chan int),
}
deviceMgr.logicalDeviceMgr = logicalDeviceMgr
- adapterMgr.SetAdapterRestartedCallback(deviceMgr.adapterRestarted)
+ adapterMgr.SetAdapterRestartedCallback(deviceMgr.adapterRestartedHandler)
return deviceMgr, logicalDeviceMgr
}
-func (dMgr *Manager) Start(ctx context.Context) {
+func (dMgr *Manager) Start(ctx context.Context, serviceName string) error {
logger.Info(ctx, "starting-device-manager")
- probe.UpdateStatusFromContext(ctx, "device-manager", probe.ServiceStatusPreparing)
+ probe.UpdateStatusFromContext(ctx, serviceName, probe.ServiceStatusPreparing)
// Load all the devices from the dB
var devices []*voltha.Device
if err := dMgr.dProxy.List(ctx, &devices); err != nil {
// Any error from the dB means if we proceed we may end up with corrupted data
- logger.Fatalw(ctx, "failed-to-list-devices-from-KV", log.Fields{"error": err})
+ logger.Errorw(ctx, "failed-to-list-devices-from-KV", log.Fields{"error": err, "service-name": serviceName})
+ return err
+ }
+
+ defer probe.UpdateStatusFromContext(ctx, serviceName, probe.ServiceStatusRunning)
+
+ if len(devices) == 0 {
+ logger.Info(ctx, "no-device-to-load")
+ return nil
}
for _, device := range devices {
// Create an agent for each device
- agent := newAgent(dMgr.adapterProxy, device, dMgr, dMgr.dbPath, dMgr.dProxy, dMgr.defaultTimeout)
+ agent := newAgent(device, dMgr, dMgr.dbPath, dMgr.dProxy, dMgr.internalTimeout, dMgr.rpcTimeout)
if _, err := agent.start(ctx, true, device); err != nil {
logger.Warnw(ctx, "failure-starting-agent", log.Fields{"device-id": device.Id})
} else {
@@ -119,8 +121,9 @@
// TODO: Need to trigger a reconcile at this point
- probe.UpdateStatusFromContext(ctx, "device-manager", probe.ServiceStatusRunning)
logger.Info(ctx, "device-manager-started")
+
+ return nil
}
func (dMgr *Manager) addDeviceAgentToMap(agent *Agent) {
@@ -172,165 +175,6 @@
return result
}
-// CreateDevice creates a new parent device in the data model
-func (dMgr *Manager) CreateDevice(ctx context.Context, device *voltha.Device) (*voltha.Device, error) {
- if device.MacAddress == "" && device.GetHostAndPort() == "" {
- logger.Errorf(ctx, "no-device-info-present")
- return &voltha.Device{}, errors.New("no-device-info-present; MAC or HOSTIP&PORT")
- }
- ctx = utils.WithRPCMetadataContext(ctx, "CreateDevice")
- logger.Debugw(ctx, "create-device", log.Fields{"device": *device})
-
- deviceExist, err := dMgr.isParentDeviceExist(ctx, device)
- if err != nil {
- logger.Errorf(ctx, "failed-to-fetch-parent-device-info")
- return nil, err
- }
- if deviceExist {
- logger.Errorf(ctx, "device-is-pre-provisioned-already-with-same-ip-port-or-mac-address")
- return nil, errors.New("device is already pre-provisioned")
- }
- logger.Debugw(ctx, "create-device", log.Fields{"device": device, "aproxy": dMgr.adapterProxy})
-
- // Ensure this device is set as root
- device.Root = true
- // Create and start a device agent for that device
- agent := newAgent(dMgr.adapterProxy, device, dMgr, dMgr.dbPath, dMgr.dProxy, dMgr.defaultTimeout)
- device, err = agent.start(ctx, false, device)
- if err != nil {
- logger.Errorw(ctx, "fail-to-start-device", log.Fields{"device-id": agent.deviceID, "error": err})
- return nil, err
- }
- dMgr.addDeviceAgentToMap(agent)
- return device, nil
-}
-
-// EnableDevice activates a device by invoking the adopt_device API on the appropriate adapter
-func (dMgr *Manager) EnableDevice(ctx context.Context, id *voltha.ID) (*empty.Empty, error) {
- ctx = utils.WithRPCMetadataContext(ctx, "EnableDevice")
- log.EnrichSpan(ctx, log.Fields{"device-id": id.Id})
- logger.Debugw(ctx, "enable-device", log.Fields{"device-id": id.Id})
- agent := dMgr.getDeviceAgent(ctx, id.Id)
- if agent == nil {
- return nil, status.Errorf(codes.NotFound, "%s", id.Id)
- }
- return &empty.Empty{}, agent.enableDevice(ctx)
-}
-
-// DisableDevice disables a device along with any child device it may have
-func (dMgr *Manager) DisableDevice(ctx context.Context, id *voltha.ID) (*empty.Empty, error) {
- ctx = utils.WithRPCMetadataContext(ctx, "DisableDevice")
- log.EnrichSpan(ctx, log.Fields{"device-id": id.Id})
- logger.Debugw(ctx, "disable-device", log.Fields{"device-id": id.Id})
- agent := dMgr.getDeviceAgent(ctx, id.Id)
- if agent == nil {
- return nil, status.Errorf(codes.NotFound, "%s", id.Id)
- }
- return &empty.Empty{}, agent.disableDevice(ctx)
-}
-
-//RebootDevice invoked the reboot API to the corresponding adapter
-func (dMgr *Manager) RebootDevice(ctx context.Context, id *voltha.ID) (*empty.Empty, error) {
- ctx = utils.WithRPCMetadataContext(ctx, "RebootDevice")
- log.EnrichSpan(ctx, log.Fields{"device-id": id.Id})
- logger.Debugw(ctx, "reboot-device", log.Fields{"device-id": id.Id})
- agent := dMgr.getDeviceAgent(ctx, id.Id)
- if agent == nil {
- return nil, status.Errorf(codes.NotFound, "%s", id.Id)
- }
- return &empty.Empty{}, agent.rebootDevice(ctx)
-}
-
-// DeleteDevice removes a device from the data model
-func (dMgr *Manager) DeleteDevice(ctx context.Context, id *voltha.ID) (*empty.Empty, error) {
- ctx = utils.WithRPCMetadataContext(ctx, "DeleteDevice")
- log.EnrichSpan(ctx, log.Fields{"device-id": id.Id})
- logger.Debugw(ctx, "delete-device", log.Fields{"device-id": id.Id})
- agent := dMgr.getDeviceAgent(ctx, id.Id)
- if agent == nil {
- return nil, status.Errorf(codes.NotFound, "%s", id.Id)
- }
- return &empty.Empty{}, agent.deleteDevice(ctx)
-}
-
-// ForceDeleteDevice removes a device from the data model forcefully without successfully waiting for the adapters.
-func (dMgr *Manager) ForceDeleteDevice(ctx context.Context, id *voltha.ID) (*empty.Empty, error) {
- ctx = utils.WithRPCMetadataContext(ctx, "ForceDeleteDevice")
- log.EnrichSpan(ctx, log.Fields{"device-id": id.Id})
- logger.Debugw(ctx, "force-delete-device", log.Fields{"device-id": id.Id})
- agent := dMgr.getDeviceAgent(ctx, id.Id)
- if agent == nil {
- return nil, status.Errorf(codes.NotFound, "%s", id.Id)
- }
- return &empty.Empty{}, agent.deleteDeviceForce(ctx)
-}
-
-// GetDevicePort returns the port details for a specific device port entry
-func (dMgr *Manager) GetDevicePort(ctx context.Context, deviceID string, portID uint32) (*voltha.Port, error) {
- logger.Debugw(ctx, "get-device-port", log.Fields{"device-id": deviceID})
- agent := dMgr.getDeviceAgent(ctx, deviceID)
- if agent == nil {
- return nil, status.Errorf(codes.NotFound, "device-%s", deviceID)
- }
- return agent.getDevicePort(portID)
-}
-
-// ListDevicePorts returns the ports details for a specific device entry
-func (dMgr *Manager) ListDevicePorts(ctx context.Context, id *voltha.ID) (*voltha.Ports, error) {
- ctx = utils.WithRPCMetadataContext(ctx, "ListDevicePorts")
- log.EnrichSpan(ctx, log.Fields{"device-id": id.Id})
- logger.Debugw(ctx, "list-device-ports", log.Fields{"device-id": id.Id})
- agent := dMgr.getDeviceAgent(ctx, id.Id)
- if agent == nil {
- return nil, status.Errorf(codes.NotFound, "device-%s", id.Id)
- }
-
- ports := agent.listDevicePorts()
- ctr, ret := 0, make([]*voltha.Port, len(ports))
- for _, port := range ports {
- ret[ctr] = port
- ctr++
- }
- return &voltha.Ports{Items: ret}, nil
-}
-
-// ListDeviceFlows returns the flow details for a specific device entry
-func (dMgr *Manager) ListDeviceFlows(ctx context.Context, id *voltha.ID) (*ofp.Flows, error) {
- ctx = utils.WithRPCMetadataContext(ctx, "ListDeviceFlows")
- log.EnrichSpan(ctx, log.Fields{"device-id": id.Id})
- logger.Debugw(ctx, "list-device-flows", log.Fields{"device-id": id.Id})
- agent := dMgr.getDeviceAgent(ctx, id.Id)
- if agent == nil {
- return nil, status.Errorf(codes.NotFound, "device-%s", id.Id)
- }
-
- flows := agent.listDeviceFlows()
- ctr, ret := 0, make([]*ofp.OfpFlowStats, len(flows))
- for _, flow := range flows {
- ret[ctr] = flow
- ctr++
- }
- return &openflow_13.Flows{Items: ret}, nil
-}
-
-// ListDeviceFlowGroups returns the flow group details for a specific device entry
-func (dMgr *Manager) ListDeviceFlowGroups(ctx context.Context, id *voltha.ID) (*voltha.FlowGroups, error) {
- ctx = utils.WithRPCMetadataContext(ctx, "ListDeviceFlowGroups")
- log.EnrichSpan(ctx, log.Fields{"device-id": id.Id})
- logger.Debugw(ctx, "list-device-flow-groups", log.Fields{"device-id": id.Id})
- agent := dMgr.getDeviceAgent(ctx, id.Id)
- if agent == nil {
- return nil, status.Errorf(codes.NotFound, "device-%s", id.Id)
- }
- groups := agent.listDeviceGroups()
- ctr, ret := 0, make([]*openflow_13.OfpGroupEntry, len(groups))
- for _, group := range groups {
- ret[ctr] = group
- ctr++
- }
- return &voltha.FlowGroups{Items: ret}, nil
-}
-
// stopManagingDevice stops the management of the device as well as any of its reference device and logical device.
// This function is called only in the Core that does not own this device. In the Core that owns this device then a
// deletion deletion also includes removal of any reference of this device.
@@ -350,21 +194,6 @@
}
}
-// RunPostDeviceDelete removes any reference of this device
-func (dMgr *Manager) RunPostDeviceDelete(ctx context.Context, cDevice *voltha.Device) error {
- logger.Infow(ctx, "run-post-device-delete", log.Fields{"device-id": cDevice.Id})
- dMgr.stopManagingDevice(ctx, cDevice.Id)
- return nil
-}
-
-// GetDevice exists primarily to implement the gRPC interface.
-// Internal functions should call getDeviceReadOnly instead.
-func (dMgr *Manager) GetDevice(ctx context.Context, id *voltha.ID) (*voltha.Device, error) {
- ctx = utils.WithRPCMetadataContext(ctx, "GetDevice")
- log.EnrichSpan(ctx, log.Fields{"device-id": id.Id})
- return dMgr.getDeviceReadOnly(ctx, id.Id)
-}
-
// getDeviceReadOnly will returns a device, either from memory or from the dB, if present
func (dMgr *Manager) getDeviceReadOnly(ctx context.Context, id string) (*voltha.Device, error) {
logger.Debugw(ctx, "get-device-read-only", log.Fields{"device-id": id})
@@ -383,119 +212,12 @@
return agent.listDevicePorts(), nil
}
-// GetChildDevice will return a device, either from memory or from the dB, if present
-func (dMgr *Manager) GetChildDevice(ctx context.Context, parentDeviceID string, serialNumber string, onuID int64, parentPortNo int64) (*voltha.Device, error) {
- logger.Debugw(ctx, "get-child-device", log.Fields{"parent-device-id": parentDeviceID, "serialNumber": serialNumber,
- "parent-port-no": parentPortNo, "onu-id": onuID})
-
- parentDevicePorts, err := dMgr.listDevicePorts(ctx, parentDeviceID)
- if err != nil {
- return nil, status.Errorf(codes.Aborted, "%s", err.Error())
- }
- childDeviceIds := dMgr.getAllChildDeviceIds(ctx, parentDevicePorts)
- if len(childDeviceIds) == 0 {
- logger.Debugw(ctx, "no-child-devices", log.Fields{"parent-device-id": parentDeviceID, "serial-number": serialNumber, "onu-id": onuID})
- return nil, status.Errorf(codes.NotFound, "%s", parentDeviceID)
- }
-
- var foundChildDevice *voltha.Device
- for childDeviceID := range childDeviceIds {
- var found bool
- if searchDevice, err := dMgr.getDeviceReadOnly(ctx, childDeviceID); err == nil {
-
- foundOnuID := false
- if searchDevice.ProxyAddress.OnuId == uint32(onuID) {
- if searchDevice.ParentPortNo == uint32(parentPortNo) {
- logger.Debugw(ctx, "found-child-by-onuid", log.Fields{"parent-device-id": parentDeviceID, "onu-id": onuID})
- foundOnuID = true
- }
- }
-
- foundSerialNumber := false
- if searchDevice.SerialNumber == serialNumber {
- logger.Debugw(ctx, "found-child-by-serial-number", log.Fields{"parent-device-id": parentDeviceID, "serial-number": serialNumber})
- foundSerialNumber = true
- }
-
- // if both onuId and serialNumber are provided both must be true for the device to be found
- // otherwise whichever one found a match is good enough
- if onuID > 0 && serialNumber != "" {
- found = foundOnuID && foundSerialNumber
- } else {
- found = foundOnuID || foundSerialNumber
- }
-
- if found {
- foundChildDevice = searchDevice
- break
- }
- }
- }
-
- if foundChildDevice != nil {
- logger.Debugw(ctx, "child-device-found", log.Fields{"parent-device-id": parentDeviceID, "found-child-device": foundChildDevice})
- return foundChildDevice, nil
- }
-
- logger.Debugw(ctx, "child-device-not-found", log.Fields{"parent-device-id": parentDeviceID,
- "serial-number": serialNumber, "onu-id": onuID, "parent-port-no": parentPortNo})
- return nil, status.Errorf(codes.NotFound, "%s", parentDeviceID)
-}
-
-// GetChildDeviceWithProxyAddress will return a device based on proxy address
-func (dMgr *Manager) GetChildDeviceWithProxyAddress(ctx context.Context, proxyAddress *voltha.Device_ProxyAddress) (*voltha.Device, error) {
- logger.Debugw(ctx, "get-child-device-with-proxy-address", log.Fields{"proxy-address": proxyAddress})
-
- parentDevicePorts, err := dMgr.listDevicePorts(ctx, proxyAddress.DeviceId)
- if err != nil {
- return nil, status.Errorf(codes.Aborted, "%s", err.Error())
- }
- childDeviceIds := dMgr.getAllChildDeviceIds(ctx, parentDevicePorts)
- if len(childDeviceIds) == 0 {
- logger.Debugw(ctx, "no-child-devices", log.Fields{"parent-device-id": proxyAddress.DeviceId})
- return nil, status.Errorf(codes.NotFound, "%s", proxyAddress)
- }
-
- var foundChildDevice *voltha.Device
- for childDeviceID := range childDeviceIds {
- if searchDevice, err := dMgr.getDeviceReadOnly(ctx, childDeviceID); err == nil {
- if searchDevice.ProxyAddress == proxyAddress {
- foundChildDevice = searchDevice
- break
- }
- }
- }
-
- if foundChildDevice != nil {
- logger.Debugw(ctx, "child-device-found", log.Fields{"proxy-address": proxyAddress})
- return foundChildDevice, nil
- }
-
- logger.Warnw(ctx, "child-device-not-found", log.Fields{"proxy-address": proxyAddress})
- return nil, status.Errorf(codes.NotFound, "%s", proxyAddress)
-}
-
// IsDeviceInCache returns true if device is found in the map
func (dMgr *Manager) IsDeviceInCache(id string) bool {
_, exist := dMgr.deviceAgents.Load(id)
return exist
}
-// ListDevices retrieves the latest devices from the data model
-func (dMgr *Manager) ListDevices(ctx context.Context, _ *empty.Empty) (*voltha.Devices, error) {
- ctx = utils.WithRPCMetadataContext(ctx, "ListDevices")
- logger.Debug(ctx, "list-devices")
- result := &voltha.Devices{}
-
- dMgr.deviceAgents.Range(func(key, value interface{}) bool {
- result.Items = append(result.Items, value.(*Agent).device)
- return true
- })
-
- logger.Debugw(ctx, "list-devices-end", log.Fields{"len": len(result.Items)})
- return result, nil
-}
-
//isParentDeviceExist checks whether device is already preprovisioned.
func (dMgr *Manager) isParentDeviceExist(ctx context.Context, newDevice *voltha.Device) (bool, error) {
hostPort := newDevice.GetHostAndPort()
@@ -547,7 +269,7 @@
// Proceed with the loading only if the device exist in the Model (could have been deleted)
if device, err = dMgr.getDeviceFromModel(ctx, deviceID); err == nil {
logger.Debugw(ctx, "loading-device", log.Fields{"device-id": deviceID})
- agent := newAgent(dMgr.adapterProxy, device, dMgr, dMgr.dbPath, dMgr.dProxy, dMgr.defaultTimeout)
+ agent := newAgent(device, dMgr, dMgr.dbPath, dMgr.dProxy, dMgr.internalTimeout, dMgr.rpcTimeout)
if _, err = agent.start(ctx, true, device); err != nil {
logger.Warnw(ctx, "failure-loading-device", log.Fields{"device-id": deviceID, "error": err})
} else {
@@ -648,132 +370,34 @@
return nil
}
-// ListDeviceIds retrieves the latest device IDs information from the data model (memory data only)
-func (dMgr *Manager) ListDeviceIds(ctx context.Context, _ *empty.Empty) (*voltha.IDs, error) {
- ctx = utils.WithRPCMetadataContext(ctx, "ListDeviceIds")
- logger.Debug(ctx, "list-device-ids")
- // Report only device IDs that are in the device agent map
- return dMgr.listDeviceIdsFromMap(), nil
-}
-
-// ReconcileDevices is a request to a voltha core to update its list of managed devices. This will
-// trigger loading the devices along with their children and parent in memory
-func (dMgr *Manager) ReconcileDevices(ctx context.Context, ids *voltha.IDs) (*empty.Empty, error) {
- ctx = utils.WithRPCMetadataContext(ctx, "ReconcileDevices")
-
- numDevices := 0
- if ids != nil {
- numDevices = len(ids.Items)
- }
-
- logger.Debugw(ctx, "reconcile-devices", log.Fields{"num-devices": numDevices})
- if ids != nil && len(ids.Items) != 0 {
- toReconcile := len(ids.Items)
- reconciled := 0
- var err error
- for _, id := range ids.Items {
- if err = dMgr.load(ctx, id.Id); err != nil {
- logger.Warnw(ctx, "failure-reconciling-device", log.Fields{"device-id": id.Id, "error": err})
- } else {
- reconciled++
- }
- }
- if toReconcile != reconciled {
- return nil, status.Errorf(codes.DataLoss, "less-device-reconciled-than-requested:%d/%d", reconciled, toReconcile)
- }
- } else {
- return nil, status.Errorf(codes.InvalidArgument, "empty-list-of-ids")
- }
- return &empty.Empty{}, nil
-}
-
// adapterRestarted is invoked whenever an adapter is restarted
func (dMgr *Manager) adapterRestarted(ctx context.Context, adapter *voltha.Adapter) error {
logger.Debugw(ctx, "adapter-restarted", log.Fields{"adapter-id": adapter.Id, "vendor": adapter.Vendor,
- "current-replica": adapter.CurrentReplica, "total-replicas": adapter.TotalReplicas, "endpoint": adapter.Endpoint})
+ "current-replica": adapter.CurrentReplica, "total-replicas": adapter.TotalReplicas, "restarted-endpoint": adapter.Endpoint})
- // Let's reconcile the device managed by this Core only
- if len(dMgr.rootDevices) == 0 {
- logger.Debugw(ctx, "nothing-to-reconcile", log.Fields{"adapter-id": adapter.Id})
- return nil
- }
-
- if len(dMgr.rootDevices) == 0 {
- logger.Debugw(ctx, "no-managed-device-to-reconcile", log.Fields{"adapter-id": adapter.Id})
- return nil
- }
-
- for rootDeviceID := range dMgr.rootDevices {
- dAgent := dMgr.getDeviceAgent(ctx, rootDeviceID)
- if dAgent == nil {
- continue
- }
- logger.Debugw(ctx, "checking-adapter-type", log.Fields{"agentType": dAgent.deviceType, "adapter-type": adapter.Type})
- if dAgent.deviceType == adapter.Type {
- rootDevice, _ := dAgent.getDeviceReadOnly(ctx)
- if rootDevice == nil {
- continue
- }
- isDeviceOwnedByService, err := dMgr.adapterProxy.IsDeviceOwnedByService(ctx, rootDeviceID, adapter.Type, adapter.CurrentReplica)
- if err != nil {
- logger.Warnw(ctx, "is-device-owned-by-service", log.Fields{"error": err, "root-device-id": rootDeviceID, "adapter-type": adapter.Type, "replica-number": adapter.CurrentReplica})
- continue
- }
- if isDeviceOwnedByService {
- if rootDevice.AdminState != voltha.AdminState_PREPROVISIONED {
- logger.Debugw(ctx, "reconciling-root-device", log.Fields{"rootId": rootDevice.Id})
- go dAgent.ReconcileDevice(ctx, rootDevice)
- } else {
- logger.Debugw(ctx, "not-reconciling-root-device", log.Fields{"rootId": rootDevice.Id, "state": rootDevice.AdminState})
- }
- } else { // Should we be reconciling the root's children instead?
- rootDevicePorts, _ := dMgr.listDevicePorts(ctx, rootDeviceID)
- childManagedByAdapter:
- for _, port := range rootDevicePorts {
- for _, peer := range port.Peers {
- if childDevice, _ := dMgr.getDeviceFromModel(ctx, peer.DeviceId); childDevice != nil {
- isDeviceOwnedByService, err := dMgr.adapterProxy.IsDeviceOwnedByService(ctx, childDevice.Id, adapter.Type, adapter.CurrentReplica)
- if err != nil {
- logger.Warnw(ctx, "is-device-owned-by-service", log.Fields{"error": err, "child-device-id": childDevice.Id, "adapter-type": adapter.Type, "replica-number": adapter.CurrentReplica})
- }
- if isDeviceOwnedByService {
- if childDevice.AdminState != voltha.AdminState_PREPROVISIONED {
- logger.Debugw(ctx, "reconciling-child-device", log.Fields{"child-device-id": childDevice.Id})
- go dAgent.ReconcileDevice(ctx, childDevice)
- } else {
- logger.Debugw(ctx, "not-reconciling-child-device", log.Fields{"child-device-id": childDevice.Id, "state": childDevice.AdminState})
- }
- } else {
- // All child devices under a parent device are typically managed by the same adapter type.
- // Therefore we only need to check whether the first device we retrieved is managed by that adapter
- break childManagedByAdapter
- }
- }
- }
- }
+ numberOfDevicesToReconcile := 0
+ dMgr.deviceAgents.Range(func(key, value interface{}) bool {
+ deviceAgent, ok := value.(*Agent)
+ if ok && deviceAgent.adapterEndpoint == adapter.Endpoint {
+ // Before reconciling, abort in-process request
+ if err := deviceAgent.abortAllProcessing(utils.WithNewSpanAndRPCMetadataContext(ctx, "AbortProcessingOnRestart")); err == nil {
+ logger.Debugw(ctx, "reconciling-device",
+ log.Fields{
+ "device-id": deviceAgent.deviceID,
+ "root-device": deviceAgent.isRootDevice,
+ "restarted-endpoint": adapter.Endpoint,
+ "device-type": deviceAgent.deviceType,
+ "adapter-type": adapter.Type,
+ })
+ go deviceAgent.ReconcileDevice(utils.WithNewSpanAndRPCMetadataContext(ctx, "ReconcileDevice"))
+ numberOfDevicesToReconcile++
+ } else {
+ logger.Errorw(ctx, "failed-aborting-exisiting-processing", log.Fields{"error": err})
}
}
- }
- logger.Debugw(ctx, "Reconciling for device on adapter restart is initiated", log.Fields{"adapter-id": adapter.Id})
-
- return nil
-}
-
-func (dMgr *Manager) ReconcileChildDevices(ctx context.Context, parentDeviceID string) error {
- dAgent := dMgr.getDeviceAgent(ctx, parentDeviceID)
- if dAgent == nil {
- return status.Errorf(codes.NotFound, "error-unable to get agent from device")
- }
- if parentDevicePorts, err := dMgr.listDevicePorts(ctx, parentDeviceID); err == nil {
- for _, port := range parentDevicePorts {
- for _, peer := range port.Peers {
- if childDevice, err := dMgr.getDeviceFromModel(ctx, peer.DeviceId); err == nil {
- go dAgent.ReconcileDevice(ctx, childDevice)
- }
- }
- }
- logger.Debugw(ctx, "Reconcile initiated for child devices", log.Fields{"parent-device-id": parentDeviceID})
- }
+ return true
+ })
+ logger.Debugw(ctx, "reconciling-on-adapter-restart-initiated", log.Fields{"adapter-endpoint": adapter.Endpoint, "number-of-devices-to-reconcile": numberOfDevicesToReconcile})
return nil
}
@@ -831,6 +455,28 @@
return status.Errorf(codes.NotFound, "%s", deviceID)
}
+func (dMgr *Manager) canMultipleAdapterRequestProceed(ctx context.Context, deviceIDs []string) error {
+ ready := len(deviceIDs) > 0
+ for _, deviceID := range deviceIDs {
+ agent := dMgr.getDeviceAgent(ctx, deviceID)
+ if agent == nil {
+ logger.Errorw(ctx, "adapter-nil", log.Fields{"device-id": deviceID})
+ return status.Errorf(codes.Unavailable, "adapter-nil-for-%s", deviceID)
+ }
+ ready = ready && agent.isAdapterConnectionUp(ctx)
+ if !ready {
+ return status.Errorf(codes.Unavailable, "adapter-connection-down-for-%s", deviceID)
+ }
+ if err := agent.canDeviceRequestProceed(ctx); err != nil {
+ return err
+ }
+ }
+ if !ready {
+ return status.Error(codes.Unavailable, "adapter(s)-not-ready")
+ }
+ return nil
+}
+
func (dMgr *Manager) addFlowsAndGroups(ctx context.Context, deviceID string, flows []*ofp.OfpFlowStats, groups []*ofp.OfpGroupEntry, flowMetadata *voltha.FlowMetadata) error {
logger.Debugw(ctx, "add-flows-and-groups", log.Fields{"device-id": deviceID, "groups:": groups, "flow-metadata": flowMetadata})
if agent := dMgr.getDeviceAgent(ctx, deviceID); agent != nil {
@@ -867,21 +513,6 @@
return status.Errorf(codes.NotFound, "%s", deviceID)
}
-// UpdateDevicePmConfigs updates the PM configs. This is executed when the northbound gRPC API is invoked, typically
-// following a user action
-func (dMgr *Manager) UpdateDevicePmConfigs(ctx context.Context, configs *voltha.PmConfigs) (*empty.Empty, error) {
- ctx = utils.WithRPCMetadataContext(ctx, "UpdateDevicePmConfigs")
- log.EnrichSpan(ctx, log.Fields{"device-id": configs.Id})
- if configs.Id == "" {
- return nil, status.Error(codes.FailedPrecondition, "invalid-device-Id")
- }
- agent := dMgr.getDeviceAgent(ctx, configs.Id)
- if agent == nil {
- return nil, status.Errorf(codes.NotFound, "%s", configs.Id)
- }
- return &empty.Empty{}, agent.updatePmConfigs(ctx, configs)
-}
-
// InitPmConfigs initialize the pm configs as defined by the adapter.
func (dMgr *Manager) InitPmConfigs(ctx context.Context, deviceID string, pmConfigs *voltha.PmConfigs) error {
if pmConfigs.Id == "" {
@@ -893,17 +524,6 @@
return status.Errorf(codes.NotFound, "%s", deviceID)
}
-// ListDevicePmConfigs returns pm configs of device
-func (dMgr *Manager) ListDevicePmConfigs(ctx context.Context, id *voltha.ID) (*voltha.PmConfigs, error) {
- ctx = utils.WithRPCMetadataContext(ctx, "ListDevicePmConfigs")
- log.EnrichSpan(ctx, log.Fields{"device-id": id.Id})
- agent := dMgr.getDeviceAgent(ctx, id.Id)
- if agent == nil {
- return nil, status.Errorf(codes.NotFound, "%s", id.Id)
- }
- return agent.listPmConfigs(ctx)
-}
-
func (dMgr *Manager) getSwitchCapability(ctx context.Context, deviceID string) (*ic.SwitchCapability, error) {
logger.Debugw(ctx, "get-switch-capability", log.Fields{"device-id": deviceID})
if agent := dMgr.getDeviceAgent(ctx, deviceID); agent != nil {
@@ -912,15 +532,6 @@
return nil, status.Errorf(codes.NotFound, "%s", deviceID)
}
-func (dMgr *Manager) GetPorts(ctx context.Context, deviceID string, portType voltha.Port_PortType) (*voltha.Ports, error) {
- logger.Debugw(ctx, "get-ports", log.Fields{"device-id": deviceID, "port-type": portType})
- agent := dMgr.getDeviceAgent(ctx, deviceID)
- if agent == nil {
- return nil, status.Errorf(codes.NotFound, "%s", deviceID)
- }
- return agent.getPorts(ctx, portType), nil
-}
-
func (dMgr *Manager) UpdateDeviceStatus(ctx context.Context, deviceID string, operStatus voltha.OperStatus_Types, connStatus voltha.ConnectStatus_Types) error {
logger.Debugw(ctx, "update-device-status", log.Fields{"device-id": deviceID, "oper-status": operStatus, "conn-status": connStatus})
if agent := dMgr.getDeviceAgent(ctx, deviceID); agent != nil {
@@ -974,31 +585,6 @@
return status.Errorf(codes.NotFound, "%s", deviceID)
}
-func (dMgr *Manager) DeleteAllPorts(ctx context.Context, deviceID string) error {
- logger.Debugw(ctx, "delete-all-ports", log.Fields{"device-id": deviceID})
- if agent := dMgr.getDeviceAgent(ctx, deviceID); agent != nil {
- if err := agent.deleteAllPorts(ctx); err != nil {
- return err
- }
- // Notify the logical device manager to remove all logical ports, if needed.
- // At this stage the device itself may gave been deleted already at a DeleteAllPorts
- // typically is part of a device deletion phase.
- if device, err := dMgr.getDeviceReadOnly(ctx, deviceID); err == nil {
- go func() {
- subCtx := utils.WithSpanAndRPCMetadataFromContext(ctx)
- if err := dMgr.logicalDeviceMgr.deleteAllLogicalPorts(subCtx, device); err != nil {
- logger.Errorw(ctx, "unable-to-delete-logical-ports", log.Fields{"error": err})
- }
- }()
- } else {
- logger.Warnw(ctx, "failed-to-retrieve-device", log.Fields{"device-id": deviceID})
- return err
- }
- return nil
- }
- return status.Errorf(codes.NotFound, "%s", deviceID)
-}
-
//UpdatePortsState updates all ports on the device
func (dMgr *Manager) UpdatePortsState(ctx context.Context, deviceID string, portTypeFilter uint32, state voltha.OperStatus_Types) error {
logger.Debugw(ctx, "update-ports-state", log.Fields{"device-id": deviceID})
@@ -1016,80 +602,6 @@
return nil
}
-func (dMgr *Manager) ChildDeviceDetected(ctx context.Context, parentDeviceID string, parentPortNo int64, deviceType string,
- channelID int64, vendorID string, serialNumber string, onuID int64) (*voltha.Device, error) {
- logger.Debugw(ctx, "child-device-detected", log.Fields{"parent-device-id": parentDeviceID, "parent-port-no": parentPortNo, "device-type": deviceType, "channel-id": channelID, "vendor-id": vendorID, "serial-number": serialNumber, "onu-id": onuID})
-
- if deviceType == "" && vendorID != "" {
- logger.Debug(ctx, "device-type-is-nil-fetching-device-type")
- deviceTypes, err := dMgr.adapterMgr.ListDeviceTypes(ctx, nil)
- if err != nil {
- return nil, err
- }
- OLoop:
- for _, dType := range deviceTypes.Items {
- for _, v := range dType.VendorIds {
- if v == vendorID {
- deviceType = dType.Adapter
- break OLoop
- }
- }
- }
- }
- //if no match found for the vendorid,report adapter with the custom error message
- if deviceType == "" {
- logger.Errorw(ctx, "failed-to-fetch-adapter-name ", log.Fields{"vendor-id": vendorID})
- return nil, status.Errorf(codes.NotFound, "%s", vendorID)
- }
-
- // Create the ONU device
- childDevice := &voltha.Device{}
- childDevice.Type = deviceType
- childDevice.ParentId = parentDeviceID
- childDevice.ParentPortNo = uint32(parentPortNo)
- childDevice.VendorId = vendorID
- childDevice.SerialNumber = serialNumber
- childDevice.Root = false
-
- // Get parent device type
- pAgent := dMgr.getDeviceAgent(ctx, parentDeviceID)
- if pAgent == nil {
- return nil, status.Errorf(codes.NotFound, "%s", parentDeviceID)
- }
- if pAgent.deviceType == "" {
- return nil, status.Errorf(codes.FailedPrecondition, "device Type not set %s", parentDeviceID)
- }
-
- if device, err := dMgr.GetChildDevice(ctx, parentDeviceID, serialNumber, onuID, parentPortNo); err == nil {
- logger.Warnw(ctx, "child-device-exists", log.Fields{"parent-device-id": parentDeviceID, "serial-number": serialNumber})
- return device, status.Errorf(codes.AlreadyExists, "%s", serialNumber)
- }
-
- childDevice.ProxyAddress = &voltha.Device_ProxyAddress{DeviceId: parentDeviceID, DeviceType: pAgent.deviceType, ChannelId: uint32(channelID), OnuId: uint32(onuID)}
-
- // Create and start a device agent for that device
- agent := newAgent(dMgr.adapterProxy, childDevice, dMgr, dMgr.dbPath, dMgr.dProxy, dMgr.defaultTimeout)
- insertedChildDevice, err := agent.start(ctx, false, childDevice)
- if err != nil {
- logger.Errorw(ctx, "error-starting-child-device", log.Fields{"parent-device-id": childDevice.ParentId, "child-device-id": agent.deviceID, "error": err})
- return nil, err
- }
- dMgr.addDeviceAgentToMap(agent)
-
- // Activate the child device
- if agent = dMgr.getDeviceAgent(ctx, agent.deviceID); agent != nil {
- go func() {
- subCtx := utils.WithFromTopicMetadataFromContext(utils.WithSpanAndRPCMetadataFromContext(ctx), ctx)
- err := agent.enableDevice(subCtx)
- if err != nil {
- logger.Errorw(ctx, "unable-to-enable-device", log.Fields{"error": err})
- }
- }()
- }
-
- return insertedChildDevice, nil
-}
-
func (dMgr *Manager) packetOut(ctx context.Context, deviceID string, outPort uint32, packet *ofp.OfpPacketOut) error {
logger.Debugw(ctx, "packet-out", log.Fields{"device-id": deviceID, "out-port": outPort})
if agent := dMgr.getDeviceAgent(ctx, deviceID); agent != nil {
@@ -1113,7 +625,7 @@
return status.Errorf(codes.FailedPrecondition, "%s", deviceID)
}
- if err := dMgr.logicalDeviceMgr.packetIn(ctx, device.ParentId, port, transactionID, packet); err != nil {
+ if err := dMgr.logicalDeviceMgr.packetIn(ctx, device.ParentId, port, packet); err != nil {
return err
}
return nil
@@ -1127,47 +639,6 @@
return status.Errorf(codes.NotFound, "%s", device.Id)
}
-//
-//CreateLogicalDevice creates logical device in core
-func (dMgr *Manager) CreateLogicalDevice(ctx context.Context, cDevice *voltha.Device) error {
- logger.Info(ctx, "create-logical-device")
- // Verify whether the logical device has already been created
- if cDevice.ParentId != "" {
- logger.Debugw(ctx, "parent-device-already-exist", log.Fields{"device-id": cDevice.Id, "logical-device-id": cDevice.Id})
- return nil
- }
- var err error
- if _, err = dMgr.logicalDeviceMgr.createLogicalDevice(ctx, cDevice); err != nil {
- logger.Warnw(ctx, "create-logical-device-error", log.Fields{"device": cDevice})
- return err
- }
- return nil
-}
-
-// DeleteLogicalDevice deletes logical device from core
-func (dMgr *Manager) DeleteLogicalDevice(ctx context.Context, cDevice *voltha.Device) error {
- logger.Info(ctx, "delete-logical-device")
- var err error
- if err = dMgr.logicalDeviceMgr.deleteLogicalDevice(ctx, cDevice); err != nil {
- logger.Warnw(ctx, "delete-logical-device-error", log.Fields{"device-id": cDevice.Id})
- return err
- }
- // Remove the logical device Id from the parent device
- logicalID := ""
- dMgr.UpdateDeviceAttribute(ctx, cDevice.Id, "ParentId", logicalID)
- return nil
-}
-
-// DeleteLogicalPorts removes the logical ports associated with that deviceId
-func (dMgr *Manager) DeleteLogicalPorts(ctx context.Context, cDevice *voltha.Device) error {
- logger.Debugw(ctx, "delete-all-logical-ports", log.Fields{"device-id": cDevice.Id})
- if err := dMgr.logicalDeviceMgr.deleteLogicalPorts(ctx, cDevice.Id); err != nil {
- // Just log the error. The logical device or port may already have been deleted before this callback is invoked.
- logger.Warnw(ctx, "delete-logical-ports-error", log.Fields{"device-id": cDevice.Id, "error": err})
- }
- return nil
-}
-
func (dMgr *Manager) getParentDevice(ctx context.Context, childDevice *voltha.Device) *voltha.Device {
// Sanity check
if childDevice.Root {
@@ -1178,54 +649,6 @@
return parentDevice
}
-//ChildDevicesLost is invoked by an adapter to indicate 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 (dMgr *Manager) ChildDevicesLost(ctx context.Context, parentDeviceID string) error {
- logger.Debug(ctx, "child-devices-lost")
- parentDevice, err := dMgr.getDeviceReadOnly(ctx, parentDeviceID)
- if err != nil {
- logger.Warnw(ctx, "failed-getting-device", log.Fields{"parent-device-id": parentDeviceID, "error": err})
- return err
- }
- return dMgr.DisableAllChildDevices(ctx, parentDevice)
-}
-
-//ChildDevicesDetected is 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 (dMgr *Manager) ChildDevicesDetected(ctx context.Context, parentDeviceID string) error {
- logger.Debug(ctx, "child-devices-detected")
- parentDevicePorts, err := dMgr.listDevicePorts(ctx, parentDeviceID)
- if err != nil {
- logger.Warnw(ctx, "failed-getting-device", log.Fields{"device-id": parentDeviceID, "error": err})
- return err
- }
- childDeviceIds := dMgr.getAllChildDeviceIds(ctx, parentDevicePorts)
- if len(childDeviceIds) == 0 {
- logger.Debugw(ctx, "no-child-device", log.Fields{"parent-device-id": parentDeviceID})
- }
- allChildEnableRequestSent := true
- for childDeviceID := range childDeviceIds {
- if agent := dMgr.getDeviceAgent(ctx, childDeviceID); agent != nil {
- subCtx := utils.WithSpanAndRPCMetadataFromContext(ctx)
- // Run the children re-registration in its own routine
- go func(ctx context.Context) {
- err = agent.enableDevice(ctx)
- if err != nil {
- logger.Errorw(ctx, "unable-to-enable-device", log.Fields{"error": err})
- }
- }(subCtx)
- } else {
- err = status.Errorf(codes.Unavailable, "no agent for child device %s", childDeviceID)
- logger.Errorw(ctx, "no-child-device-agent", log.Fields{"parent-device-id": parentDeviceID, "child-id": childDeviceID})
- allChildEnableRequestSent = false
- }
- }
- if !allChildEnableRequestSent {
- return err
- }
- return nil
-}
-
/*
All the functions below are callback functions where they are invoked with the latest and previous data. We can
therefore use the data as is without trying to get the latest from the model.
@@ -1246,63 +669,6 @@
return nil
}
-//DeleteAllChildDevices is invoked as a callback when the parent device is deleted
-func (dMgr *Manager) DeleteAllChildDevices(ctx context.Context, parentCurrDevice *voltha.Device) error {
- logger.Debug(ctx, "delete-all-child-devices")
- force := false
- // Get the parent device Transient state, if its FORCE_DELETED(go for force delete for child devices)
- // So in cases when this handler is getting called other than DELETE operation, no force option would be used.
- agent := dMgr.getDeviceAgent(ctx, parentCurrDevice.Id)
- if agent == nil {
- return status.Errorf(codes.NotFound, "%s", parentCurrDevice.Id)
- }
-
- force = agent.getTransientState() == voltha.DeviceTransientState_FORCE_DELETING
-
- ports, _ := dMgr.listDevicePorts(ctx, parentCurrDevice.Id)
- for childDeviceID := range dMgr.getAllChildDeviceIds(ctx, ports) {
- if agent := dMgr.getDeviceAgent(ctx, childDeviceID); agent != nil {
- if force {
- if err := agent.deleteDeviceForce(ctx); err != nil {
- logger.Warnw(ctx, "failure-delete-device-force", log.Fields{"device-id": childDeviceID,
- "error": err.Error()})
- }
- } else {
- if err := agent.deleteDevice(ctx); err != nil {
- logger.Warnw(ctx, "failure-delete-device", log.Fields{"device-id": childDeviceID,
- "error": err.Error()})
- }
- }
- // No further action is required here. The deleteDevice will change the device state where the resulting
- // callback will take care of cleaning the child device agent.
- }
- }
- return nil
-}
-
-//DeleteAllLogicalPorts is invoked as a callback when the parent device's connection status moves to UNREACHABLE
-func (dMgr *Manager) DeleteAllLogicalPorts(ctx context.Context, parentDevice *voltha.Device) error {
- logger.Debugw(ctx, "delete-all-logical-ports", log.Fields{"parent-device-id": parentDevice.Id})
- if err := dMgr.logicalDeviceMgr.deleteAllLogicalPorts(ctx, parentDevice); err != nil {
- // Just log error as logical device may already have been deleted
- logger.Warnw(ctx, "delete-all-logical-ports-fail", log.Fields{"parent-device-id": parentDevice.Id, "error": err})
- }
- return nil
-}
-
-//DeleteAllDeviceFlows is invoked as a callback when the parent device's connection status moves to UNREACHABLE
-func (dMgr *Manager) DeleteAllDeviceFlows(ctx context.Context, parentDevice *voltha.Device) error {
- logger.Debugw(ctx, "delete-all-device-flows", log.Fields{"parent-device-id": parentDevice.Id})
- if agent := dMgr.getDeviceAgent(ctx, parentDevice.Id); agent != nil {
- if err := agent.deleteAllFlows(ctx); err != nil {
- logger.Errorw(ctx, "error-deleting-all-device-flows", log.Fields{"parent-device-id": parentDevice.Id})
- return err
- }
- return nil
- }
- return status.Errorf(codes.NotFound, "%s", parentDevice.Id)
-}
-
//getAllChildDeviceIds is a helper method to get all the child device IDs from the device passed as parameter
func (dMgr *Manager) getAllChildDeviceIds(ctx context.Context, parentDevicePorts map[uint32]*voltha.Port) map[string]struct{} {
logger.Debug(ctx, "get-all-child-device-ids")
@@ -1316,8 +682,8 @@
return childDeviceIds
}
-//GetAllChildDevices is a helper method to get all the child device IDs from the device passed as parameter
-func (dMgr *Manager) GetAllChildDevices(ctx context.Context, parentDeviceID string) (*voltha.Devices, error) {
+//GgtAllChildDevices is a helper method to get all the child device IDs from the device passed as parameter
+func (dMgr *Manager) getAllChildDevices(ctx context.Context, parentDeviceID string) (*voltha.Devices, error) {
logger.Debugw(ctx, "get-all-child-devices", log.Fields{"parent-device-id": parentDeviceID})
if parentDevicePorts, err := dMgr.listDevicePorts(ctx, parentDeviceID); err == nil {
childDevices := make([]*voltha.Device, 0)
@@ -1331,165 +697,6 @@
return nil, status.Errorf(codes.NotFound, "%s", parentDeviceID)
}
-// SetupUNILogicalPorts creates UNI ports on the logical device that represents a child UNI interface
-func (dMgr *Manager) SetupUNILogicalPorts(ctx context.Context, cDevice *voltha.Device) error {
- logger.Info(ctx, "setup-uni-logical-ports")
- cDevicePorts, err := dMgr.listDevicePorts(ctx, cDevice.Id)
- if err != nil {
- return err
- }
- if err := dMgr.logicalDeviceMgr.setupUNILogicalPorts(ctx, cDevice, cDevicePorts); err != nil {
- logger.Warnw(ctx, "setup-uni-logical-ports-error", log.Fields{"device": cDevice, "err": err})
- return err
- }
- return nil
-}
-
-// convenience to avoid redefining
-var operationFailureResp = &common.OperationResp{Code: voltha.OperationResp_OPERATION_FAILURE}
-
-// DownloadImage execute an image download request
-func (dMgr *Manager) DownloadImage(ctx context.Context, img *voltha.ImageDownload) (*common.OperationResp, error) {
- ctx = utils.WithRPCMetadataContext(ctx, "DownloadImage")
- log.EnrichSpan(ctx, log.Fields{"device-id": img.Id})
- logger.Debugw(ctx, "download-image", log.Fields{"device-id": img.Id, "image-name": img.Name})
- agent := dMgr.getDeviceAgent(ctx, img.Id)
- if agent == nil {
- return operationFailureResp, status.Errorf(codes.NotFound, "%s", img.Id)
- }
- resp, err := agent.downloadImage(ctx, img)
- if err != nil {
- return operationFailureResp, err
- }
- return resp, nil
-}
-
-// CancelImageDownload cancels image download request
-func (dMgr *Manager) CancelImageDownload(ctx context.Context, img *voltha.ImageDownload) (*common.OperationResp, error) {
- ctx = utils.WithRPCMetadataContext(ctx, "CancelImageDownload")
- log.EnrichSpan(ctx, log.Fields{"device-id": img.Id})
- logger.Debugw(ctx, "cancel-image-download", log.Fields{"device-id": img.Id, "image-name": img.Name})
- agent := dMgr.getDeviceAgent(ctx, img.Id)
- if agent == nil {
- return operationFailureResp, status.Errorf(codes.NotFound, "%s", img.Id)
- }
- resp, err := agent.cancelImageDownload(ctx, img)
- if err != nil {
- return operationFailureResp, err
- }
- return resp, nil
-}
-
-// ActivateImageUpdate activates image update request
-func (dMgr *Manager) ActivateImageUpdate(ctx context.Context, img *voltha.ImageDownload) (*common.OperationResp, error) {
- ctx = utils.WithRPCMetadataContext(ctx, "ActivateImageUpdate")
- log.EnrichSpan(ctx, log.Fields{"device-id": img.Id})
- logger.Debugw(ctx, "activate-image-update", log.Fields{"device-id": img.Id, "image-name": img.Name})
- agent := dMgr.getDeviceAgent(ctx, img.Id)
- if agent == nil {
- return operationFailureResp, status.Errorf(codes.NotFound, "%s", img.Id)
- }
- resp, err := agent.activateImage(ctx, img)
- if err != nil {
- return operationFailureResp, err
- }
- return resp, nil
-}
-
-// RevertImageUpdate reverts image update
-func (dMgr *Manager) RevertImageUpdate(ctx context.Context, img *voltha.ImageDownload) (*common.OperationResp, error) {
- ctx = utils.WithRPCMetadataContext(ctx, "RevertImageUpdate")
- log.EnrichSpan(ctx, log.Fields{"device-id": img.Id})
- logger.Debugw(ctx, "rever-image-update", log.Fields{"device-id": img.Id, "image-name": img.Name})
- agent := dMgr.getDeviceAgent(ctx, img.Id)
- if agent == nil {
- return operationFailureResp, status.Errorf(codes.NotFound, "%s", img.Id)
- }
- resp, err := agent.revertImage(ctx, img)
- if err != nil {
- return operationFailureResp, err
- }
- return resp, nil
-}
-
-// convenience to avoid redefining
-var imageDownloadFailureResp = &voltha.ImageDownload{DownloadState: voltha.ImageDownload_DOWNLOAD_UNKNOWN}
-
-// GetImageDownloadStatus returns status of image download
-func (dMgr *Manager) GetImageDownloadStatus(ctx context.Context, img *voltha.ImageDownload) (*voltha.ImageDownload, error) {
- ctx = utils.WithRPCMetadataContext(ctx, "GetImageDownloadStatus")
- log.EnrichSpan(ctx, log.Fields{"device-id": img.Id})
- logger.Debugw(ctx, "get-image-download-status", log.Fields{"device-id": img.Id, "image-name": img.Name})
- agent := dMgr.getDeviceAgent(ctx, img.Id)
- if agent == nil {
- return imageDownloadFailureResp, status.Errorf(codes.NotFound, "%s", img.Id)
- }
- resp, err := agent.getImageDownloadStatus(ctx, img)
- if err != nil {
- return imageDownloadFailureResp, err
- }
- return resp, nil
-}
-
-func (dMgr *Manager) UpdateImageDownload(ctx context.Context, deviceID string, img *voltha.ImageDownload) error {
- ctx = utils.WithRPCMetadataContext(ctx, "UpdateImageDownload")
- log.EnrichSpan(ctx, log.Fields{"device-id": img.Id})
- logger.Debugw(ctx, "update-image-download", log.Fields{"device-id": img.Id, "image-name": img.Name})
- if agent := dMgr.getDeviceAgent(ctx, deviceID); agent != nil {
- if err := agent.updateImageDownload(ctx, img); err != nil {
- logger.Debugw(ctx, "update-image-download-failed", log.Fields{"err": err, "image-name": img.Name})
- return err
- }
- } else {
- return status.Errorf(codes.NotFound, "%s", img.Id)
- }
- return nil
-}
-
-// GetImageDownload returns image download
-func (dMgr *Manager) GetImageDownload(ctx context.Context, img *voltha.ImageDownload) (*voltha.ImageDownload, error) {
- ctx = utils.WithRPCMetadataContext(ctx, "GetImageDownload")
- log.EnrichSpan(ctx, log.Fields{"device-id": img.Id})
- logger.Debugw(ctx, "get-image-download", log.Fields{"device-id": img.Id, "image-name": img.Name})
- agent := dMgr.getDeviceAgent(ctx, img.Id)
- if agent == nil {
- return imageDownloadFailureResp, status.Errorf(codes.NotFound, "%s", img.Id)
- }
- resp, err := agent.getImageDownload(ctx, img)
- if err != nil {
- return imageDownloadFailureResp, err
- }
- return resp, nil
-}
-
-// ListImageDownloads returns image downloads
-func (dMgr *Manager) ListImageDownloads(ctx context.Context, id *voltha.ID) (*voltha.ImageDownloads, error) {
- ctx = utils.WithRPCMetadataContext(ctx, "ListImageDownloads")
- log.EnrichSpan(ctx, log.Fields{"device-id": id.Id})
- logger.Debugw(ctx, "list-image-downloads", log.Fields{"device-id": id.Id})
- agent := dMgr.getDeviceAgent(ctx, id.Id)
- if agent == nil {
- return &voltha.ImageDownloads{Items: []*voltha.ImageDownload{imageDownloadFailureResp}}, status.Errorf(codes.NotFound, "%s", id.Id)
- }
- resp, err := agent.listImageDownloads(ctx, id.Id)
- if err != nil {
- return &voltha.ImageDownloads{Items: []*voltha.ImageDownload{imageDownloadFailureResp}}, err
- }
- return resp, nil
-}
-
-// GetImages returns all images for a specific device entry
-func (dMgr *Manager) GetImages(ctx context.Context, id *voltha.ID) (*voltha.Images, error) {
- ctx = utils.WithRPCMetadataContext(ctx, "GetImages")
- log.EnrichSpan(ctx, log.Fields{"device-id": id.Id})
- logger.Debugw(ctx, "get-images", log.Fields{"device-id": id.Id})
- device, err := dMgr.getDeviceReadOnly(ctx, id.Id)
- if err != nil {
- return nil, err
- }
- return device.Images, nil
-}
-
func (dMgr *Manager) NotifyInvalidTransition(ctx context.Context, device *voltha.Device) error {
logger.Errorw(ctx, "notify-invalid-transition", log.Fields{
"device": device.Id,
@@ -1517,21 +724,6 @@
return ""
}
-func (dMgr *Manager) SimulateAlarm(ctx context.Context, simulateReq *voltha.SimulateAlarmRequest) (*common.OperationResp, error) {
- ctx = utils.WithRPCMetadataContext(ctx, "SimulateAlarm")
- logger.Debugw(ctx, "simulate-alarm", log.Fields{"id": simulateReq.Id, "indicator": simulateReq.Indicator, "intf-id": simulateReq.IntfId,
- "port-type-name": simulateReq.PortTypeName, "onu-device-id": simulateReq.OnuDeviceId, "inverse-bit-error-rate": simulateReq.InverseBitErrorRate,
- "drift": simulateReq.Drift, "new-eqd": simulateReq.NewEqd, "onu-serial-number": simulateReq.OnuSerialNumber, "operation": simulateReq.Operation})
- agent := dMgr.getDeviceAgent(ctx, simulateReq.Id)
- if agent == nil {
- return nil, status.Errorf(codes.NotFound, "%s", simulateReq.Id)
- }
- if err := agent.simulateAlarm(ctx, simulateReq); err != nil {
- return nil, err
- }
- return &common.OperationResp{Code: common.OperationResp_OPERATION_SUCCESS}, nil
-}
-
func (dMgr *Manager) UpdateDeviceReason(ctx context.Context, deviceID string, reason string) error {
logger.Debugw(ctx, "update-device-reason", log.Fields{"device-id": deviceID, "reason": reason})
if agent := dMgr.getDeviceAgent(ctx, deviceID); agent != nil {
@@ -1540,368 +732,20 @@
return status.Errorf(codes.NotFound, "%s", deviceID)
}
-func (dMgr *Manager) EnablePort(ctx context.Context, port *voltha.Port) (*empty.Empty, error) {
- ctx = utils.WithRPCMetadataContext(ctx, "EnablePort")
- log.EnrichSpan(ctx, log.Fields{"device-id": port.DeviceId})
- logger.Debugw(ctx, "enable-port", log.Fields{"device-id": port.DeviceId, "port-no": port.PortNo})
- agent := dMgr.getDeviceAgent(ctx, port.DeviceId)
- if agent == nil {
- return nil, status.Errorf(codes.NotFound, "%s", port.DeviceId)
- }
- return &empty.Empty{}, agent.enablePort(ctx, port.PortNo)
-}
-
-func (dMgr *Manager) DisablePort(ctx context.Context, port *voltha.Port) (*empty.Empty, error) {
- ctx = utils.WithRPCMetadataContext(ctx, "DisablePort")
- log.EnrichSpan(ctx, log.Fields{"device-id": port.DeviceId})
- logger.Debugw(ctx, "disable-port", log.Fields{"device-id": port.DeviceId, "port-no": port.PortNo})
- agent := dMgr.getDeviceAgent(ctx, port.DeviceId)
- if agent == nil {
- return nil, status.Errorf(codes.NotFound, "%s", port.DeviceId)
- }
- return &empty.Empty{}, agent.disablePort(ctx, port.PortNo)
-}
-
-// ChildDeviceLost calls parent adapter to delete child device and all its references
-func (dMgr *Manager) ChildDeviceLost(ctx context.Context, curr *voltha.Device) error {
- logger.Debugw(ctx, "child-device-lost", log.Fields{"child-device-id": curr.Id, "parent-device-id": curr.ParentId})
- if parentAgent := dMgr.getDeviceAgent(ctx, curr.ParentId); parentAgent != nil {
- if err := parentAgent.ChildDeviceLost(ctx, curr); err != nil {
- // Just log the message and let the remaining pipeline proceed.
- logger.Warnw(ctx, "childDeviceLost", log.Fields{"child-device-id": curr.Id, "parent-device-id": curr.ParentId, "error": err})
- }
- }
- // Do not return an error as parent device may also have been deleted. Let the remaining pipeline proceed.
- return nil
-}
-
-func (dMgr *Manager) StartOmciTestAction(ctx context.Context, request *voltha.OmciTestRequest) (*voltha.TestResponse, error) {
- ctx = utils.WithRPCMetadataContext(ctx, "StartOmciTestAction")
- log.EnrichSpan(ctx, log.Fields{"device-id": request.Id})
- logger.Debugw(ctx, "start-omci-test-action", log.Fields{"device-id": request.Id, "uuid": request.Uuid})
- agent := dMgr.getDeviceAgent(ctx, request.Id)
- if agent == nil {
- return nil, status.Errorf(codes.NotFound, "%s", request.Id)
- }
- return agent.startOmciTest(ctx, request)
-}
-
-func (dMgr *Manager) GetExtValue(ctx context.Context, value *voltha.ValueSpecifier) (*voltha.ReturnValues, error) {
- ctx = utils.WithRPCMetadataContext(ctx, "GetExtValue")
- log.EnrichSpan(ctx, log.Fields{"device-id": value.Id})
- logger.Debugw(ctx, "get-ext-value", log.Fields{"onu-id": value.Id})
- cDevice, err := dMgr.getDeviceReadOnly(ctx, value.Id)
- if err != nil {
- return nil, status.Errorf(codes.Aborted, "%s", err.Error())
- }
- pDevice, err := dMgr.getDeviceReadOnly(ctx, cDevice.ParentId)
- if err != nil {
- return nil, status.Errorf(codes.Aborted, "%s", err.Error())
- }
- if agent := dMgr.getDeviceAgent(ctx, cDevice.ParentId); agent != nil {
- resp, err := agent.getExtValue(ctx, pDevice, cDevice, value)
- if err != nil {
- return nil, err
- }
- logger.Debugw(ctx, "get-ext-value-result", log.Fields{"result": resp})
- return resp, nil
- }
- return nil, status.Errorf(codes.NotFound, "%s", value.Id)
-
-}
-
-// SetExtValue set some given configs or value
-func (dMgr *Manager) SetExtValue(ctx context.Context, value *voltha.ValueSet) (*empty.Empty, error) {
- ctx = utils.WithRPCMetadataContext(ctx, "SetExtValue")
- logger.Debugw(ctx, "set-ext-value", log.Fields{"onu-id": value.Id})
- device, err := dMgr.getDeviceReadOnly(ctx, value.Id)
- if err != nil {
- return nil, status.Errorf(codes.Aborted, "%s", err.Error())
- }
- if agent := dMgr.getDeviceAgent(ctx, device.Id); agent != nil {
- resp, err := agent.setExtValue(ctx, device, value)
- if err != nil {
- return nil, err
- }
- logger.Debugw(ctx, "set-ext-value-result", log.Fields{"result": resp})
- return resp, nil
- }
- return nil, status.Errorf(codes.NotFound, "%s", value.Id)
-
-}
-
func (dMgr *Manager) SendRPCEvent(ctx context.Context, id string, rpcEvent *voltha.RPCEvent,
category voltha.EventCategory_Types, subCategory *voltha.EventSubCategory_Types, raisedTs int64) {
//TODO Instead of directly sending to the kafka bus, queue the message and send it asynchronously
dMgr.Agent.SendRPCEvent(ctx, id, rpcEvent, category, subCategory, raisedTs)
}
-func (dMgr *Manager) GetTransientState(ctx context.Context, id string) (voltha.DeviceTransientState_Types, error) {
+func (dMgr *Manager) GetTransientState(ctx context.Context, id string) (core.DeviceTransientState_Types, error) {
agent := dMgr.getDeviceAgent(ctx, id)
if agent == nil {
- return voltha.DeviceTransientState_NONE, status.Errorf(codes.NotFound, "%s", id)
+ return core.DeviceTransientState_NONE, status.Errorf(codes.NotFound, "%s", id)
}
return agent.getTransientState(), nil
}
-func (dMgr *Manager) DownloadImageToDevice(ctx context.Context, request *voltha.DeviceImageDownloadRequest) (*voltha.DeviceImageResponse, error) {
- if err := dMgr.validateImageDownloadRequest(request); err != nil {
- return nil, err
- }
-
- ctx = utils.WithRPCMetadataContext(ctx, "DownloadImageToDevice")
- respCh := make(chan []*voltha.DeviceImageState, len(request.GetDeviceId()))
-
- for index, deviceID := range request.DeviceId {
- // Create download request per device
- downloadReq := &voltha.DeviceImageDownloadRequest{
- Image: request.Image,
- ActivateOnSuccess: request.ActivateOnSuccess,
- CommitOnSuccess: request.CommitOnSuccess,
- }
-
- //slice-out only single deviceID from the request
- downloadReq.DeviceId = request.DeviceId[index : index+1]
-
- go func(deviceID string, req *voltha.DeviceImageDownloadRequest, ch chan []*voltha.DeviceImageState) {
- agent := dMgr.getDeviceAgent(ctx, deviceID)
- if agent == nil {
- logger.Errorw(ctx, "Not-found", log.Fields{"device-id": deviceID})
- ch <- nil
- return
- }
-
- resp, err := agent.downloadImageToDevice(ctx, req)
- if err != nil {
- logger.Errorw(ctx, "download-image-to-device-failed", log.Fields{"device-id": deviceID, "error": err})
- ch <- nil
- return
- }
-
- err = dMgr.validateDeviceImageResponse(resp)
- if err != nil {
- logger.Errorw(ctx, "download-image-to-device-failed", log.Fields{"device-id": deviceID, "error": err})
- ch <- nil
- return
- }
- ch <- resp.GetDeviceImageStates()
- }(deviceID.GetId(), downloadReq, respCh)
-
- }
-
- return dMgr.waitForAllResponses(ctx, "download-image-to-device", respCh, len(request.GetDeviceId()))
-}
-
-func (dMgr *Manager) GetImageStatus(ctx context.Context, request *voltha.DeviceImageRequest) (*voltha.DeviceImageResponse, error) {
- if err := dMgr.validateImageRequest(request); err != nil {
- return nil, err
- }
-
- ctx = utils.WithRPCMetadataContext(ctx, "GetImageStatus")
-
- respCh := make(chan []*voltha.DeviceImageState, len(request.GetDeviceId()))
- for index, deviceID := range request.DeviceId {
- // Create status request per device
- imageStatusReq := &voltha.DeviceImageRequest{
- Version: request.Version,
- CommitOnSuccess: request.CommitOnSuccess,
- }
-
- //slice-out only single deviceID from the request
- imageStatusReq.DeviceId = request.DeviceId[index : index+1]
-
- go func(deviceID string, req *voltha.DeviceImageRequest, ch chan []*voltha.DeviceImageState) {
- agent := dMgr.getDeviceAgent(ctx, deviceID)
- if agent == nil {
- logger.Errorw(ctx, "Not-found", log.Fields{"device-id": deviceID})
- ch <- nil
- return
- }
-
- resp, err := agent.getImageStatus(ctx, req)
- if err != nil {
- logger.Errorw(ctx, "get-image-status-failed", log.Fields{"device-id": deviceID, "error": err})
- ch <- nil
- return
- }
-
- err = dMgr.validateDeviceImageResponse(resp)
- if err != nil {
- logger.Errorw(ctx, "get-image-status-failed", log.Fields{"device-id": deviceID, "error": err})
- ch <- nil
- return
- }
- ch <- resp.GetDeviceImageStates()
- }(deviceID.GetId(), imageStatusReq, respCh)
-
- }
-
- return dMgr.waitForAllResponses(ctx, "get-image-status", respCh, len(request.GetDeviceId()))
-}
-
-func (dMgr *Manager) AbortImageUpgradeToDevice(ctx context.Context, request *voltha.DeviceImageRequest) (*voltha.DeviceImageResponse, error) {
- if err := dMgr.validateImageRequest(request); err != nil {
- return nil, err
- }
-
- ctx = utils.WithRPCMetadataContext(ctx, "AbortImageUpgradeToDevice")
- respCh := make(chan []*voltha.DeviceImageState, len(request.GetDeviceId()))
-
- for index, deviceID := range request.DeviceId {
- // Create abort request per device
- abortImageReq := &voltha.DeviceImageRequest{
- Version: request.Version,
- CommitOnSuccess: request.CommitOnSuccess,
- }
-
- //slice-out only single deviceID from the request
- abortImageReq.DeviceId = request.DeviceId[index : index+1]
-
- go func(deviceID string, req *voltha.DeviceImageRequest, ch chan []*voltha.DeviceImageState) {
- agent := dMgr.getDeviceAgent(ctx, deviceID)
- if agent == nil {
- logger.Errorw(ctx, "Not-found", log.Fields{"device-id": deviceID})
- ch <- nil
- return
- }
-
- resp, err := agent.abortImageUpgradeToDevice(ctx, req)
- if err != nil {
- logger.Errorw(ctx, "abort-image-upgrade-to-device-failed", log.Fields{"device-id": deviceID, "error": err})
- ch <- nil
- return
- }
-
- err = dMgr.validateDeviceImageResponse(resp)
- if err != nil {
- logger.Errorw(ctx, "abort-image-upgrade-to-device-failed", log.Fields{"device-id": deviceID, "error": err})
- ch <- nil
- return
- }
- ch <- resp.GetDeviceImageStates()
- }(deviceID.GetId(), abortImageReq, respCh)
-
- }
-
- return dMgr.waitForAllResponses(ctx, "abort-image-upgrade-to-device", respCh, len(request.GetDeviceId()))
-}
-
-func (dMgr *Manager) GetOnuImages(ctx context.Context, id *common.ID) (*voltha.OnuImages, error) {
- if id == nil || id.Id == "" {
- return nil, status.Errorf(codes.InvalidArgument, "empty device id")
- }
-
- ctx = utils.WithRPCMetadataContext(ctx, "GetOnuImages")
- log.EnrichSpan(ctx, log.Fields{"device-id": id.Id})
- agent := dMgr.getDeviceAgent(ctx, id.Id)
- if agent == nil {
- return nil, status.Errorf(codes.NotFound, "%s", id.Id)
- }
-
- resp, err := agent.getOnuImages(ctx, id)
- if err != nil {
- return nil, err
- }
-
- logger.Debugw(ctx, "get-onu-images-result", log.Fields{"onu-image": resp.Items})
-
- return resp, nil
-}
-
-func (dMgr *Manager) ActivateImage(ctx context.Context, request *voltha.DeviceImageRequest) (*voltha.DeviceImageResponse, error) {
- if err := dMgr.validateImageRequest(request); err != nil {
- return nil, err
- }
-
- ctx = utils.WithRPCMetadataContext(ctx, "ActivateImage")
- respCh := make(chan []*voltha.DeviceImageState, len(request.GetDeviceId()))
-
- for index, deviceID := range request.DeviceId {
- // Create activate request per device
- activateImageReq := &voltha.DeviceImageRequest{
- Version: request.Version,
- CommitOnSuccess: request.CommitOnSuccess,
- }
-
- //slice-out only single deviceID from the request
- activateImageReq.DeviceId = request.DeviceId[index : index+1]
-
- go func(deviceID string, req *voltha.DeviceImageRequest, ch chan []*voltha.DeviceImageState) {
- agent := dMgr.getDeviceAgent(ctx, deviceID)
- if agent == nil {
- logger.Errorw(ctx, "Not-found", log.Fields{"device-id": deviceID})
- ch <- nil
- return
- }
-
- resp, err := agent.activateImageOnDevice(ctx, req)
- if err != nil {
- logger.Errorw(ctx, "activate-image-failed", log.Fields{"device-id": deviceID, "error": err})
- ch <- nil
- return
- }
-
- err = dMgr.validateDeviceImageResponse(resp)
- if err != nil {
- logger.Errorw(ctx, "activate-image-failed", log.Fields{"device-id": deviceID, "error": err})
- ch <- nil
- return
- }
-
- ch <- resp.GetDeviceImageStates()
- }(deviceID.GetId(), activateImageReq, respCh)
-
- }
-
- return dMgr.waitForAllResponses(ctx, "activate-image", respCh, len(request.GetDeviceId()))
-}
-
-func (dMgr *Manager) CommitImage(ctx context.Context, request *voltha.DeviceImageRequest) (*voltha.DeviceImageResponse, error) {
- if err := dMgr.validateImageRequest(request); err != nil {
- return nil, err
- }
-
- ctx = utils.WithRPCMetadataContext(ctx, "CommitImage")
- respCh := make(chan []*voltha.DeviceImageState, len(request.GetDeviceId()))
-
- for index, deviceID := range request.DeviceId {
- // Create commit request per device
- commitImageReq := &voltha.DeviceImageRequest{
- Version: request.Version,
- CommitOnSuccess: request.CommitOnSuccess,
- }
- //slice-out only single deviceID from the request
- commitImageReq.DeviceId = request.DeviceId[index : index+1]
-
- go func(deviceID string, req *voltha.DeviceImageRequest, ch chan []*voltha.DeviceImageState) {
- agent := dMgr.getDeviceAgent(ctx, deviceID)
- if agent == nil {
- logger.Errorw(ctx, "Not-found", log.Fields{"device-id": deviceID})
- ch <- nil
- return
- }
-
- resp, err := agent.commitImage(ctx, req)
- if err != nil {
- logger.Errorw(ctx, "commit-image-failed", log.Fields{"device-id": deviceID, "error": err})
- ch <- nil
- return
- }
-
- err = dMgr.validateDeviceImageResponse(resp)
- if err != nil {
- logger.Errorf(ctx, "commit-image-failed", log.Fields{"device-id": deviceID, "error": err})
- ch <- nil
- return
- }
- ch <- resp.GetDeviceImageStates()
- }(deviceID.GetId(), commitImageReq, respCh)
-
- }
-
- return dMgr.waitForAllResponses(ctx, "commit-image", respCh, len(request.GetDeviceId()))
-}
-
func (dMgr *Manager) validateImageDownloadRequest(request *voltha.DeviceImageDownloadRequest) error {
if request == nil || request.Image == nil || len(request.DeviceId) == 0 {
return status.Errorf(codes.InvalidArgument, "invalid argument")
@@ -1979,3 +823,12 @@
}
return nil
}
+
+func (dMgr *Manager) adapterRestartedHandler(ctx context.Context, endpoint string) error {
+ // Get the adapter corresponding to that endpoint
+ if a, _ := dMgr.adapterMgr.GetAdapterWithEndpoint(ctx, endpoint); a != nil {
+ return dMgr.adapterRestarted(ctx, a)
+ }
+ logger.Errorw(ctx, "restarted-adapter-not-found", log.Fields{"endpoint": endpoint})
+ return fmt.Errorf("restarted adapter at endpoint %s not found", endpoint)
+}