rw_core/core/device/logical_agent_flow.go

/*
 * 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"
	"errors"
	"fmt"
	"github.com/gogo/protobuf/proto"
	"github.com/opencord/voltha-go/rw_core/route"
	coreutils "github.com/opencord/voltha-go/rw_core/utils"
	fu "github.com/opencord/voltha-lib-go/v3/pkg/flows"
	"github.com/opencord/voltha-lib-go/v3/pkg/log"
	ofp "github.com/opencord/voltha-protos/v3/go/openflow_13"
	"github.com/opencord/voltha-protos/v3/go/voltha"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"strconv"
)

//updateFlowTable updates the flow table of that logical device
func (agent *LogicalAgent) updateFlowTable(ctx context.Context, flow *ofp.OfpFlowMod) error {
	logger.Debug("UpdateFlowTable")
	if flow == nil {
		return nil
	}

	switch flow.GetCommand() {
	case ofp.OfpFlowModCommand_OFPFC_ADD:
		return agent.flowAdd(ctx, flow)
	case ofp.OfpFlowModCommand_OFPFC_DELETE:
		return agent.flowDelete(ctx, flow)
	case ofp.OfpFlowModCommand_OFPFC_DELETE_STRICT:
		return agent.flowDeleteStrict(ctx, flow)
	case ofp.OfpFlowModCommand_OFPFC_MODIFY:
		return agent.flowModify(flow)
	case ofp.OfpFlowModCommand_OFPFC_MODIFY_STRICT:
		return agent.flowModifyStrict(flow)
	}
	return status.Errorf(codes.Internal,
		"unhandled-command: lDeviceId:%s, command:%s", agent.logicalDeviceID, flow.GetCommand())
}

//flowAdd adds a flow to the flow table of that logical device
func (agent *LogicalAgent) flowAdd(ctx context.Context, mod *ofp.OfpFlowMod) error {
	logger.Debugw("flowAdd", log.Fields{"flow": mod})
	if mod == nil {
		return nil
	}
	flow, err := fu.FlowStatsEntryFromFlowModMessage(mod)
	if err != nil {
		logger.Errorw("flowAdd-failed", log.Fields{"flowMod": mod, "err": err})
		return err
	}
	var updated bool
	var changed bool
	if changed, updated, err = agent.decomposeAndAdd(ctx, flow, mod); err != nil {
		logger.Errorw("flow-decompose-and-add-failed ", log.Fields{"flowMod": mod, "err": err})
		return err
	}
	if changed && !updated {
		if dbupdated := agent.updateFlowCountOfMeterStats(ctx, mod, flow, false); !dbupdated {
			return fmt.Errorf("couldnt-updated-flow-stats-%s", strconv.FormatUint(flow.Id, 10))
		}
	}
	return nil

}

func (agent *LogicalAgent) decomposeAndAdd(ctx context.Context, flow *ofp.OfpFlowStats, mod *ofp.OfpFlowMod) (bool, bool, error) {
	changed := false
	updated := false
	alreadyExist := true
	var flowToReplace *ofp.OfpFlowStats

	//if flow is not found in the map, create a new entry, otherwise get the existing one.
	agent.flowLock.Lock()
	flowChunk, ok := agent.flows[flow.Id]
	if !ok {
		flowChunk = &FlowChunk{
			flow: flow,
		}
		agent.flows[flow.Id] = flowChunk
		alreadyExist = false
		flowChunk.lock.Lock() //acquire chunk lock before releasing map lock
		defer flowChunk.lock.Unlock()
		agent.flowLock.Unlock()
	} else {
		agent.flowLock.Unlock() //release map lock before acquiring chunk lock
		flowChunk.lock.Lock()
		defer flowChunk.lock.Unlock()
	}

	if !alreadyExist {
		flowID := strconv.FormatUint(flow.Id, 10)
		if err := agent.clusterDataProxy.AddWithID(ctx, "logical_flows/"+agent.logicalDeviceID, flowID, flow); err != nil {
			logger.Errorw("failed-adding-flow-to-db", log.Fields{"deviceID": agent.logicalDeviceID, "flowID": flowID, "err": err})
			//Revert the map
			//TODO: Solve the condition:If we have two flow Adds of the same flow (at least same priority and match) in quick succession
			//then if the first one fails while the second one was waiting on the flowchunk, we will end up with an instance of flowChunk that is no longer in the map.
			agent.flowLock.Lock()
			delete(agent.flows, flow.Id)
			agent.flowLock.Unlock()
			return changed, updated, err
		}
	}
	flows := make([]*ofp.OfpFlowStats, 0)
	updatedFlows := make([]*ofp.OfpFlowStats, 0)
	checkOverlap := (mod.Flags & uint32(ofp.OfpFlowModFlags_OFPFF_CHECK_OVERLAP)) != 0
	if checkOverlap {
		if overlapped := fu.FindOverlappingFlows(flows, mod); len(overlapped) != 0 {
			//	TODO:  should this error be notified other than being logged?
			logger.Warnw("overlapped-flows", log.Fields{"logicaldeviceId": agent.logicalDeviceID})
		} else {
			//	Add flow
			changed = true
		}
	} else {
		if alreadyExist {
			flowToReplace = flowChunk.flow
			if (mod.Flags & uint32(ofp.OfpFlowModFlags_OFPFF_RESET_COUNTS)) != 0 {
				flow.ByteCount = flowToReplace.ByteCount
				flow.PacketCount = flowToReplace.PacketCount
			}
			if !proto.Equal(flowToReplace, flow) {
				changed = true
				updated = true
			}
		} else {
			changed = true
		}
	}
	logger.Debugw("flowAdd-changed", log.Fields{"changed": changed, "updated": updated})
	if changed {
		updatedFlows = append(updatedFlows, flow)
		var flowMetadata voltha.FlowMetadata
		lMeters, _ := agent.ListLogicalDeviceMeters(ctx)
		if err := agent.GetMeterConfig(updatedFlows, lMeters.Items, &flowMetadata); err != nil {
			logger.Error("Meter-referred-in-flow-not-present")
			return changed, updated, err
		}
		flowGroups, _ := agent.ListLogicalDeviceFlowGroups(ctx)
		deviceRules, err := agent.flowDecomposer.DecomposeRules(ctx, agent, ofp.Flows{Items: updatedFlows}, *flowGroups)
		if err != nil {
			return changed, updated, err
		}

		logger.Debugw("rules", log.Fields{"rules": deviceRules.String()})
		//	Update store and cache
		if updated {
			if err := agent.updateLogicalDeviceFlow(ctx, flow, flowChunk); err != nil {
				return changed, updated, err
			}
		}

		respChannels := agent.addFlowsAndGroupsToDevices(deviceRules, &flowMetadata)
		// Create the go routines to wait
		go func() {
			// Wait for completion
			if res := coreutils.WaitForNilOrErrorResponses(agent.defaultTimeout, respChannels...); res != nil {
				logger.Infow("failed-to-add-flows-will-attempt-deletion", log.Fields{"errors": res, "logical-device-id": agent.logicalDeviceID})
				// Revert added flows
				if err := agent.revertAddedFlows(context.Background(), mod, flow, flowToReplace, deviceRules, &flowMetadata); err != nil {
					logger.Errorw("failure-to-delete-flows-after-failed-addition", log.Fields{"logical-device-id": agent.logicalDeviceID, "error": err})
				}
			}
		}()
	}
	return changed, updated, nil
}

// revertAddedFlows reverts flows after the flowAdd request has failed.  All flows corresponding to that flowAdd request
// will be reverted, both from the logical devices and the devices.
func (agent *LogicalAgent) revertAddedFlows(ctx context.Context, mod *ofp.OfpFlowMod, addedFlow *ofp.OfpFlowStats, replacedFlow *ofp.OfpFlowStats, deviceRules *fu.DeviceRules, metadata *voltha.FlowMetadata) error {
	logger.Debugw("revertFlowAdd", log.Fields{"added-flow": addedFlow, "replaced-flow": replacedFlow, "device-rules": deviceRules, "metadata": metadata})

	agent.flowLock.RLock()
	flowChunk, ok := agent.flows[addedFlow.Id]
	agent.flowLock.RUnlock()
	if !ok {
		// Not found - do nothing
		log.Debugw("flow-not-found", log.Fields{"added-flow": addedFlow})
		return nil
	}
	//Leave the map lock and syncronize per flow
	flowChunk.lock.Lock()
	defer flowChunk.lock.Unlock()

	if replacedFlow != nil {
		if err := agent.updateLogicalDeviceFlow(ctx, replacedFlow, flowChunk); err != nil {
			return err
		}
	} else {
		if err := agent.removeLogicalDeviceFlow(ctx, addedFlow.Id); err != nil {
			return err
		}
	}
	// Revert meters
	if changedMeterStats := agent.updateFlowCountOfMeterStats(ctx, mod, addedFlow, true); !changedMeterStats {
		return fmt.Errorf("Unable-to-revert-meterstats-for-flow-%s", strconv.FormatUint(addedFlow.Id, 10))
	}

	// Update the devices
	respChnls := agent.deleteFlowsAndGroupsFromDevices(deviceRules, metadata)

	// Wait for the responses
	go func() {
		// Since this action is taken following an add failure, we may also receive a failure for the revert
		if res := coreutils.WaitForNilOrErrorResponses(agent.defaultTimeout, respChnls...); res != nil {
			logger.Warnw("failure-reverting-added-flows", log.Fields{"logicalDeviceId": agent.logicalDeviceID, "errors": res})
		}
	}()

	return nil
}

//flowDelete deletes a flow from the flow table of that logical device
func (agent *LogicalAgent) flowDelete(ctx context.Context, mod *ofp.OfpFlowMod) error {
	logger.Debug("flowDelete")
	if mod == nil {
		return nil
	}

	fs, err := fu.FlowStatsEntryFromFlowModMessage(mod)
	if err != nil {
		return err
	}

	//build a list of what to delete
	toDelete := make([]*ofp.OfpFlowStats, 0)
	toDeleteChunks := make([]*FlowChunk, 0)
	//Lock the map to search the matched flows
	agent.flowLock.RLock()
	for _, f := range agent.flows {
		if fu.FlowMatch(f.flow, fs) {
			toDelete = append(toDelete, f.flow)
			toDeleteChunks = append(toDeleteChunks, f)
			continue
		}
		// Check wild card match
		if fu.FlowMatchesMod(f.flow, mod) {
			toDelete = append(toDelete, f.flow)
			toDeleteChunks = append(toDeleteChunks, f)
		}
	}
	agent.flowLock.RUnlock()
	//Delete the matched flows
	if len(toDelete) > 0 {
		logger.Debugw("flowDelete", log.Fields{"logicalDeviceId": agent.logicalDeviceID, "toDelete": len(toDelete)})
		var meters []*ofp.OfpMeterEntry
		var flowGroups []*ofp.OfpGroupEntry
		if ofpMeters, err := agent.ListLogicalDeviceMeters(ctx); err != nil {
			meters = ofpMeters.Items
		}

		if groups, err := agent.ListLogicalDeviceFlowGroups(ctx); err != nil {
			flowGroups = groups.Items
		}

		for _, fc := range toDeleteChunks {
			if err := agent.deleteFlowAndUpdateMeterStats(ctx, mod, fc); err != nil {
				return err
			}
		}
		var flowMetadata voltha.FlowMetadata
		if err := agent.GetMeterConfig(toDelete, meters, &flowMetadata); err != nil { // This should never happen
			logger.Error("Meter-referred-in-flows-not-present")
			return err
		}
		var respChnls []coreutils.Response
		var partialRoute bool
		var deviceRules *fu.DeviceRules
		deviceRules, err = agent.flowDecomposer.DecomposeRules(ctx, agent, ofp.Flows{Items: toDelete}, ofp.FlowGroups{Items: flowGroups})
		if err != nil {
			// A no route error means no route exists between the ports specified in the flow. This can happen when the
			// child device is deleted and a request to delete flows from the parent device is received
			if !errors.Is(err, route.ErrNoRoute) {
				logger.Errorw("unexpected-error-received", log.Fields{"flows-to-delete": toDelete, "error": err})
				return err
			}
			partialRoute = true
		}

		// Update the devices
		if partialRoute {
			respChnls = agent.deleteFlowsFromParentDevice(ofp.Flows{Items: toDelete}, &flowMetadata)
		} else {
			respChnls = agent.deleteFlowsAndGroupsFromDevices(deviceRules, &flowMetadata)
		}

		// Wait for the responses
		go func() {
			// Wait for completion
			if res := coreutils.WaitForNilOrErrorResponses(agent.defaultTimeout, respChnls...); res != nil {
				logger.Errorw("failure-updating-device-flows", log.Fields{"logicalDeviceId": agent.logicalDeviceID, "errors": res})
				// TODO: Revert the flow deletion
			}
		}()
	}
	//TODO: send announcement on delete
	return nil
}

//flowDeleteStrict deletes a flow from the flow table of that logical device
func (agent *LogicalAgent) flowDeleteStrict(ctx context.Context, mod *ofp.OfpFlowMod) error {
	logger.Debugw("flowDeleteStrict", log.Fields{"mod": mod})
	if mod == nil {
		return nil
	}

	flow, err := fu.FlowStatsEntryFromFlowModMessage(mod)
	if err != nil {
		return err
	}
	logger.Debugw("flow-id-in-flow-delete-strict", log.Fields{"flowID": flow.Id})
	agent.flowLock.RLock()
	flowChunk, ok := agent.flows[flow.Id]
	agent.flowLock.RUnlock()
	if !ok {
		logger.Debugw("Skipping-flow-delete-strict-request. No-flow-found", log.Fields{"flowMod": mod})
		return nil
	}
	//Release the map lock and syncronize per flow
	flowChunk.lock.Lock()
	defer flowChunk.lock.Unlock()

	var meters []*ofp.OfpMeterEntry
	var flowGroups []*ofp.OfpGroupEntry
	if ofMeters, er := agent.ListLogicalDeviceMeters(ctx); er == nil {
		meters = ofMeters.Items
	}
	if ofGroups, er := agent.ListLogicalDeviceFlowGroups(ctx); er == nil {
		flowGroups = ofGroups.Items
	}
	if changedMeter := agent.updateFlowCountOfMeterStats(ctx, mod, flow, false); !changedMeter {
		return fmt.Errorf("Cannot delete flow - %s. Meter update failed", flow)
	}

	var flowMetadata voltha.FlowMetadata
	flowsToDelete := []*ofp.OfpFlowStats{flowChunk.flow}
	if err := agent.GetMeterConfig(flowsToDelete, meters, &flowMetadata); err != nil {
		logger.Error("meter-referred-in-flows-not-present")
		return err
	}
	var respChnls []coreutils.Response
	var partialRoute bool
	deviceRules, err := agent.flowDecomposer.DecomposeRules(ctx, agent, ofp.Flows{Items: flowsToDelete}, ofp.FlowGroups{Items: flowGroups})
	if err != nil {
		// A no route error means no route exists between the ports specified in the flow. This can happen when the
		// child device is deleted and a request to delete flows from the parent device is received
		if !errors.Is(err, route.ErrNoRoute) {
			logger.Errorw("unexpected-error-received", log.Fields{"flows-to-delete": flowsToDelete, "error": err})
			return err
		}
		partialRoute = true
	}

	// Update the model
	if err := agent.removeLogicalDeviceFlow(ctx, flow.Id); err != nil {
		return err
	}
	// Update the devices
	if partialRoute {
		respChnls = agent.deleteFlowsFromParentDevice(ofp.Flows{Items: flowsToDelete}, &flowMetadata)
	} else {
		respChnls = agent.deleteFlowsAndGroupsFromDevices(deviceRules, &flowMetadata)
	}

	// Wait for completion
	go func() {
		if res := coreutils.WaitForNilOrErrorResponses(agent.defaultTimeout, respChnls...); res != nil {
			logger.Warnw("failure-deleting-device-flows", log.Fields{"logicalDeviceId": agent.logicalDeviceID, "errors": res})
			//TODO: Revert flow changes
		}
	}()

	return nil
}

//flowModify modifies a flow from the flow table of that logical device
func (agent *LogicalAgent) flowModify(mod *ofp.OfpFlowMod) error {
	return errors.New("flowModify not implemented")
}

//flowModifyStrict deletes a flow from the flow table of that logical device
func (agent *LogicalAgent) flowModifyStrict(mod *ofp.OfpFlowMod) error {
	return errors.New("flowModifyStrict not implemented")
}