rw_core/core/device/agent_flow.go - voltha-go - Gitiles

 /*
  * Copyright 2018-present Open Networking Foundation

  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at

  * http://www.apache.org/licenses/LICENSE-2.0

  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 package device

 import (
 	"context"
 	"fmt"

 	ca "github.com/opencord/voltha-protos/v5/go/core_adapter"

 	"github.com/gogo/protobuf/proto"
 	coreutils "github.com/opencord/voltha-go/rw_core/utils"
 	fu "github.com/opencord/voltha-lib-go/v7/pkg/flows"
 	"github.com/opencord/voltha-lib-go/v7/pkg/log"
 	"github.com/opencord/voltha-protos/v5/go/common"
 	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"
 )

 // listDeviceFlows returns device flows
 func (agent *Agent) listDeviceFlows() map[uint64]*ofp.OfpFlowStats {
 	flowIDs := agent.flowCache.ListIDs()
 	flows := make(map[uint64]*ofp.OfpFlowStats, len(flowIDs))
 	for flowID := range flowIDs {
 		if flowHandle, have := agent.flowCache.Lock(flowID); have {
 			flows[flowID] = flowHandle.GetReadOnly()
 			flowHandle.Unlock()
 		}
 	}
 	return flows
 }

 func (agent *Agent) addFlowsToAdapter(ctx context.Context, newFlows []*ofp.OfpFlowStats, flowMetadata *ofp.FlowMetadata) (coreutils.Response, error) {
 	logger.Debugw(ctx, "add-flows-to-adapters", log.Fields{"device-id": agent.deviceID, "flows": newFlows, "flow-metadata": flowMetadata})

 	var err error
 	var desc string
 	operStatus := &common.OperationResp{Code: common.OperationResp_OPERATION_FAILURE}
 	defer func() { agent.logDeviceUpdate(ctx, nil, nil, operStatus, err, desc) }()

 	if (len(newFlows)) == 0 {
 		logger.Debugw(ctx, "nothing-to-update", log.Fields{"device-id": agent.deviceID, "flows": newFlows})
 		return coreutils.DoneResponse(), nil
 	}
 	device, err := agent.getDeviceReadOnly(ctx)
 	if err != nil {
 		return coreutils.DoneResponse(), status.Errorf(codes.Aborted, "%s", err)
 	}

 	if !agent.proceedWithRequest(device) {
 		err = status.Errorf(codes.FailedPrecondition, "%s", "cannot complete operation as device deletion is in progress or reconciling is in progress/failed")
 		return coreutils.DoneResponse(), err
 	}

 	dType, err := agent.adapterMgr.GetDeviceType(ctx, &voltha.ID{Id: device.Type})
 	if err != nil {
 		return coreutils.DoneResponse(), err
 	}

 	flowsToAdd := make([]*ofp.OfpFlowStats, 0)
 	flowsToDelete := make([]*ofp.OfpFlowStats, 0)
 	for _, flow := range newFlows {
 		flowHandle, created, err := agent.flowCache.LockOrCreate(ctx, flow)
 		if err != nil {
 			desc = err.Error()
 			return coreutils.DoneResponse(), err
 		}
 		if created {
 			flowsToAdd = append(flowsToAdd, flow)
 		} else {
 			flowToReplace := flowHandle.GetReadOnly()
 			if !proto.Equal(flowToReplace, flow) {
 				//Flow needs to be updated.
 				if err := flowHandle.Update(ctx, flow); err != nil {
 					flowHandle.Unlock()
 					return coreutils.DoneResponse(), err
 				}
 				flowsToDelete = append(flowsToDelete, flowToReplace)
 				flowsToAdd = append(flowsToAdd, flow)
 			} else {
 				//No need to change the flow. It is already exist.
 				logger.Debugw(ctx, "no-need-to-change-already-existing-flow", log.Fields{"device-id": agent.deviceID, "flows": newFlows, "flow-metadata": flowMetadata})
 			}
 		}
 		flowHandle.Unlock()
 	}

 	// Sanity check
 	if (len(flowsToAdd)) == 0 {
 		logger.Debugw(ctx, "no-flows-to-update", log.Fields{"device-id": agent.deviceID, "flows": newFlows})
 		operStatus.Code = common.OperationResp_OPERATION_SUCCESS
 		return coreutils.DoneResponse(), nil
 	}

 	// Send update to adapters
 	response := coreutils.NewResponse()
 	subCtx := coreutils.WithSpanAndRPCMetadataFromContext(ctx)
 	if !dType.AcceptsAddRemoveFlowUpdates {
 		updatedAllFlows := agent.listDeviceFlows()
 		ctr, flowSlice := 0, make([]*ofp.OfpFlowStats, len(updatedAllFlows))
 		for _, flow := range updatedAllFlows {
 			flowSlice[ctr] = flow
 			ctr++
 		}
 		go agent.sendBulkFlows(subCtx, device, &ofp.Flows{Items: flowSlice}, nil, flowMetadata, response)
 	} else {
 		flowChanges := &ofp.FlowChanges{
 			ToAdd:    &ofp.Flows{Items: flowsToAdd},
 			ToRemove: &ofp.Flows{Items: flowsToDelete},
 		}
 		groupChanges := &ofp.FlowGroupChanges{
 			ToAdd:    &ofp.FlowGroups{Items: []*ofp.OfpGroupEntry{}},
 			ToRemove: &ofp.FlowGroups{Items: []*ofp.OfpGroupEntry{}},
 			ToUpdate: &ofp.FlowGroups{Items: []*ofp.OfpGroupEntry{}},
 		}
 		go agent.sendIncrementalFlows(subCtx, device, flowChanges, groupChanges, flowMetadata, response)
 	}
 	operStatus.Code = common.OperationResp_OPERATION_IN_PROGRESS
 	return response, nil
 }

 func (agent *Agent) sendBulkFlows(
 	ctx context.Context,
 	device *voltha.Device,
 	flows *ofp.Flows,
 	groups *ofp.FlowGroups,
 	flowMetadata *ofp.FlowMetadata,
 	response coreutils.Response,
 ) {
 	var err error
 	var desc string
 	operStatus := &common.OperationResp{Code: common.OperationResp_OPERATION_FAILURE}
 	defer func() { agent.logDeviceUpdate(ctx, nil, nil, operStatus, err, desc) }()

 	// Get a grpc client
 	client, err := agent.adapterMgr.GetAdapterClient(ctx, agent.adapterEndpoint)
 	if err != nil {
 		logger.Errorw(ctx, "grpc-client-nil",
 			log.Fields{
 				"error":            err,
 				"device-id":        agent.deviceID,
 				"device-type":      agent.deviceType,
 				"adapter-endpoint": device.AdapterEndpoint,
 			})
 		response.Error(err)
 		return
 	}
 	subCtx, cancel := context.WithTimeout(ctx, agent.flowTimeout)
 	defer cancel()

 	if _, err = client.UpdateFlowsBulk(subCtx, &ca.BulkFlows{
 		Device:       device,
 		Flows:        flows,
 		Groups:       groups,
 		FlowMetadata: flowMetadata,
 	}); err != nil {
 		response.Error(err)
 	} else {
 		response.Done()
 		operStatus.Code = common.OperationResp_OPERATION_SUCCESS
 	}
 }

 func (agent *Agent) sendIncrementalFlows(
 	ctx context.Context,
 	device *voltha.Device,
 	flowChanges *ofp.FlowChanges,
 	groupChanges *ofp.FlowGroupChanges,
 	flowMetadata *ofp.FlowMetadata,
 	response coreutils.Response,
 ) {
 	var err error
 	var desc string
 	operStatus := &common.OperationResp{Code: common.OperationResp_OPERATION_FAILURE}
 	defer func() { agent.logDeviceUpdate(ctx, nil, nil, operStatus, err, desc) }()

 	// Get a grpc client
 	client, err := agent.adapterMgr.GetAdapterClient(ctx, agent.adapterEndpoint)
 	if err != nil {
 		logger.Errorw(ctx, "grpc-client-nil",
 			log.Fields{
 				"error":            err,
 				"device-id":        agent.deviceID,
 				"device-type":      agent.deviceType,
 				"adapter-endpoint": device.AdapterEndpoint,
 			})
 		response.Error(err)
 		return
 	}
 	subCtx, cancel := context.WithTimeout(ctx, agent.flowTimeout)
 	defer cancel()
 	if _, err = client.UpdateFlowsIncrementally(subCtx, &ca.IncrementalFlows{
 		Device:       device,
 		Flows:        flowChanges,
 		Groups:       groupChanges,
 		FlowMetadata: flowMetadata,
 	}); err != nil {
 		response.Error(err)
 	} else {
 		response.Done()
 		operStatus.Code = common.OperationResp_OPERATION_SUCCESS
 	}
 }

 func (agent *Agent) deleteFlowsFromAdapter(ctx context.Context, flowsToDel []*ofp.OfpFlowStats, flowMetadata *ofp.FlowMetadata) (coreutils.Response, error) {
 	logger.Debugw(ctx, "delete-flows-from-adapter", log.Fields{"device-id": agent.deviceID, "flows": flowsToDel})

 	var desc string
 	var err error
 	operStatus := &common.OperationResp{Code: common.OperationResp_OPERATION_FAILURE}
 	defer func() { agent.logDeviceUpdate(ctx, nil, nil, operStatus, err, desc) }()

 	if (len(flowsToDel)) == 0 {
 		logger.Debugw(ctx, "nothing-to-delete", log.Fields{"device-id": agent.deviceID, "flows": flowsToDel})
 		operStatus.Code = common.OperationResp_OPERATION_SUCCESS
 		return coreutils.DoneResponse(), nil
 	}

 	device, err := agent.getDeviceReadOnly(ctx)
 	if err != nil {
 		return coreutils.DoneResponse(), err
 	}

 	if !agent.proceedWithRequest(device) {
 		err = status.Errorf(codes.FailedPrecondition, "%s", "cannot complete operation as device deletion is in progress or reconciling is in progress/failed")
 		return coreutils.DoneResponse(), err
 	}

 	dType, err := agent.adapterMgr.GetDeviceType(ctx, &voltha.ID{Id: device.Type})
 	if err != nil {
 		return coreutils.DoneResponse(), err
 	}

 	for _, flow := range flowsToDel {
 		if flowHandle, have := agent.flowCache.Lock(flow.Id); have {
 			// Update the store and cache
 			if err = flowHandle.Delete(ctx); err != nil {
 				flowHandle.Unlock()
 				desc = err.Error()
 				return coreutils.DoneResponse(), err
 			}
 			flowHandle.Unlock()
 		}
 	}

 	// Send update to adapters
 	response := coreutils.NewResponse()
 	subCtx := coreutils.WithSpanAndRPCMetadataFromContext(ctx)
 	if !dType.AcceptsAddRemoveFlowUpdates {
 		updatedAllFlows := agent.listDeviceFlows()
 		ctr, flowSlice := 0, make([]*ofp.OfpFlowStats, len(updatedAllFlows))
 		for _, flow := range updatedAllFlows {
 			flowSlice[ctr] = flow
 			ctr++
 		}
 		go agent.sendBulkFlows(subCtx, device, &ofp.Flows{Items: flowSlice}, nil, flowMetadata, response)
 	} else {
 		flowChanges := &ofp.FlowChanges{
 			ToAdd:    &ofp.Flows{Items: []*ofp.OfpFlowStats{}},
 			ToRemove: &ofp.Flows{Items: flowsToDel},
 		}
 		groupChanges := &ofp.FlowGroupChanges{
 			ToAdd:    &ofp.FlowGroups{Items: []*ofp.OfpGroupEntry{}},
 			ToRemove: &ofp.FlowGroups{Items: []*ofp.OfpGroupEntry{}},
 			ToUpdate: &ofp.FlowGroups{Items: []*ofp.OfpGroupEntry{}},
 		}
 		go agent.sendIncrementalFlows(subCtx, device, flowChanges, groupChanges, flowMetadata, response)
 	}
 	operStatus.Code = common.OperationResp_OPERATION_IN_PROGRESS
 	return response, nil
 }

 func (agent *Agent) updateFlowsToAdapter(ctx context.Context, updatedFlows []*ofp.OfpFlowStats, flowMetadata *ofp.FlowMetadata) (coreutils.Response, error) {
 	logger.Debugw(ctx, "update-flows-to-adapter", log.Fields{"device-id": agent.deviceID, "flows": updatedFlows})

 	var err error
 	var desc string
 	operStatus := &common.OperationResp{Code: common.OperationResp_OPERATION_FAILURE}
 	defer func() { agent.logDeviceUpdate(ctx, nil, nil, operStatus, err, desc) }()

 	if (len(updatedFlows)) == 0 {
 		logger.Debugw(ctx, "nothing-to-update", log.Fields{"device-id": agent.deviceID, "flows": updatedFlows})
 		operStatus.Code = common.OperationResp_OPERATION_SUCCESS
 		return coreutils.DoneResponse(), nil
 	}

 	device, err := agent.getDeviceReadOnly(ctx)
 	if err != nil {
 		return coreutils.DoneResponse(), err
 	}

 	if !agent.proceedWithRequest(device) {
 		err = status.Errorf(codes.FailedPrecondition, "%s", "cannot complete operation as device deletion is in progress or reconciling is in progress/failed")
 		return coreutils.DoneResponse(), err
 	}

 	if device.OperStatus != voltha.OperStatus_ACTIVE || device.ConnectStatus != voltha.ConnectStatus_REACHABLE || device.AdminState != voltha.AdminState_ENABLED {
 		err = status.Errorf(codes.FailedPrecondition, "invalid device states")
 		return coreutils.DoneResponse(), err
 	}

 	dType, err := agent.adapterMgr.GetDeviceType(ctx, &voltha.ID{Id: device.Type})
 	if err != nil {
 		return coreutils.DoneResponse(), err
 	}

 	flowsToAdd := make([]*ofp.OfpFlowStats, 0, len(updatedFlows))
 	flowsToDelete := make([]*ofp.OfpFlowStats, 0, len(updatedFlows))
 	for _, flow := range updatedFlows {
 		if flowHandle, have := agent.flowCache.Lock(flow.Id); have {
 			flowToDelete := flowHandle.GetReadOnly()
 			// Update the store and cache
 			if err = flowHandle.Update(ctx, flow); err != nil {
 				flowHandle.Unlock()
 				return coreutils.DoneResponse(), err
 			}

 			flowsToDelete = append(flowsToDelete, flowToDelete)
 			flowsToAdd = append(flowsToAdd, flow)
 			flowHandle.Unlock()
 		}
 	}

 	response := coreutils.NewResponse()
 	subCtx := coreutils.WithSpanAndRPCMetadataFromContext(ctx)
 	// Process bulk flow update differently than incremental update
 	if !dType.AcceptsAddRemoveFlowUpdates {
 		updatedAllFlows := agent.listDeviceFlows()
 		ctr, flowSlice := 0, make([]*ofp.OfpFlowStats, len(updatedAllFlows))
 		for _, flow := range updatedAllFlows {
 			flowSlice[ctr] = flow
 			ctr++
 		}
 		go agent.sendBulkFlows(subCtx, device, &ofp.Flows{Items: flowSlice}, nil, flowMetadata, response)
 	} else {
 		logger.Debugw(ctx, "updating-flows-and-groups",
 			log.Fields{
 				"device-id":       agent.deviceID,
 				"flows-to-add":    flowsToAdd,
 				"flows-to-delete": flowsToDelete,
 			})
 		// Sanity check
 		if (len(flowsToAdd) | len(flowsToDelete)) == 0 {
 			logger.Debugw(ctx, "nothing-to-update", log.Fields{"device-id": agent.deviceID, "flows": updatedFlows})
 			operStatus.Code = common.OperationResp_OPERATION_SUCCESS
 			return coreutils.DoneResponse(), nil
 		}

 		flowChanges := &ofp.FlowChanges{
 			ToAdd:    &ofp.Flows{Items: flowsToAdd},
 			ToRemove: &ofp.Flows{Items: flowsToDelete},
 		}
 		groupChanges := &ofp.FlowGroupChanges{
 			ToAdd:    &ofp.FlowGroups{Items: []*ofp.OfpGroupEntry{}},
 			ToRemove: &ofp.FlowGroups{Items: []*ofp.OfpGroupEntry{}},
 			ToUpdate: &ofp.FlowGroups{Items: []*ofp.OfpGroupEntry{}},
 		}
 		go agent.sendIncrementalFlows(subCtx, device, flowChanges, groupChanges, flowMetadata, response)
 	}
 	operStatus.Code = common.OperationResp_OPERATION_IN_PROGRESS
 	return response, nil
 }

 //filterOutFlows removes flows from a device using the uni-port as filter
 func (agent *Agent) filterOutFlows(ctx context.Context, uniPort uint32, flowMetadata *ofp.FlowMetadata) error {
 	var flowsToDelete []*ofp.OfpFlowStats
 	// If an existing flow has the uniPort as an InPort or OutPort or as a Tunnel ID then it needs to be removed
 	for flowID := range agent.flowCache.ListIDs() {
 		if flowHandle, have := agent.flowCache.Lock(flowID); have {
 			flow := flowHandle.GetReadOnly()
 			if flow != nil && (fu.GetInPort(flow) == uniPort || fu.GetOutPort(flow) == uniPort || fu.GetTunnelId(flow) == uint64(uniPort)) {
 				flowsToDelete = append(flowsToDelete, flow)
 			}
 			flowHandle.Unlock()
 		}
 	}

 	logger.Debugw(ctx, "flows-to-delete", log.Fields{"device-id": agent.deviceID, "uni-port": uniPort, "flows": flowsToDelete})
 	if len(flowsToDelete) == 0 {
 		return nil
 	}

 	response, err := agent.deleteFlowsFromAdapter(ctx, flowsToDelete, flowMetadata)
 	if err != nil {
 		return err
 	}
 	if res := coreutils.WaitForNilOrErrorResponses(agent.flowTimeout, response); res != nil {
 		return status.Errorf(codes.Aborted, "errors-%s", res)
 	}
 	return nil
 }

 //deleteAllFlows deletes all flows in the device table
 func (agent *Agent) deleteAllFlows(ctx context.Context) error {
 	logger.Debugw(ctx, "deleteAllFlows", log.Fields{"device-id": agent.deviceID})

 	var err error
 	var errFlows string
 	var desc string
 	operStatus := &common.OperationResp{Code: common.OperationResp_OPERATION_FAILURE}
 	defer func() { agent.logDeviceUpdate(ctx, nil, nil, operStatus, err, desc) }()

 	for flowID := range agent.flowCache.ListIDs() {
 		if flowHandle, have := agent.flowCache.Lock(flowID); have {
 			// Update the store and cache
 			if err = flowHandle.Delete(ctx); err != nil {
 				flowHandle.Unlock()
 				errFlows += fmt.Sprintf("%v ", flowID)
 				logger.Errorw(ctx, "unable-to-delete-flow", log.Fields{"device-id": agent.deviceID, "flowID": flowID, "error": err})
 				continue
 			}
 			flowHandle.Unlock()
 		}
 	}

 	if errFlows != "" {
 		err = fmt.Errorf("unable to delete flows : %s", errFlows)
 	} else {
 		operStatus.Code = common.OperationResp_OPERATION_SUCCESS
 	}
 	return err
 }
	/*
	* Copyright 2018-present Open Networking Foundation

	* Licensed under the Apache License, Version 2.0 (the "License");
	* you may not use this file except in compliance with the License.
	* You may obtain a copy of the License at

	* http://www.apache.org/licenses/LICENSE-2.0

	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	package device

	import (
	"context"
	"fmt"

	ca "github.com/opencord/voltha-protos/v5/go/core_adapter"

	"github.com/gogo/protobuf/proto"
	coreutils "github.com/opencord/voltha-go/rw_core/utils"
	fu "github.com/opencord/voltha-lib-go/v7/pkg/flows"
	"github.com/opencord/voltha-lib-go/v7/pkg/log"
	"github.com/opencord/voltha-protos/v5/go/common"
	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"
	)

	// listDeviceFlows returns device flows
	func (agent Agent) listDeviceFlows() map[uint64]ofp.OfpFlowStats {
	flowIDs := agent.flowCache.ListIDs()
	flows := make(map[uint64]*ofp.OfpFlowStats, len(flowIDs))
	for flowID := range flowIDs {
	if flowHandle, have := agent.flowCache.Lock(flowID); have {
	flows[flowID] = flowHandle.GetReadOnly()
	flowHandle.Unlock()
	}
	}
	return flows
	}

	func (agent Agent) addFlowsToAdapter(ctx context.Context, newFlows []ofp.OfpFlowStats, flowMetadata *ofp.FlowMetadata) (coreutils.Response, error) {
	logger.Debugw(ctx, "add-flows-to-adapters", log.Fields{"device-id": agent.deviceID, "flows": newFlows, "flow-metadata": flowMetadata})

	var err error
	var desc string
	operStatus := &common.OperationResp{Code: common.OperationResp_OPERATION_FAILURE}
	defer func() { agent.logDeviceUpdate(ctx, nil, nil, operStatus, err, desc) }()

	if (len(newFlows)) == 0 {
	logger.Debugw(ctx, "nothing-to-update", log.Fields{"device-id": agent.deviceID, "flows": newFlows})
	return coreutils.DoneResponse(), nil
	}
	device, err := agent.getDeviceReadOnly(ctx)
	if err != nil {
	return coreutils.DoneResponse(), status.Errorf(codes.Aborted, "%s", err)
	}

	if !agent.proceedWithRequest(device) {
	err = status.Errorf(codes.FailedPrecondition, "%s", "cannot complete operation as device deletion is in progress or reconciling is in progress/failed")
	return coreutils.DoneResponse(), err
	}

	dType, err := agent.adapterMgr.GetDeviceType(ctx, &voltha.ID{Id: device.Type})
	if err != nil {
	return coreutils.DoneResponse(), err
	}

	flowsToAdd := make([]*ofp.OfpFlowStats, 0)
	flowsToDelete := make([]*ofp.OfpFlowStats, 0)
	for _, flow := range newFlows {
	flowHandle, created, err := agent.flowCache.LockOrCreate(ctx, flow)
	if err != nil {
	desc = err.Error()
	return coreutils.DoneResponse(), err
	}
	if created {
	flowsToAdd = append(flowsToAdd, flow)
	} else {
	flowToReplace := flowHandle.GetReadOnly()
	if !proto.Equal(flowToReplace, flow) {
	//Flow needs to be updated.
	if err := flowHandle.Update(ctx, flow); err != nil {
	flowHandle.Unlock()
	return coreutils.DoneResponse(), err
	}
	flowsToDelete = append(flowsToDelete, flowToReplace)
	flowsToAdd = append(flowsToAdd, flow)
	} else {
	//No need to change the flow. It is already exist.
	logger.Debugw(ctx, "no-need-to-change-already-existing-flow", log.Fields{"device-id": agent.deviceID, "flows": newFlows, "flow-metadata": flowMetadata})
	}
	}
	flowHandle.Unlock()
	}

	// Sanity check
	if (len(flowsToAdd)) == 0 {
	logger.Debugw(ctx, "no-flows-to-update", log.Fields{"device-id": agent.deviceID, "flows": newFlows})
	operStatus.Code = common.OperationResp_OPERATION_SUCCESS
	return coreutils.DoneResponse(), nil
	}

	// Send update to adapters
	response := coreutils.NewResponse()
	subCtx := coreutils.WithSpanAndRPCMetadataFromContext(ctx)
	if !dType.AcceptsAddRemoveFlowUpdates {
	updatedAllFlows := agent.listDeviceFlows()
	ctr, flowSlice := 0, make([]*ofp.OfpFlowStats, len(updatedAllFlows))
	for _, flow := range updatedAllFlows {
	flowSlice[ctr] = flow
	ctr++
	}
	go agent.sendBulkFlows(subCtx, device, &ofp.Flows{Items: flowSlice}, nil, flowMetadata, response)
	} else {
	flowChanges := &ofp.FlowChanges{
	ToAdd: &ofp.Flows{Items: flowsToAdd},
	ToRemove: &ofp.Flows{Items: flowsToDelete},
	}
	groupChanges := &ofp.FlowGroupChanges{
	ToAdd: &ofp.FlowGroups{Items: []*ofp.OfpGroupEntry{}},
	ToRemove: &ofp.FlowGroups{Items: []*ofp.OfpGroupEntry{}},
	ToUpdate: &ofp.FlowGroups{Items: []*ofp.OfpGroupEntry{}},
	}
	go agent.sendIncrementalFlows(subCtx, device, flowChanges, groupChanges, flowMetadata, response)
	}
	operStatus.Code = common.OperationResp_OPERATION_IN_PROGRESS
	return response, nil
	}

	func (agent *Agent) sendBulkFlows(
	ctx context.Context,
	device *voltha.Device,
	flows *ofp.Flows,
	groups *ofp.FlowGroups,
	flowMetadata *ofp.FlowMetadata,
	response coreutils.Response,
	) {
	var err error
	var desc string
	operStatus := &common.OperationResp{Code: common.OperationResp_OPERATION_FAILURE}
	defer func() { agent.logDeviceUpdate(ctx, nil, nil, operStatus, err, desc) }()

	// Get a grpc client
	client, err := agent.adapterMgr.GetAdapterClient(ctx, agent.adapterEndpoint)
	if err != nil {
	logger.Errorw(ctx, "grpc-client-nil",
	log.Fields{
	"error": err,
	"device-id": agent.deviceID,
	"device-type": agent.deviceType,
	"adapter-endpoint": device.AdapterEndpoint,
	})
	response.Error(err)
	return
	}
	subCtx, cancel := context.WithTimeout(ctx, agent.flowTimeout)
	defer cancel()

	if _, err = client.UpdateFlowsBulk(subCtx, &ca.BulkFlows{
	Device: device,
	Flows: flows,
	Groups: groups,
	FlowMetadata: flowMetadata,
	}); err != nil {
	response.Error(err)
	} else {
	response.Done()
	operStatus.Code = common.OperationResp_OPERATION_SUCCESS
	}
	}

	func (agent *Agent) sendIncrementalFlows(
	ctx context.Context,
	device *voltha.Device,
	flowChanges *ofp.FlowChanges,
	groupChanges *ofp.FlowGroupChanges,
	flowMetadata *ofp.FlowMetadata,
	response coreutils.Response,
	) {
	var err error
	var desc string
	operStatus := &common.OperationResp{Code: common.OperationResp_OPERATION_FAILURE}
	defer func() { agent.logDeviceUpdate(ctx, nil, nil, operStatus, err, desc) }()

	// Get a grpc client
	client, err := agent.adapterMgr.GetAdapterClient(ctx, agent.adapterEndpoint)
	if err != nil {
	logger.Errorw(ctx, "grpc-client-nil",
	log.Fields{
	"error": err,
	"device-id": agent.deviceID,
	"device-type": agent.deviceType,
	"adapter-endpoint": device.AdapterEndpoint,
	})
	response.Error(err)
	return
	}
	subCtx, cancel := context.WithTimeout(ctx, agent.flowTimeout)
	defer cancel()
	if _, err = client.UpdateFlowsIncrementally(subCtx, &ca.IncrementalFlows{
	Device: device,
	Flows: flowChanges,
	Groups: groupChanges,
	FlowMetadata: flowMetadata,
	}); err != nil {
	response.Error(err)
	} else {
	response.Done()
	operStatus.Code = common.OperationResp_OPERATION_SUCCESS
	}
	}

	func (agent Agent) deleteFlowsFromAdapter(ctx context.Context, flowsToDel []ofp.OfpFlowStats, flowMetadata *ofp.FlowMetadata) (coreutils.Response, error) {
	logger.Debugw(ctx, "delete-flows-from-adapter", log.Fields{"device-id": agent.deviceID, "flows": flowsToDel})

	var desc string
	var err error
	operStatus := &common.OperationResp{Code: common.OperationResp_OPERATION_FAILURE}
	defer func() { agent.logDeviceUpdate(ctx, nil, nil, operStatus, err, desc) }()

	if (len(flowsToDel)) == 0 {
	logger.Debugw(ctx, "nothing-to-delete", log.Fields{"device-id": agent.deviceID, "flows": flowsToDel})
	operStatus.Code = common.OperationResp_OPERATION_SUCCESS
	return coreutils.DoneResponse(), nil
	}

	device, err := agent.getDeviceReadOnly(ctx)
	if err != nil {
	return coreutils.DoneResponse(), err
	}

	if !agent.proceedWithRequest(device) {
	err = status.Errorf(codes.FailedPrecondition, "%s", "cannot complete operation as device deletion is in progress or reconciling is in progress/failed")
	return coreutils.DoneResponse(), err
	}

	dType, err := agent.adapterMgr.GetDeviceType(ctx, &voltha.ID{Id: device.Type})
	if err != nil {
	return coreutils.DoneResponse(), err
	}

	for _, flow := range flowsToDel {
	if flowHandle, have := agent.flowCache.Lock(flow.Id); have {
	// Update the store and cache
	if err = flowHandle.Delete(ctx); err != nil {
	flowHandle.Unlock()
	desc = err.Error()
	return coreutils.DoneResponse(), err
	}
	flowHandle.Unlock()
	}
	}

	// Send update to adapters
	response := coreutils.NewResponse()
	subCtx := coreutils.WithSpanAndRPCMetadataFromContext(ctx)
	if !dType.AcceptsAddRemoveFlowUpdates {
	updatedAllFlows := agent.listDeviceFlows()
	ctr, flowSlice := 0, make([]*ofp.OfpFlowStats, len(updatedAllFlows))
	for _, flow := range updatedAllFlows {
	flowSlice[ctr] = flow
	ctr++
	}
	go agent.sendBulkFlows(subCtx, device, &ofp.Flows{Items: flowSlice}, nil, flowMetadata, response)
	} else {
	flowChanges := &ofp.FlowChanges{
	ToAdd: &ofp.Flows{Items: []*ofp.OfpFlowStats{}},
	ToRemove: &ofp.Flows{Items: flowsToDel},
	}
	groupChanges := &ofp.FlowGroupChanges{
	ToAdd: &ofp.FlowGroups{Items: []*ofp.OfpGroupEntry{}},
	ToRemove: &ofp.FlowGroups{Items: []*ofp.OfpGroupEntry{}},
	ToUpdate: &ofp.FlowGroups{Items: []*ofp.OfpGroupEntry{}},
	}
	go agent.sendIncrementalFlows(subCtx, device, flowChanges, groupChanges, flowMetadata, response)
	}
	operStatus.Code = common.OperationResp_OPERATION_IN_PROGRESS
	return response, nil
	}

	func (agent Agent) updateFlowsToAdapter(ctx context.Context, updatedFlows []ofp.OfpFlowStats, flowMetadata *ofp.FlowMetadata) (coreutils.Response, error) {
	logger.Debugw(ctx, "update-flows-to-adapter", log.Fields{"device-id": agent.deviceID, "flows": updatedFlows})

	var err error
	var desc string
	operStatus := &common.OperationResp{Code: common.OperationResp_OPERATION_FAILURE}
	defer func() { agent.logDeviceUpdate(ctx, nil, nil, operStatus, err, desc) }()

	if (len(updatedFlows)) == 0 {
	logger.Debugw(ctx, "nothing-to-update", log.Fields{"device-id": agent.deviceID, "flows": updatedFlows})
	operStatus.Code = common.OperationResp_OPERATION_SUCCESS
	return coreutils.DoneResponse(), nil
	}

	device, err := agent.getDeviceReadOnly(ctx)
	if err != nil {
	return coreutils.DoneResponse(), err
	}

	if !agent.proceedWithRequest(device) {
	err = status.Errorf(codes.FailedPrecondition, "%s", "cannot complete operation as device deletion is in progress or reconciling is in progress/failed")
	return coreutils.DoneResponse(), err
	}

	if device.OperStatus != voltha.OperStatus_ACTIVE \|\| device.ConnectStatus != voltha.ConnectStatus_REACHABLE \|\| device.AdminState != voltha.AdminState_ENABLED {
	err = status.Errorf(codes.FailedPrecondition, "invalid device states")
	return coreutils.DoneResponse(), err
	}

	dType, err := agent.adapterMgr.GetDeviceType(ctx, &voltha.ID{Id: device.Type})
	if err != nil {
	return coreutils.DoneResponse(), err
	}

	flowsToAdd := make([]*ofp.OfpFlowStats, 0, len(updatedFlows))
	flowsToDelete := make([]*ofp.OfpFlowStats, 0, len(updatedFlows))
	for _, flow := range updatedFlows {
	if flowHandle, have := agent.flowCache.Lock(flow.Id); have {
	flowToDelete := flowHandle.GetReadOnly()
	// Update the store and cache
	if err = flowHandle.Update(ctx, flow); err != nil {
	flowHandle.Unlock()
	return coreutils.DoneResponse(), err
	}

	flowsToDelete = append(flowsToDelete, flowToDelete)
	flowsToAdd = append(flowsToAdd, flow)
	flowHandle.Unlock()
	}
	}

	response := coreutils.NewResponse()
	subCtx := coreutils.WithSpanAndRPCMetadataFromContext(ctx)
	// Process bulk flow update differently than incremental update
	if !dType.AcceptsAddRemoveFlowUpdates {
	updatedAllFlows := agent.listDeviceFlows()
	ctr, flowSlice := 0, make([]*ofp.OfpFlowStats, len(updatedAllFlows))
	for _, flow := range updatedAllFlows {
	flowSlice[ctr] = flow
	ctr++
	}
	go agent.sendBulkFlows(subCtx, device, &ofp.Flows{Items: flowSlice}, nil, flowMetadata, response)
	} else {
	logger.Debugw(ctx, "updating-flows-and-groups",
	log.Fields{
	"device-id": agent.deviceID,
	"flows-to-add": flowsToAdd,
	"flows-to-delete": flowsToDelete,
	})
	// Sanity check
	if (len(flowsToAdd) \| len(flowsToDelete)) == 0 {
	logger.Debugw(ctx, "nothing-to-update", log.Fields{"device-id": agent.deviceID, "flows": updatedFlows})
	operStatus.Code = common.OperationResp_OPERATION_SUCCESS
	return coreutils.DoneResponse(), nil
	}

	flowChanges := &ofp.FlowChanges{
	ToAdd: &ofp.Flows{Items: flowsToAdd},
	ToRemove: &ofp.Flows{Items: flowsToDelete},
	}
	groupChanges := &ofp.FlowGroupChanges{
	ToAdd: &ofp.FlowGroups{Items: []*ofp.OfpGroupEntry{}},
	ToRemove: &ofp.FlowGroups{Items: []*ofp.OfpGroupEntry{}},
	ToUpdate: &ofp.FlowGroups{Items: []*ofp.OfpGroupEntry{}},
	}
	go agent.sendIncrementalFlows(subCtx, device, flowChanges, groupChanges, flowMetadata, response)
	}
	operStatus.Code = common.OperationResp_OPERATION_IN_PROGRESS
	return response, nil
	}

	//filterOutFlows removes flows from a device using the uni-port as filter
	func (agent Agent) filterOutFlows(ctx context.Context, uniPort uint32, flowMetadata ofp.FlowMetadata) error {
	var flowsToDelete []*ofp.OfpFlowStats
	// If an existing flow has the uniPort as an InPort or OutPort or as a Tunnel ID then it needs to be removed
	for flowID := range agent.flowCache.ListIDs() {
	if flowHandle, have := agent.flowCache.Lock(flowID); have {
	flow := flowHandle.GetReadOnly()
	if flow != nil && (fu.GetInPort(flow) == uniPort \|\| fu.GetOutPort(flow) == uniPort \|\| fu.GetTunnelId(flow) == uint64(uniPort)) {
	flowsToDelete = append(flowsToDelete, flow)
	}
	flowHandle.Unlock()
	}
	}

	logger.Debugw(ctx, "flows-to-delete", log.Fields{"device-id": agent.deviceID, "uni-port": uniPort, "flows": flowsToDelete})
	if len(flowsToDelete) == 0 {
	return nil
	}

	response, err := agent.deleteFlowsFromAdapter(ctx, flowsToDelete, flowMetadata)
	if err != nil {
	return err
	}
	if res := coreutils.WaitForNilOrErrorResponses(agent.flowTimeout, response); res != nil {
	return status.Errorf(codes.Aborted, "errors-%s", res)
	}
	return nil
	}

	//deleteAllFlows deletes all flows in the device table
	func (agent *Agent) deleteAllFlows(ctx context.Context) error {
	logger.Debugw(ctx, "deleteAllFlows", log.Fields{"device-id": agent.deviceID})

	var err error
	var errFlows string
	var desc string
	operStatus := &common.OperationResp{Code: common.OperationResp_OPERATION_FAILURE}
	defer func() { agent.logDeviceUpdate(ctx, nil, nil, operStatus, err, desc) }()

	for flowID := range agent.flowCache.ListIDs() {
	if flowHandle, have := agent.flowCache.Lock(flowID); have {
	// Update the store and cache
	if err = flowHandle.Delete(ctx); err != nil {
	flowHandle.Unlock()
	errFlows += fmt.Sprintf("%v ", flowID)
	logger.Errorw(ctx, "unable-to-delete-flow", log.Fields{"device-id": agent.deviceID, "flowID": flowID, "error": err})
	continue
	}
	flowHandle.Unlock()
	}
	}

	if errFlows != "" {
	err = fmt.Errorf("unable to delete flows : %s", errFlows)
	} else {
	operStatus.Code = common.OperationResp_OPERATION_SUCCESS
	}
	return err
	}