rw_core/core/device/manager_sbi.go

/*
* Copyright 2021-2024 Open Networking Foundation (ONF) and the ONF Contributors

* 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 device

import (
	"context"
	"fmt"
	"time"

	"github.com/golang/protobuf/ptypes/empty"
	"github.com/opencord/voltha-go/rw_core/utils"
	"github.com/opencord/voltha-lib-go/v7/pkg/log"
	"github.com/opencord/voltha-protos/v5/go/common"
	ca "github.com/opencord/voltha-protos/v5/go/core_adapter"
	"github.com/opencord/voltha-protos/v5/go/core_service"
	"github.com/opencord/voltha-protos/v5/go/health"
	"github.com/opencord/voltha-protos/v5/go/voltha"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

func (dMgr *Manager) PortCreated(ctx context.Context, port *voltha.Port) (*empty.Empty, error) {
	ctx = utils.WithNewSpanAndRPCMetadataContext(ctx, "PortCreated")

	logger.Debugw(ctx, "port-created", log.Fields{"port": port})

	agent := dMgr.getDeviceAgent(ctx, port.DeviceId)
	if agent != nil {
		if err := agent.addPort(ctx, port); err != nil {
			return nil, err
		}
		//	Setup peer ports in its own routine

		if err := dMgr.addPeerPort(ctx, port.DeviceId, port); err != nil {
			logger.Errorw(ctx, "unable-to-add-peer-port", log.Fields{"error": err, "device-id": port.DeviceId})
		}

		return &empty.Empty{}, nil
	}
	return nil, status.Errorf(codes.NotFound, "%s", port.DeviceId)
}

func (dMgr *Manager) DeviceUpdate(ctx context.Context, device *voltha.Device) (*empty.Empty, error) {
	ctx = utils.WithNewSpanAndRPCMetadataContext(ctx, "DeviceUpdate")
	logger.Debugw(ctx, "device-update", log.Fields{"device-id": device.Id, "device": device})

	if agent := dMgr.getDeviceAgent(ctx, device.Id); agent != nil {
		if err := agent.updateDeviceUsingAdapterData(ctx, device); err != nil {
			return nil, err
		}
		return &empty.Empty{}, nil
	}
	return nil, status.Errorf(codes.NotFound, "%s", device.Id)
}

func (dMgr *Manager) DeviceStateUpdate(ctx context.Context, ds *ca.DeviceStateFilter) (*empty.Empty, error) {
	ctx = utils.WithNewSpanAndRPCMetadataContext(ctx, "DeviceStateUpdate")
	logger.Debugw(ctx, "device-state-update", log.Fields{"device-id": ds.DeviceId, "operStatus": ds.OperStatus, "connStatus": ds.ConnStatus})

	if agent := dMgr.getDeviceAgent(ctx, ds.DeviceId); agent != nil {
		if err := agent.updateDeviceStatus(ctx, ds.OperStatus, ds.ConnStatus); err != nil {
			return nil, err
		}
		return &empty.Empty{}, nil
	}
	return nil, status.Errorf(codes.NotFound, "%s", ds.DeviceId)
}

func (dMgr *Manager) ChildDeviceDetected(ctx context.Context, dd *ca.DeviceDiscovery) (*voltha.Device, error) {
	ctx = utils.WithNewSpanAndRPCMetadataContext(ctx, "ChildDeviceDetected")
	logger.Debugw(ctx, "child-device-detected",
		log.Fields{
			"parent-device-id": dd.ParentId,
			"parentPortNo":     dd.ParentPortNo,
			"deviceType":       dd.ChildDeviceType,
			"channelId":        dd.ChannelId,
			"vendorId":         dd.VendorId,
			"serialNumber":     dd.SerialNumber,
			"onuId":            dd.OnuId,
		})

	var err error
	if dd.ChildDeviceType == "" && dd.VendorId != "" {
		logger.Debug(ctx, "device-type-is-nil-fetching-device-type")
		if dd.ChildDeviceType, err = dMgr.adapterMgr.GetAdapterTypeByVendorID(dd.VendorId); err != nil {
			return nil, err
		}
	}
	//if no match found for the vendorid,report adapter with the custom error message
	if dd.ChildDeviceType == "" {
		logger.Errorw(ctx, "failed-to-fetch-adapter-name ", log.Fields{"vendorId": dd.VendorId})
		return nil, status.Errorf(codes.NotFound, "%s", dd.VendorId)
	}

	// Create the ONU device
	childDevice := &voltha.Device{}
	childDevice.Type = dd.ChildDeviceType
	childDevice.ParentId = dd.ParentId
	childDevice.ParentPortNo = uint32(dd.ParentPortNo)
	childDevice.VendorId = dd.VendorId
	childDevice.SerialNumber = dd.SerialNumber
	childDevice.Root = false

	// Get parent device type
	pAgent := dMgr.getDeviceAgent(ctx, dd.ParentId)
	if pAgent == nil {
		return nil, status.Errorf(codes.NotFound, "%s", dd.ParentId)
	}
	if pAgent.deviceType == "" {
		pDevice, err := pAgent.getDeviceReadOnly(ctx)
		logger.Errorw(ctx, "device-type-not-set", log.Fields{"parent-device": pDevice, "error": err})
		return nil, status.Errorf(codes.FailedPrecondition, "device Type not set %s", dd.ParentId)
	}

	if device, err := dMgr.GetChildDevice(ctx, &ca.ChildDeviceFilter{
		ParentId:     dd.ParentId,
		SerialNumber: dd.SerialNumber,
		OnuId:        dd.OnuId,
		ParentPortNo: dd.ParentPortNo}); err == nil {
		logger.Warnw(ctx, "child-device-exists", log.Fields{"parent-device-id": dd.ParentId, "serialNumber": dd.SerialNumber})
		return device, status.Errorf(codes.AlreadyExists, "%s", dd.SerialNumber)
	}

	//Get parent endpoint
	pEndPoint, err := dMgr.adapterMgr.GetAdapterEndpoint(ctx, pAgent.deviceID, pAgent.deviceType)
	if err != nil {
		logger.Errorw(ctx, "endpoint-error", log.Fields{"error": err, "parent-id": pAgent.deviceID, "parent-device-type": pAgent.deviceType})
		return nil, status.Errorf(codes.NotFound, "parent-endpoint-%s", dd.ParentId)
	}

	childDevice.ProxyAddress = &voltha.Device_ProxyAddress{DeviceId: dd.ParentId, DeviceType: pAgent.deviceType, ChannelId: dd.ChannelId, OnuId: dd.OnuId, AdapterEndpoint: pEndPoint}

	// Set child device ID -- needed to get the device endpoint
	childDevice.Id = utils.CreateDeviceID()

	// Set the child adapter endpoint
	childDevice.AdapterEndpoint, err = dMgr.adapterMgr.GetAdapterEndpoint(ctx, childDevice.Id, childDevice.Type)
	if err != nil {
		return nil, status.Errorf(codes.NotFound, "child-endpoint-%s", childDevice.Id)
	}

	// Create and start a device agent for that device
	agent := newAgent(childDevice, dMgr, dMgr.dbPath, dMgr.dProxy, dMgr.internalTimeout, dMgr.rpcTimeout, dMgr.flowTimeout)
	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() {
			err := agent.enableDevice(utils.WithSpanAndRPCMetadataFromContext(ctx))
			if err != nil {
				logger.Errorw(ctx, "unable-to-enable-device", log.Fields{"error": err, "device-id": agent.deviceID})
			}
		}()
	}

	return insertedChildDevice, nil
}

func (dMgr *Manager) GetChildDevice(ctx context.Context, df *ca.ChildDeviceFilter) (*voltha.Device, error) {
	ctx = utils.WithNewSpanAndRPCMetadataContext(ctx, "GetChildDevice")
	logger.Debugw(ctx, "get-child-device", log.Fields{"filter": df})

	childDeviceIds := dMgr.getAllChildDeviceIds(ctx, df.ParentId)
	if len(childDeviceIds) == 0 {
		logger.Debugw(ctx, "no-child-devices", log.Fields{"parent-device-id": df.ParentId, "serial-number": df.SerialNumber, "onu-id": df.OnuId})
		return nil, status.Errorf(codes.NotFound, "%s", df.ParentId)
	}

	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(df.OnuId) {
				if searchDevice.ParentPortNo == uint32(df.ParentPortNo) {
					logger.Debugw(ctx, "found-child-by-onu-id", log.Fields{"parent-device-id": df.ParentId, "onuId": df.OnuId})
					foundOnuID = true
				}
			}

			foundSerialNumber := false
			if searchDevice.SerialNumber == df.SerialNumber {
				logger.Debugw(ctx, "found-child-by-serial-number", log.Fields{"parent-device-id": df.ParentId, "serialNumber": df.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 df.OnuId > 0 && df.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": df.ParentId, "foundChildDevice": foundChildDevice})
		return foundChildDevice, nil
	}

	logger.Debugw(ctx, "child-device-not-found", log.Fields{"parent-device-id": df.ParentId,
		"serialNumber": df.SerialNumber, "onuId": df.OnuId, "parentPortNo": df.ParentPortNo})
	return nil, status.Errorf(codes.NotFound, "%s", df.ParentId)
}

// PortsStateUpdate updates the operational status of all ports on the device
func (dMgr *Manager) PortsStateUpdate(ctx context.Context, ps *ca.PortStateFilter) (*empty.Empty, error) {
	ctx = utils.WithNewSpanAndRPCMetadataContext(ctx, "PortsStateUpdate")
	logger.Debugw(ctx, "ports-state-update", log.Fields{"device-id": ps.DeviceId})

	agent := dMgr.getDeviceAgent(ctx, ps.DeviceId)
	if agent == nil {
		return nil, status.Errorf(codes.NotFound, "%s", ps.DeviceId)
	}
	if ps.OperStatus != voltha.OperStatus_ACTIVE && ps.OperStatus != voltha.OperStatus_UNKNOWN {
		return nil, status.Error(codes.Unimplemented, "state-change-not-implemented")
	}
	if err := agent.updatePortsOperState(ctx, ps.PortTypeFilter, ps.OperStatus); err != nil {
		logger.Warnw(ctx, "ports-state-update-failed", log.Fields{"device-id": ps.DeviceId, "error": err})
		return nil, err
	}
	return &empty.Empty{}, nil
}

// 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, parentID *common.ID) (*empty.Empty, error) {
	ctx = utils.WithNewSpanAndRPCMetadataContext(ctx, "ChildDevicesLost")
	logger.Debugw(ctx, "child-devices-lost", log.Fields{"parent-id": parentID.Id})

	parentDevice, err := dMgr.getDeviceReadOnly(ctx, parentID.Id)
	if err != nil {
		logger.Warnw(ctx, "failed-getting-device", log.Fields{"parent-device-id": parentID.Id, "error": err})
		return nil, err
	}
	if err = dMgr.DisableAllChildDevices(ctx, parentDevice); err != nil {
		return nil, err
	}
	return &empty.Empty{}, nil
}

// 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 *common.ID) (*empty.Empty, error) {
	ctx = utils.WithNewSpanAndRPCMetadataContext(ctx, "ChildDevicesDetected")
	logger.Debugw(ctx, "child-devices-detected", log.Fields{"parent-device-id": parentDeviceID})

	var err error
	childDeviceIds := dMgr.getAllChildDeviceIds(ctx, parentDeviceID.Id)
	if len(childDeviceIds) == 0 {
		logger.Debugw(ctx, "no-child-device", log.Fields{"parent-device-id": parentDeviceID.Id})
	}
	allChildEnableRequestSent := true
	for childDeviceID := range childDeviceIds {
		if agent := dMgr.getDeviceAgent(ctx, childDeviceID); agent != nil {
			// 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})
				}
			}(log.WithSpanFromContext(context.Background(), ctx))
		} 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.Id, "childId": childDeviceID})
			allChildEnableRequestSent = false
		}
	}
	if !allChildEnableRequestSent {
		return nil, err
	}
	return &empty.Empty{}, nil
}

// GetChildDeviceWithProxyAddress will return a device based on proxy address
func (dMgr *Manager) GetChildDeviceWithProxyAddress(ctx context.Context, proxyAddress *voltha.Device_ProxyAddress) (*voltha.Device, error) {
	ctx = utils.WithNewSpanAndRPCMetadataContext(ctx, "GetChildDeviceWithProxyAddress")

	logger.Debugw(ctx, "get-child-device-with-proxy-address", log.Fields{"proxyAddress": proxyAddress})

	childDeviceIds := dMgr.getAllChildDeviceIds(ctx, proxyAddress.DeviceId)
	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{"proxyAddress": proxyAddress})
		return foundChildDevice, nil
	}

	logger.Warnw(ctx, "child-device-not-found", log.Fields{"proxyAddress": proxyAddress})
	return nil, status.Errorf(codes.NotFound, "%s", proxyAddress)
}

func (dMgr *Manager) GetPorts(ctx context.Context, pf *ca.PortFilter) (*voltha.Ports, error) {
	ctx = utils.WithNewSpanAndRPCMetadataContext(ctx, "GetPorts")
	logger.Debugw(ctx, "get-ports", log.Fields{"device-id": pf.DeviceId, "portType": pf.PortType})

	agent := dMgr.getDeviceAgent(ctx, pf.DeviceId)
	if agent == nil {
		return nil, status.Errorf(codes.NotFound, "%s", pf.DeviceId)
	}
	return agent.getPorts(ctx, pf.PortType), nil
}

func (dMgr *Manager) GetChildDevices(ctx context.Context, parentDeviceID *common.ID) (*voltha.Devices, error) {
	ctx = utils.WithNewSpanAndRPCMetadataContext(ctx, "GetChildDevices")

	logger.Debugw(ctx, "get-child-devices", log.Fields{"parent-device-id": parentDeviceID.Id})
	return dMgr.getAllChildDevices(ctx, parentDeviceID.Id)
}

func (dMgr *Manager) ChildrenStateUpdate(ctx context.Context, ds *ca.DeviceStateFilter) (*empty.Empty, error) {
	ctx = utils.WithNewSpanAndRPCMetadataContext(ctx, "ChildrenStateUpdate")
	logger.Debugw(ctx, "children-state-update", log.Fields{"parent-device-id": ds.ParentDeviceId, "operStatus": ds.OperStatus, "connStatus": ds.ConnStatus})

	for childDeviceID := range dMgr.getAllChildDeviceIds(ctx, ds.ParentDeviceId) {
		if agent := dMgr.getDeviceAgent(ctx, childDeviceID); agent != nil {
			if err := agent.updateDeviceStatus(ctx, ds.OperStatus, ds.ConnStatus); err != nil {
				return nil, status.Errorf(codes.Aborted, "childDevice:%s, error:%s", childDeviceID, err.Error())
			}
		}
	}
	return &empty.Empty{}, nil
}

func (dMgr *Manager) PortStateUpdate(ctx context.Context, ps *ca.PortState) (*empty.Empty, error) {
	ctx = utils.WithNewSpanAndRPCMetadataContext(ctx, "PortStateUpdate")
	logger.Debugw(ctx, "port-state-update", log.Fields{"device-id": ps.DeviceId, "portType": ps.PortType, "portNo": ps.PortNo, "operStatus": ps.OperStatus})

	if agent := dMgr.getDeviceAgent(ctx, ps.DeviceId); agent != nil {
		if err := agent.updatePortState(ctx, ps.PortType, ps.PortNo, ps.OperStatus); err != nil {
			logger.Errorw(ctx, "updating-port-state-failed", log.Fields{"device-id": ps.DeviceId, "portNo": ps.PortNo, "error": err})
			return nil, err
		}
		// Notify the logical device manager to change the port state
		// Do this for NNI and UNIs only. PON ports are not known by logical device
		if ps.PortType == voltha.Port_ETHERNET_NNI || ps.PortType == voltha.Port_ETHERNET_UNI {
			go func() {
				err := dMgr.logicalDeviceMgr.updatePortState(log.WithSpanFromContext(context.Background(), ctx), ps.DeviceId, ps.PortNo, ps.OperStatus)
				if err != nil {
					// While we want to handle (catch) and log when
					// an update to a port was not able to be
					// propagated to the logical port, we can report
					// it as a warning and not an error because it
					// doesn't stop or modify processing.
					// TODO: VOL-2707
					logger.Warnw(ctx, "unable-to-update-logical-port-state", log.Fields{"error": err})
				}
			}()
		}
		return &empty.Empty{}, nil
	}
	return nil, status.Errorf(codes.NotFound, "%s", ps.DeviceId)
}

func (dMgr *Manager) DeleteAllPorts(ctx context.Context, deviceID *common.ID) (*empty.Empty, error) {
	ctx = utils.WithNewSpanAndRPCMetadataContext(ctx, "DeleteAllPorts")
	logger.Debugw(ctx, "delete-all-ports", log.Fields{"device-id": deviceID.Id})

	if agent := dMgr.getDeviceAgent(ctx, deviceID.Id); agent != nil {
		if err := agent.deleteAllPorts(ctx); err != nil {
			return nil, 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.Id); 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.Id})
			return nil, err
		}
		return &empty.Empty{}, nil
	}
	return nil, status.Errorf(codes.NotFound, "%s", deviceID.Id)
}

// GetDevicePort returns the port details for a specific device port entry
func (dMgr *Manager) GetDevicePort(ctx context.Context, pf *ca.PortFilter) (*voltha.Port, error) {
	ctx = utils.WithNewSpanAndRPCMetadataContext(ctx, "GetDevicePort")
	logger.Debugw(ctx, "get-device-port", log.Fields{"device-id": pf.DeviceId})

	agent := dMgr.getDeviceAgent(ctx, pf.DeviceId)
	if agent == nil {
		return nil, status.Errorf(codes.NotFound, "device-%s", pf.DeviceId)
	}
	return agent.getDevicePort(pf.Port)
}

// DevicePMConfigUpdate updates the pm configs as defined by the adapter.
func (dMgr *Manager) DevicePMConfigUpdate(ctx context.Context, pc *voltha.PmConfigs) (*empty.Empty, error) {
	ctx = utils.WithNewSpanAndRPCMetadataContext(ctx, "DevicePMConfigUpdate")
	logger.Debugw(ctx, "device-pm-config-update", log.Fields{"device-id": pc.Id})

	if pc.Id == "" {
		return nil, status.Errorf(codes.FailedPrecondition, "invalid-device-Id")
	}
	if agent := dMgr.getDeviceAgent(ctx, pc.Id); agent != nil {
		if err := agent.initPmConfigs(ctx, pc); err != nil {
			return nil, err
		}
		return &empty.Empty{}, nil
	}
	return nil, status.Errorf(codes.NotFound, "%s", pc.Id)
}

// SendPacketIn receives packetIn request from adapter
func (dMgr *Manager) SendPacketIn(ctx context.Context, pi *ca.PacketIn) (*empty.Empty, error) {
	ctx = utils.WithNewSpanAndRPCMetadataContext(ctx, "SendPacketIn")
	logger.Debugw(ctx, "packet-in", log.Fields{"device-id": pi.DeviceId, "port": pi.Port})

	// Get the logical device Id based on the deviceId
	var device *voltha.Device
	var err error
	if device, err = dMgr.getDeviceReadOnly(ctx, pi.DeviceId); err != nil {
		logger.Errorw(ctx, "device-not-found", log.Fields{"device-id": pi.DeviceId})
		return nil, err
	}
	if !device.Root {
		logger.Errorw(ctx, "device-not-root", log.Fields{"device-id": pi.DeviceId})
		return nil, status.Errorf(codes.FailedPrecondition, "%s", pi.DeviceId)
	}

	if err := dMgr.logicalDeviceMgr.packetIn(ctx, device.ParentId, pi.Port, pi.Packet); err != nil {
		return nil, err
	}
	return &empty.Empty{}, nil
}

func (dMgr *Manager) DeviceReasonUpdate(ctx context.Context, dr *ca.DeviceReason) (*empty.Empty, error) {
	ctx = utils.WithNewSpanAndRPCMetadataContext(ctx, "DeviceReasonUpdate")
	logger.Debugw(ctx, "update-device-reason", log.Fields{"device-id": dr.DeviceId, "reason": dr.Reason})

	if agent := dMgr.getDeviceAgent(ctx, dr.DeviceId); agent != nil {
		if err := agent.updateDeviceReason(ctx, dr.Reason); err != nil {
			return nil, err
		}
		return &empty.Empty{}, nil
	}
	return nil, status.Errorf(codes.NotFound, "%s", dr.DeviceId)
}

func (dMgr *Manager) ReconcileChildDevices(ctx context.Context, parentDeviceID *common.ID) (*empty.Empty, error) {
	ctx = utils.WithNewSpanAndRPCMetadataContext(ctx, "ReconcileChildDevices")
	logger.Debugw(ctx, "reconcile-child-devices", log.Fields{"device-id": parentDeviceID.Id})

	numberOfDevicesToReconcile := 0
	dMgr.deviceAgents.Range(func(key, value interface{}) bool {
		deviceAgent, ok := value.(*Agent)
		if ok && deviceAgent.parentID == parentDeviceID.Id {
			go deviceAgent.ReconcileDevice(utils.WithNewSpanAndRPCMetadataContext(ctx, "ReconcileChildDevices"))
			numberOfDevicesToReconcile++
		}
		return true
	})
	logger.Debugw(ctx, "reconciling-child-devices-initiated", log.Fields{"parent-device-id": parentDeviceID.Id, "number-of-child-devices-to-reconcile": numberOfDevicesToReconcile})
	return &empty.Empty{}, nil
}

func (dMgr *Manager) UpdateImageDownload(ctx context.Context, img *voltha.ImageDownload) (*empty.Empty, error) {
	ctx = utils.WithNewSpanAndRPCMetadataContext(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, img.Id); 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 nil, err
		}
	} else {
		return nil, status.Errorf(codes.NotFound, "%s", img.Id)
	}
	return &empty.Empty{}, nil
}

func (dMgr *Manager) GetHealthStatus(stream core_service.CoreService_GetHealthStatusServer) error {
	ctx := utils.WithRPCMetadataContext(context.Background(), "keep-alive-connection")
	logger.Debugw(ctx, "receive-stream-connection", log.Fields{"stream": stream})

	if stream == nil {
		return fmt.Errorf("conn-is-nil %v", stream)
	}
	initialRequestTime := time.Now()
	var remoteClient *common.Connection
	var tempClient *common.Connection
	var err error
loop:
	for {
		tempClient, err = stream.Recv()
		if err != nil {
			logger.Warnw(ctx, "received-stream-error", log.Fields{"remote-client": remoteClient, "error": err})
			dMgr.adapterMgr.SignalOnRxStreamCloseCh(ctx, remoteClient.Endpoint)
			break loop
		}
		// Send a response back
		err = stream.Send(&health.HealthStatus{State: health.HealthStatus_HEALTHY})
		if err != nil {
			logger.Warnw(ctx, "sending-stream-error", log.Fields{"remote-client": remoteClient, "error": err})
			dMgr.adapterMgr.SignalOnRxStreamCloseCh(ctx, remoteClient.Endpoint)
			break loop
		}

		remoteClient = tempClient
		logger.Debugw(ctx, "received-keep-alive", log.Fields{"remote-client": remoteClient})

		select {
		case <-stream.Context().Done():
			logger.Infow(ctx, "stream-keep-alive-context-done", log.Fields{"remote-client": remoteClient, "error": stream.Context().Err()})
			break loop
		case <-dMgr.doneCh:
			logger.Warnw(ctx, "received-stop", log.Fields{"remote-client": remoteClient, "initial-conn-time": initialRequestTime})
			break loop
		default:
		}
	}
	logger.Errorw(ctx, "connection-down", log.Fields{"remote-client": remoteClient, "error": err, "initial-conn-time": initialRequestTime})
	return err
}