blob: ad95e9c95736015be57cfe84095b68b64381c88d [file] [log] [blame]
/*
* 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"
"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
}