| /* |
| * Copyright 2018-2023 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" |
| |
| "github.com/gogo/protobuf/proto" |
| coreutils "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" |
| 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" |
| ) |
| |
| // listDeviceGroups returns logical device flow groups |
| func (agent *Agent) listDeviceGroups() map[uint32]*ofp.OfpGroupEntry { |
| groupIDs := agent.groupCache.ListIDs() |
| groups := make(map[uint32]*ofp.OfpGroupEntry, len(groupIDs)) |
| for groupID := range groupIDs { |
| if groupHandle, have := agent.groupCache.Lock(groupID); have { |
| groups[groupID] = groupHandle.GetReadOnly() |
| groupHandle.Unlock() |
| } |
| } |
| return groups |
| } |
| |
| func (agent *Agent) addGroupsToAdapter(ctx context.Context, newGroups []*ofp.OfpGroupEntry, flowMetadata *ofp.FlowMetadata) (coreutils.Response, error) { |
| logger.Debugw(ctx, "add-groups-to-adapters", log.Fields{"device-id": agent.deviceID, "groups": newGroups, "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(newGroups)) == 0 { |
| desc = "no new groups" |
| 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 |
| } |
| |
| groupsToAdd := make([]*ofp.OfpGroupEntry, 0) |
| groupsToDelete := make([]*ofp.OfpGroupEntry, 0) |
| for _, group := range newGroups { |
| groupHandle, created, err := agent.groupCache.LockOrCreate(ctx, group) |
| if err != nil { |
| return coreutils.DoneResponse(), err |
| } |
| |
| if created { |
| groupsToAdd = append(groupsToAdd, group) |
| } else { |
| groupToChange := groupHandle.GetReadOnly() |
| if !proto.Equal(groupToChange, group) { |
| //Group needs to be updated. |
| if err = groupHandle.Update(ctx, group); err != nil { |
| groupHandle.Unlock() |
| return coreutils.DoneResponse(), err |
| } |
| groupsToDelete = append(groupsToDelete, groupToChange) |
| groupsToAdd = append(groupsToAdd, group) |
| } else { |
| //No need to change the group. It is already exist. |
| logger.Debugw(ctx, "no-need-to-change-already-existing-group", log.Fields{"device-id": agent.deviceID, "group": newGroups, "flow-metadata": flowMetadata}) |
| } |
| } |
| |
| groupHandle.Unlock() |
| } |
| // Sanity check |
| if (len(groupsToAdd)) == 0 { |
| desc = "no group to update" |
| operStatus.Code = common.OperationResp_OPERATION_SUCCESS |
| return coreutils.DoneResponse(), nil |
| } |
| |
| // Send update to adapters |
| response := coreutils.NewResponse() |
| subCtx := coreutils.WithSpanAndRPCMetadataFromContext(ctx) |
| if !dType.AcceptsAddRemoveFlowUpdates { |
| updatedAllGroups := agent.listDeviceGroups() |
| ctr, groupSlice := 0, make([]*ofp.OfpGroupEntry, len(updatedAllGroups)) |
| for _, group := range updatedAllGroups { |
| groupSlice[ctr] = group |
| ctr++ |
| } |
| go agent.sendBulkFlows(subCtx, device, nil, &ofp.FlowGroups{Items: groupSlice}, flowMetadata, response) |
| } else { |
| flowChanges := &ofp.FlowChanges{ |
| ToAdd: &ofp.Flows{Items: []*ofp.OfpFlowStats{}}, |
| ToRemove: &ofp.Flows{Items: []*ofp.OfpFlowStats{}}, |
| } |
| groupChanges := &ofp.FlowGroupChanges{ |
| ToAdd: &ofp.FlowGroups{Items: groupsToAdd}, |
| ToRemove: &ofp.FlowGroups{Items: groupsToDelete}, |
| 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) deleteGroupsFromAdapter(ctx context.Context, groupsToDel []*ofp.OfpGroupEntry, flowMetadata *ofp.FlowMetadata) (coreutils.Response, error) { |
| logger.Debugw(ctx, "delete-groups-from-adapter", log.Fields{"device-id": agent.deviceID, "groups": groupsToDel}) |
| |
| 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(groupsToDel)) == 0 { |
| desc = "nothing to delete" |
| 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 _, group := range groupsToDel { |
| if groupHandle, have := agent.groupCache.Lock(group.Desc.GroupId); have { |
| // Update the store and cache |
| if err = groupHandle.Delete(ctx); err != nil { |
| groupHandle.Unlock() |
| return coreutils.DoneResponse(), err |
| } |
| groupHandle.Unlock() |
| } |
| } |
| |
| // Send update to adapters |
| response := coreutils.NewResponse() |
| subCtx := coreutils.WithSpanAndRPCMetadataFromContext(ctx) |
| if !dType.AcceptsAddRemoveFlowUpdates { |
| updatedAllGroups := agent.listDeviceGroups() |
| ctr, groupSlice := 0, make([]*ofp.OfpGroupEntry, len(updatedAllGroups)) |
| for _, group := range updatedAllGroups { |
| groupSlice[ctr] = group |
| ctr++ |
| } |
| go agent.sendBulkFlows(subCtx, device, nil, &ofp.FlowGroups{Items: groupSlice}, flowMetadata, response) |
| } else { |
| flowChanges := &ofp.FlowChanges{ |
| ToAdd: &ofp.Flows{Items: []*ofp.OfpFlowStats{}}, |
| ToRemove: &ofp.Flows{Items: []*ofp.OfpFlowStats{}}, |
| } |
| groupChanges := &ofp.FlowGroupChanges{ |
| ToAdd: &ofp.FlowGroups{Items: []*ofp.OfpGroupEntry{}}, |
| ToRemove: &ofp.FlowGroups{Items: groupsToDel}, |
| 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) updateGroupsToAdapter(ctx context.Context, updatedGroups []*ofp.OfpGroupEntry, flowMetadata *ofp.FlowMetadata) (coreutils.Response, error) { |
| logger.Debugw(ctx, "update-groups-to-adapter", log.Fields{"device-id": agent.deviceID, "groups": updatedGroups}) |
| |
| 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(updatedGroups)) == 0 { |
| desc = "no groups to update" |
| 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-oper-%s-connect-%s-admin-%s", device.OperStatus, device.ConnectStatus, device.AdminState) |
| return coreutils.DoneResponse(), err |
| } |
| |
| dType, err := agent.adapterMgr.GetDeviceType(ctx, &voltha.ID{Id: device.Type}) |
| if err != nil { |
| err = status.Errorf(codes.FailedPrecondition, "non-existent-device-type-%s", device.Type) |
| return coreutils.DoneResponse(), err |
| } |
| |
| groupsToUpdate := make([]*ofp.OfpGroupEntry, 0) |
| for _, group := range updatedGroups { |
| if groupHandle, have := agent.groupCache.Lock(group.Desc.GroupId); have { |
| // Update the store and cache |
| if err = groupHandle.Update(ctx, group); err != nil { |
| groupHandle.Unlock() |
| return coreutils.DoneResponse(), err |
| } |
| groupsToUpdate = append(groupsToUpdate, group) |
| groupHandle.Unlock() |
| } |
| } |
| |
| response := coreutils.NewResponse() |
| subCtx := coreutils.WithSpanAndRPCMetadataFromContext(ctx) |
| // Process bulk flow update differently than incremental update |
| if !dType.AcceptsAddRemoveFlowUpdates { |
| updatedAllGroups := agent.listDeviceGroups() |
| ctr, groupSlice := 0, make([]*ofp.OfpGroupEntry, len(updatedAllGroups)) |
| for _, group := range updatedAllGroups { |
| groupSlice[ctr] = group |
| ctr++ |
| } |
| go agent.sendBulkFlows(subCtx, device, nil, &ofp.FlowGroups{Items: groupSlice}, flowMetadata, response) |
| } else { |
| logger.Debugw(ctx, "updating-groups", |
| log.Fields{ |
| "device-id": agent.deviceID, |
| "groups-to-update": groupsToUpdate, |
| }) |
| |
| // Sanity check |
| if (len(groupsToUpdate)) == 0 { |
| desc = "nothing to update" |
| operStatus.Code = common.OperationResp_OPERATION_SUCCESS |
| return coreutils.DoneResponse(), nil |
| } |
| |
| flowChanges := &ofp.FlowChanges{ |
| ToAdd: &ofp.Flows{Items: []*ofp.OfpFlowStats{}}, |
| ToRemove: &ofp.Flows{Items: []*ofp.OfpFlowStats{}}, |
| } |
| groupChanges := &ofp.FlowGroupChanges{ |
| ToAdd: &ofp.FlowGroups{Items: []*ofp.OfpGroupEntry{}}, |
| ToRemove: &ofp.FlowGroups{Items: []*ofp.OfpGroupEntry{}}, |
| ToUpdate: &ofp.FlowGroups{Items: groupsToUpdate}, |
| } |
| go agent.sendIncrementalFlows(subCtx, device, flowChanges, groupChanges, flowMetadata, response) |
| } |
| |
| operStatus.Code = common.OperationResp_OPERATION_IN_PROGRESS |
| return response, nil |
| } |