[VOL-1037, VOL-1035] This commit consists of flow and groups
handling (from NBI to Adapters, including decomposition),
Change-Id: I4f6d9ecd3dee8a9b161708b20b0a68d030c0cb23
diff --git a/rw_core/core/adapter_proxy.go b/rw_core/core/adapter_proxy.go
index 82c6dec..643c9de 100644
--- a/rw_core/core/adapter_proxy.go
+++ b/rw_core/core/adapter_proxy.go
@@ -22,6 +22,7 @@
"github.com/opencord/voltha-go/common/log"
"github.com/opencord/voltha-go/kafka"
ca "github.com/opencord/voltha-go/protos/core_adapter"
+ "github.com/opencord/voltha-go/protos/openflow_13"
"github.com/opencord/voltha-go/protos/voltha"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
@@ -249,14 +250,50 @@
return nil, nil
}
-func (ap *AdapterProxy) UpdateFlowsBulk(device voltha.Device, flows voltha.Flows, groups voltha.FlowGroups) error {
- log.Debug("UpdateFlowsBulk")
- return nil
+func (ap *AdapterProxy) UpdateFlowsBulk(device *voltha.Device, flows *voltha.Flows, groups *voltha.FlowGroups) error {
+ log.Debugw("UpdateFlowsBulk", log.Fields{"deviceId": device.Id})
+ topic := kafka.Topic{Name: device.Type}
+ rpc := "update_flows_bulk"
+ args := make([]*kafka.KVArg, 3)
+ args[0] = &kafka.KVArg{
+ Key: "device",
+ Value: device,
+ }
+ args[1] = &kafka.KVArg{
+ Key: "flows",
+ Value: flows,
+ }
+ args[2] = &kafka.KVArg{
+ Key: "groups",
+ Value: groups,
+ }
+
+ success, result := ap.kafkaProxy.InvokeRPC(nil, rpc, &topic, true, args...)
+ log.Debugw("UpdateFlowsBulk-response", log.Fields{"deviceid": device.Id, "success": success})
+ return unPackResponse(rpc, device.Id, success, result)
}
-func (ap *AdapterProxy) UpdateFlowsIncremental(device voltha.Device, flowChanges voltha.Flows, groupChanges voltha.FlowGroups) error {
- log.Debug("UpdateFlowsIncremental")
- return nil
+func (ap *AdapterProxy) UpdateFlowsIncremental(device *voltha.Device, flowChanges *openflow_13.FlowChanges, groupChanges *openflow_13.FlowGroupChanges) error {
+ log.Debugw("UpdateFlowsIncremental", log.Fields{"deviceId": device.Id})
+ topic := kafka.Topic{Name: device.Type}
+ rpc := "update_flows_bulk"
+ args := make([]*kafka.KVArg, 3)
+ args[0] = &kafka.KVArg{
+ Key: "device",
+ Value: device,
+ }
+ args[1] = &kafka.KVArg{
+ Key: "flow_changes",
+ Value: flowChanges,
+ }
+ args[2] = &kafka.KVArg{
+ Key: "group_changes",
+ Value: groupChanges,
+ }
+
+ success, result := ap.kafkaProxy.InvokeRPC(nil, rpc, &topic, true, args...)
+ log.Debugw("UpdateFlowsIncremental-response", log.Fields{"deviceid": device.Id, "success": success})
+ return unPackResponse(rpc, device.Id, success, result)
}
func (ap *AdapterProxy) UpdatePmConfig(device voltha.Device, pmConfigs voltha.PmConfigs) error {
diff --git a/rw_core/core/adapter_request_handler.go b/rw_core/core/adapter_request_handler.go
index 0c0609e..e7f69f5 100644
--- a/rw_core/core/adapter_request_handler.go
+++ b/rw_core/core/adapter_request_handler.go
@@ -17,6 +17,7 @@
import (
"errors"
+ "github.com/gogo/protobuf/proto"
"github.com/golang/protobuf/ptypes"
"github.com/golang/protobuf/ptypes/empty"
"github.com/opencord/voltha-go/common/log"
@@ -25,7 +26,6 @@
"github.com/opencord/voltha-go/protos/voltha"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
- "reflect"
)
type AdapterRequestHandlerProxy struct {
@@ -83,7 +83,7 @@
}
// Get the device via the device manager
- if device, err := rhp.deviceMgr.getDevice(pID.Id); err != nil {
+ if device, err := rhp.deviceMgr.GetDevice(pID.Id); err != nil {
return nil, status.Errorf(codes.NotFound, "%s", err.Error())
} else {
return device, nil
@@ -96,16 +96,17 @@
// First retrieve the most up to date device info
var currentDevice *voltha.Device
var err error
- if currentDevice, err = rhp.deviceMgr.getDevice(device.Id); err != nil {
+ if currentDevice, err = rhp.deviceMgr.GetDevice(device.Id); err != nil {
return nil, err
}
- cloned := reflect.ValueOf(currentDevice).Elem().Interface().(voltha.Device)
+ cloned := proto.Clone(currentDevice).(*voltha.Device)
cloned.Root = device.Root
cloned.Vendor = device.Vendor
cloned.Model = device.Model
cloned.SerialNumber = device.SerialNumber
cloned.MacAddress = device.MacAddress
- return &cloned, nil
+ cloned.Vlan = device.Vlan
+ return cloned, nil
}
func (rhp *AdapterRequestHandlerProxy) DeviceUpdate(args []*ca.Argument) (*empty.Empty, error) {
diff --git a/rw_core/core/device_agent.go b/rw_core/core/device_agent.go
index 52ab584..e045fc9 100644
--- a/rw_core/core/device_agent.go
+++ b/rw_core/core/device_agent.go
@@ -17,16 +17,18 @@
import (
"context"
- "reflect"
- "sync"
-
+ "fmt"
"github.com/gogo/protobuf/proto"
"github.com/opencord/voltha-go/common/log"
"github.com/opencord/voltha-go/db/model"
"github.com/opencord/voltha-go/protos/core_adapter"
+ ofp "github.com/opencord/voltha-go/protos/openflow_13"
"github.com/opencord/voltha-go/protos/voltha"
+ fu "github.com/opencord/voltha-go/rw_core/utils"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
+ "reflect"
+ "sync"
)
type DeviceAgent struct {
@@ -37,6 +39,8 @@
clusterDataProxy *model.Proxy
deviceProxy *model.Proxy
exitChannel chan int
+ flowProxy *model.Proxy
+ groupProxy *model.Proxy
lockDevice sync.RWMutex
}
@@ -48,6 +52,13 @@
cloned := (proto.Clone(device)).(*voltha.Device)
cloned.Id = CreateDeviceId()
cloned.AdminState = voltha.AdminState_PREPROVISIONED
+ cloned.FlowGroups = &ofp.FlowGroups{Items: nil}
+ cloned.Flows = &ofp.Flows{Items: nil}
+ if !device.GetRoot() && device.ProxyAddress != nil {
+ // Set the default vlan ID to the one specified by the parent adapter. It can be
+ // overwritten by the child adapter during a device update request
+ cloned.Vlan = device.ProxyAddress.ChannelId
+ }
agent.deviceId = cloned.Id
agent.lastData = cloned
agent.deviceMgr = deviceMgr
@@ -68,6 +79,17 @@
}
agent.deviceProxy = agent.clusterDataProxy.Root.GetProxy("/devices/"+agent.deviceId, false)
agent.deviceProxy.RegisterCallback(model.POST_UPDATE, agent.processUpdate, nil)
+
+ agent.flowProxy = agent.clusterDataProxy.Root.GetProxy(
+ fmt.Sprintf("/devices/%s/flows", agent.deviceId),
+ false)
+ agent.groupProxy = agent.clusterDataProxy.Root.GetProxy(
+ fmt.Sprintf("/devices/%s/flow_groups", agent.deviceId),
+ false)
+
+ agent.flowProxy.RegisterCallback(model.POST_UPDATE, agent.flowTableUpdated)
+ //agent.groupProxy.RegisterCallback(model.POST_UPDATE, agent.groupTableUpdated)
+
log.Debug("device-agent-started")
}
@@ -80,7 +102,7 @@
log.Debug("device-agent-stopped")
}
-// getDevice retrieves the latest device information from the data model
+// GetDevice retrieves the latest device information from the data model
func (agent *DeviceAgent) getDevice() (*voltha.Device, error) {
agent.lockDevice.Lock()
defer agent.lockDevice.Unlock()
@@ -146,6 +168,51 @@
return nil
}
+func (agent *DeviceAgent) updateFlows(flows []*ofp.OfpFlowStats) error {
+ agent.lockDevice.Lock()
+ defer agent.lockDevice.Unlock()
+ log.Debugw("updateFlows", log.Fields{"deviceId": agent.deviceId, "flows": flows})
+ var oldData *voltha.Flows
+ if storedData, err := agent.getDeviceWithoutLock(); err != nil {
+ return status.Errorf(codes.NotFound, "%s", agent.deviceId)
+ } else {
+ oldData = proto.Clone(storedData.Flows).(*voltha.Flows)
+ log.Debugw("updateFlows", log.Fields{"deviceId": agent.deviceId, "flows": flows, "old": oldData})
+ // store the changed data
+ storedData.Flows.Items = flows
+ afterUpdate := agent.clusterDataProxy.Update("/devices/"+agent.deviceId, storedData, false, "")
+ if afterUpdate == nil {
+ return status.Errorf(codes.Internal, "%s", agent.deviceId)
+ }
+
+ // For now, force the callback to occur
+ go agent.flowTableUpdated(oldData, &ofp.Flows{Items: flows})
+ return nil
+ }
+}
+
+func (agent *DeviceAgent) updateGroups(groups []*ofp.OfpGroupEntry) error {
+ agent.lockDevice.Lock()
+ defer agent.lockDevice.Unlock()
+ var oldData *voltha.FlowGroups
+ log.Debugw("updateGroups", log.Fields{"deviceId": agent.deviceId, "groups": groups})
+ if storedData, err := agent.getDeviceWithoutLock(); err != nil {
+ return status.Errorf(codes.NotFound, "%s", agent.deviceId)
+ } else {
+ oldData = proto.Clone(storedData.FlowGroups).(*voltha.FlowGroups)
+ // store the changed data
+ storedData.FlowGroups.Items = groups
+ afterUpdate := agent.clusterDataProxy.Update("/devices/"+agent.deviceId, storedData, false, "")
+ if afterUpdate == nil {
+ return status.Errorf(codes.Internal, "%s", agent.deviceId)
+ }
+
+ // For now, force the callback to occur
+ go agent.groupTableUpdated(oldData, &ofp.FlowGroups{Items: groups})
+ return nil
+ }
+}
+
//disableDevice disable a device
func (agent *DeviceAgent) disableDevice(ctx context.Context) error {
agent.lockDevice.Lock()
@@ -257,7 +324,7 @@
func (agent *DeviceAgent) getPorts(ctx context.Context, portType voltha.Port_PortType) *voltha.Ports {
log.Debugw("getPorts", log.Fields{"id": agent.deviceId, "portType": portType})
ports := &voltha.Ports{}
- if device, _ := agent.deviceMgr.getDevice(agent.deviceId); device != nil {
+ if device, _ := agent.deviceMgr.GetDevice(agent.deviceId); device != nil {
for _, port := range device.Ports {
if port.Type == portType {
ports.Items = append(ports.Items, port)
@@ -271,7 +338,7 @@
// parent device
func (agent *DeviceAgent) getSwitchCapability(ctx context.Context) (*core_adapter.SwitchCapability, error) {
log.Debugw("getSwitchCapability", log.Fields{"deviceId": agent.deviceId})
- if device, err := agent.deviceMgr.getDevice(agent.deviceId); device == nil {
+ if device, err := agent.deviceMgr.GetDevice(agent.deviceId); device == nil {
return nil, err
} else {
var switchCap *core_adapter.SwitchCapability
@@ -288,7 +355,7 @@
// device
func (agent *DeviceAgent) getPortCapability(ctx context.Context, portNo uint32) (*core_adapter.PortCapability, error) {
log.Debugw("getPortCapability", log.Fields{"deviceId": agent.deviceId})
- if device, err := agent.deviceMgr.getDevice(agent.deviceId); device == nil {
+ if device, err := agent.deviceMgr.GetDevice(agent.deviceId); device == nil {
return nil, err
} else {
var portCap *core_adapter.PortCapability
@@ -500,6 +567,164 @@
}
}
+//flowTableUpdated is the callback after flows have been updated in the model to push them
+//to the adapters
+func (agent *DeviceAgent) flowTableUpdated(args ...interface{}) interface{} {
+ log.Debugw("flowTableUpdated-callback", log.Fields{"argsLen": len(args)})
+
+ agent.lockDevice.Lock()
+ defer agent.lockDevice.Unlock()
+
+ var previousData *voltha.Flows
+ var latestData *voltha.Flows
+
+ var ok bool
+ if previousData, ok = args[0].(*ofp.Flows); !ok {
+ log.Errorw("invalid-args", log.Fields{"args0": args[0]})
+ return nil
+ }
+ if latestData, ok = args[1].(*ofp.Flows); !ok {
+ log.Errorw("invalid-args", log.Fields{"args1": args[1]})
+ return nil
+ }
+
+ // Sanity check - should not happen as this is already handled in logical device agent
+ if reflect.DeepEqual(previousData.Items, latestData.Items) {
+ log.Debugw("flow-update-not-required", log.Fields{"previous": previousData.Items, "new": latestData.Items})
+ return nil
+ }
+
+ var device *voltha.Device
+ var err error
+ if device, err = agent.getDeviceWithoutLock(); err != nil {
+ log.Errorw("no-device", log.Fields{"id": agent.deviceId, "error": err})
+ return nil
+ }
+ groups := device.FlowGroups
+
+ // Send update to adapters
+ // Check whether the device supports incremental flow changes
+ // Assume false for test
+ acceptsAddRemoveFlowUpdates := false
+ if !acceptsAddRemoveFlowUpdates {
+ if err := agent.adapterProxy.UpdateFlowsBulk(device, latestData, groups); err != nil {
+ log.Debugw("update-flow-bulk-error", log.Fields{"id": agent.lastData.Id, "error": err})
+ return err
+ }
+ return nil
+ }
+ // Incremental flow changes accepted
+ var toAdd []*ofp.OfpFlowStats
+ var toDelete []*ofp.OfpFlowStats
+
+ for _, flow := range latestData.Items {
+ if fu.FindFlowById(previousData.Items, flow) == -1 { // did not exist before
+ toAdd = append(toAdd, flow)
+ }
+ }
+ for _, flow := range previousData.Items {
+ if fu.FindFlowById(latestData.Items, flow) == -1 { // does not exist now
+ toDelete = append(toDelete, flow)
+ }
+ }
+ flowChanges := &ofp.FlowChanges{
+ ToAdd: &voltha.Flows{Items: toAdd},
+ ToRemove: &voltha.Flows{Items: toDelete},
+ }
+ // Send an empty group changes as it would be dealt with a call to groupTableUpdated
+ groupChanges := &ofp.FlowGroupChanges{}
+
+ // Send changes only
+ if err := agent.adapterProxy.UpdateFlowsIncremental(device, flowChanges, groupChanges); err != nil {
+ log.Debugw("update-flow-bulk-error", log.Fields{"id": agent.lastData.Id, "error": err})
+ return err
+ }
+
+ return nil
+}
+
+//groupTableUpdated is the callback after group table has been updated in the model to push them
+//to the adapters
+func (agent *DeviceAgent) groupTableUpdated(args ...interface{}) interface{} {
+ log.Debugw("groupTableUpdated-callback", log.Fields{"argsLen": len(args)})
+
+ agent.lockDevice.Lock()
+ defer agent.lockDevice.Unlock()
+
+ var previousData *voltha.FlowGroups
+ var latestData *voltha.FlowGroups
+
+ var ok bool
+ if previousData, ok = args[0].(*ofp.FlowGroups); !ok {
+ log.Errorw("invalid-args", log.Fields{"args0": args[0]})
+ return nil
+ }
+ if latestData, ok = args[1].(*ofp.FlowGroups); !ok {
+ log.Errorw("invalid-args", log.Fields{"args1": args[1]})
+ return nil
+ }
+
+ // Sanity check - should not happen as this is already handled in logical device agent
+ if reflect.DeepEqual(previousData.Items, latestData.Items) {
+ log.Debugw("group-table-update-not-required", log.Fields{"previous": previousData.Items, "new": latestData.Items})
+ return nil
+ }
+
+ var device *voltha.Device
+ var err error
+ if device, err = agent.getDeviceWithoutLock(); err != nil {
+ log.Errorw("no-device", log.Fields{"id": agent.deviceId, "error": err})
+ return nil
+ }
+ flows := device.Flows
+
+ // Send update to adapters
+ // Check whether the device supports incremental flow changes
+ // Assume false for test
+ acceptsAddRemoveFlowUpdates := false
+ if !acceptsAddRemoveFlowUpdates {
+ if err := agent.adapterProxy.UpdateFlowsBulk(device, flows, latestData); err != nil {
+ log.Debugw("update-flows-bulk-error", log.Fields{"id": agent.lastData.Id, "error": err})
+ return err
+ }
+ return nil
+ }
+
+ // Incremental group changes accepted
+ var toAdd []*ofp.OfpGroupEntry
+ var toDelete []*ofp.OfpGroupEntry
+ var toUpdate []*ofp.OfpGroupEntry
+
+ for _, group := range latestData.Items {
+ if idx := fu.FindGroup(previousData.Items, group.Desc.GroupId); idx == -1 { // did not exist before
+ toAdd = append(toAdd, group)
+ } else { // existed before
+ if previousData.Items[idx].String() != group.String() { // there is a change
+ toUpdate = append(toUpdate, group)
+ }
+ }
+ }
+ for _, group := range previousData.Items {
+ if fu.FindGroup(latestData.Items, group.Desc.GroupId) == -1 { // does not exist now
+ toDelete = append(toDelete, group)
+ }
+ }
+ groupChanges := &ofp.FlowGroupChanges{
+ ToAdd: &voltha.FlowGroups{Items: toAdd},
+ ToRemove: &voltha.FlowGroups{Items: toDelete},
+ ToUpdate: &voltha.FlowGroups{Items: toUpdate},
+ }
+ // Send an empty flow changes as it should have been dealt with a call to flowTableUpdated
+ flowChanges := &ofp.FlowChanges{}
+
+ // Send changes only
+ if err := agent.adapterProxy.UpdateFlowsIncremental(device, flowChanges, groupChanges); err != nil {
+ log.Debugw("update-incremental-group-error", log.Fields{"id": agent.lastData.Id, "error": err})
+ return err
+ }
+ return nil
+}
+
// TODO: A generic device update by attribute
func (agent *DeviceAgent) updateDeviceAttribute(name string, value interface{}) {
agent.lockDevice.Lock()
diff --git a/rw_core/core/device_manager.go b/rw_core/core/device_manager.go
index 1063e29..936e945 100644
--- a/rw_core/core/device_manager.go
+++ b/rw_core/core/device_manager.go
@@ -23,6 +23,7 @@
"github.com/opencord/voltha-go/db/model"
"github.com/opencord/voltha-go/kafka"
"github.com/opencord/voltha-go/protos/core_adapter"
+ ofp "github.com/opencord/voltha-go/protos/openflow_13"
"github.com/opencord/voltha-go/protos/voltha"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
@@ -167,14 +168,22 @@
sendResponse(ctx, ch, res)
}
-func (dMgr *DeviceManager) getDevice(id string) (*voltha.Device, error) {
- log.Debugw("getDevice", log.Fields{"deviceid": id})
+func (dMgr *DeviceManager) GetDevice(id string) (*voltha.Device, error) {
+ log.Debugw("GetDevice", log.Fields{"deviceid": id})
if agent := dMgr.getDeviceAgent(id); agent != nil {
return agent.getDevice()
}
return nil, status.Errorf(codes.NotFound, "%s", id)
}
+func (dMgr *DeviceManager) IsRootDevice(id string) (bool, error) {
+ device, err := dMgr.GetDevice(id)
+ if err != nil {
+ return false, err
+ }
+ return device.Root, nil
+}
+
func (dMgr *DeviceManager) ListDevices() (*voltha.Devices, error) {
log.Debug("ListDevices")
result := &voltha.Devices{}
@@ -218,6 +227,21 @@
}
}
+func (dMgr *DeviceManager) updateFlows(deviceId string, flows []*ofp.OfpFlowStats) error {
+ log.Debugw("updateFlows", log.Fields{"deviceid": deviceId})
+ if agent := dMgr.getDeviceAgent(deviceId); agent != nil {
+ return agent.updateFlows(flows)
+ }
+ return status.Errorf(codes.NotFound, "%s", deviceId)
+}
+
+func (dMgr *DeviceManager) updateGroups(deviceId string, groups []*ofp.OfpGroupEntry) error {
+ if agent := dMgr.getDeviceAgent(deviceId); agent != nil {
+ return agent.updateGroups(groups)
+ }
+ return status.Errorf(codes.NotFound, "%s", deviceId)
+}
+
func (dMgr *DeviceManager) updatePmConfigs(deviceId string, pmConfigs *voltha.PmConfigs) error {
if agent := dMgr.getDeviceAgent(deviceId); agent != nil {
return agent.updatePmConfigs(pmConfigs)
@@ -262,7 +286,7 @@
log.Debugw("updateChildrenStatus", log.Fields{"parentDeviceid": deviceId, "operStatus": operStatus, "connStatus": connStatus})
var parentDevice *voltha.Device
var err error
- if parentDevice, err = dMgr.getDevice(deviceId); err != nil {
+ if parentDevice, err = dMgr.GetDevice(deviceId); err != nil {
return status.Errorf(codes.Aborted, "%s", err.Error())
}
var childDeviceIds []string
@@ -356,11 +380,17 @@
return nil
}
-func (dMgr *DeviceManager) deleteLogicalPort(cDevice *voltha.Device) error {
+func (dMgr *DeviceManager) deleteLogicalPort(device *voltha.Device) error {
log.Info("deleteLogicalPort")
var err error
- if err = dMgr.logicalDeviceMgr.deleteLogicalPort(nil, cDevice); err != nil {
- log.Warnw("deleteLogical-port-error", log.Fields{"deviceId": cDevice.Id})
+ // Get the logical port associated with this device
+ var lPortId *voltha.LogicalPortId
+ if lPortId, err = dMgr.logicalDeviceMgr.getLogicalPortId(device); err != nil {
+ log.Warnw("getLogical-port-error", log.Fields{"deviceId": device.Id})
+ return err
+ }
+ if err = dMgr.logicalDeviceMgr.deleteLogicalPort(nil, lPortId); err != nil {
+ log.Warnw("deleteLogical-port-error", log.Fields{"deviceId": device.Id})
return err
}
return nil
@@ -372,7 +402,7 @@
// childDevice is the parent device
return childDevice
}
- parentDevice, _ := dMgr.getDevice(childDevice.ParentId)
+ parentDevice, _ := dMgr.GetDevice(childDevice.ParentId)
return parentDevice
}
@@ -502,7 +532,7 @@
}
func (dMgr *DeviceManager) GetParentDeviceId(deviceId string) *string {
- if device, _ := dMgr.getDevice(deviceId); device != nil {
+ if device, _ := dMgr.GetDevice(deviceId); device != nil {
log.Infow("GetParentDeviceId", log.Fields{"deviceId": device.Id, "parentId": device.ParentId})
return &device.ParentId
}
diff --git a/rw_core/core/grpc_nbi_api_handler.go b/rw_core/core/grpc_nbi_api_handler.go
index e036998..ed30bb7 100644
--- a/rw_core/core/grpc_nbi_api_handler.go
+++ b/rw_core/core/grpc_nbi_api_handler.go
@@ -75,11 +75,6 @@
return out, nil
}
-func processEnableDevicePort(ctx context.Context, id *voltha.LogicalPortId, ch chan interface{}) {
- log.Debugw("processEnableDevicePort", log.Fields{"id": id, "test": common.TestModeKeys_api_test.String()})
- ch <- status.Errorf(100, "%d-%s", 100, "erreur")
-}
-
func (handler *APIHandler) EnableLogicalDevicePort(ctx context.Context, id *voltha.LogicalPortId) (*empty.Empty, error) {
log.Debugw("EnableLogicalDevicePort-request", log.Fields{"id": id, "test": common.TestModeKeys_api_test.String()})
if isTestMode(ctx) {
@@ -88,7 +83,7 @@
}
ch := make(chan interface{})
defer close(ch)
- go processEnableDevicePort(ctx, id, ch)
+ go handler.logicalDeviceMgr.enableLogicalPort(ctx, id, ch)
return waitForNilResponseOnSuccess(ctx, ch)
}
@@ -98,7 +93,10 @@
out := new(empty.Empty)
return out, nil
}
- return nil, errors.New("Unimplemented")
+ ch := make(chan interface{})
+ defer close(ch)
+ go handler.logicalDeviceMgr.disableLogicalPort(ctx, id, ch)
+ return waitForNilResponseOnSuccess(ctx, ch)
}
func (handler *APIHandler) UpdateLogicalDeviceFlowTable(ctx context.Context, flow *openflow_13.FlowTableUpdate) (*empty.Empty, error) {
@@ -107,7 +105,10 @@
out := new(empty.Empty)
return out, nil
}
- return nil, errors.New("Unimplemented")
+ ch := make(chan interface{})
+ defer close(ch)
+ go handler.logicalDeviceMgr.updateFlowTable(ctx, flow.Id, flow.FlowMod, ch)
+ return waitForNilResponseOnSuccess(ctx, ch)
}
func (handler *APIHandler) UpdateLogicalDeviceFlowGroupTable(ctx context.Context, flow *openflow_13.FlowGroupTableUpdate) (*empty.Empty, error) {
@@ -116,13 +117,16 @@
out := new(empty.Empty)
return out, nil
}
- return nil, errors.New("Unimplemented")
+ ch := make(chan interface{})
+ defer close(ch)
+ go handler.logicalDeviceMgr.updateGroupTable(ctx, flow.Id, flow.GroupMod, ch)
+ return waitForNilResponseOnSuccess(ctx, ch)
}
// GetDevice must be implemented in the read-only containers - should it also be implemented here?
func (handler *APIHandler) GetDevice(ctx context.Context, id *voltha.ID) (*voltha.Device, error) {
log.Debugw("GetDevice-request", log.Fields{"id": id})
- return handler.deviceMgr.getDevice(id.Id)
+ return handler.deviceMgr.GetDevice(id.Id)
}
// GetDevice must be implemented in the read-only containers - should it also be implemented here?
@@ -143,6 +147,12 @@
return handler.logicalDeviceMgr.listLogicalDevices()
}
+// ListLogicalDevicePorts must be implemented in the read-only containers - should it also be implemented here?
+func (handler *APIHandler) ListLogicalDevicePorts(ctx context.Context, id *voltha.ID) (*voltha.LogicalPorts, error) {
+ log.Debugw("ListLogicalDevicePorts", log.Fields{"logicaldeviceid": id})
+ return handler.logicalDeviceMgr.ListLogicalDevicePorts(ctx, id.Id)
+}
+
// CreateDevice creates a new parent device in the data model
func (handler *APIHandler) CreateDevice(ctx context.Context, device *voltha.Device) (*voltha.Device, error) {
log.Debugw("createdevice", log.Fields{"device": *device})
diff --git a/rw_core/core/logical_device_agent.go b/rw_core/core/logical_device_agent.go
index 519a0a1..5c9eced 100644
--- a/rw_core/core/logical_device_agent.go
+++ b/rw_core/core/logical_device_agent.go
@@ -17,6 +17,8 @@
import (
"context"
+ "errors"
+ "fmt"
"github.com/gogo/protobuf/proto"
"github.com/opencord/voltha-go/common/log"
"github.com/opencord/voltha-go/db/model"
@@ -28,6 +30,7 @@
fu "github.com/opencord/voltha-go/rw_core/utils"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
+ "reflect"
"sync"
)
@@ -41,7 +44,10 @@
exitChannel chan int
deviceGraph *graph.DeviceGraph
DefaultFlowRules *fu.DeviceRules
+ flowProxy *model.Proxy
+ groupProxy *model.Proxy
lockLogicalDevice sync.RWMutex
+ flowDecomposer *fd.FlowDecomposer
}
func newLogicalDeviceAgent(id string, device *voltha.Device, ldeviceMgr *LogicalDeviceManager, deviceMgr *DeviceManager,
@@ -53,7 +59,7 @@
agent.deviceMgr = deviceMgr
agent.clusterDataProxy = cdProxy
agent.ldeviceMgr = ldeviceMgr
- //agent.deviceGraph =
+ agent.flowDecomposer = fd.NewFlowDecomposer(agent.deviceMgr)
agent.lockLogicalDevice = sync.RWMutex{}
return &agent
}
@@ -71,6 +77,8 @@
ld := &voltha.LogicalDevice{Id: agent.logicalDeviceId, RootDeviceId: agent.rootDeviceId}
ld.Desc = (proto.Clone(switchCap.Desc)).(*ofp.OfpDesc)
ld.SwitchFeatures = (proto.Clone(switchCap.SwitchFeatures)).(*ofp.OfpSwitchFeatures)
+ ld.Flows = &ofp.Flows{Items: nil}
+ ld.FlowGroups = &ofp.FlowGroups{Items: nil}
//Add logical ports to the logical device based on the number of NNI ports discovered
//First get the default port capability - TODO: each NNI port may have different capabilities,
@@ -82,14 +90,18 @@
}
var portCap *ca.PortCapability
for _, port := range nniPorts.Items {
- log.Infow("NNI PORTS", log.Fields{"NNI": port})
+ log.Infow("!!!!!!!NNI PORTS", log.Fields{"NNI": port})
if portCap, err = agent.deviceMgr.getPortCapability(ctx, agent.rootDeviceId, port.PortNo); err != nil {
log.Errorw("error-creating-logical-device", log.Fields{"error": err})
return err
}
-
+ portCap.Port.RootPort = true
lp := (proto.Clone(portCap.Port)).(*voltha.LogicalPort)
lp.DeviceId = agent.rootDeviceId
+ lp.Id = fmt.Sprintf("nni-%d", port.PortNo)
+ lp.OfpPort.PortNo = port.PortNo
+ lp.OfpPort.Name = portCap.Port.Id
+ lp.DevicePortNo = port.PortNo
ld.Ports = append(ld.Ports, lp)
}
agent.lockLogicalDevice.Lock()
@@ -101,6 +113,16 @@
log.Debugw("logicaldevice-created", log.Fields{"logicaldeviceId": agent.logicalDeviceId})
}
+ agent.flowProxy = agent.clusterDataProxy.Root.GetProxy(
+ fmt.Sprintf("/logical_devices/%s/flows", agent.logicalDeviceId),
+ false)
+ agent.groupProxy = agent.clusterDataProxy.Root.GetProxy(
+ fmt.Sprintf("/logical_devices/%s/flow_groups", agent.logicalDeviceId),
+ false)
+
+ agent.flowProxy.RegisterCallback(model.POST_UPDATE, agent.flowTableUpdated)
+ //agent.groupProxy.RegisterCallback(model.POST_UPDATE, agent.groupTableUpdated)
+
return nil
}
@@ -119,9 +141,9 @@
log.Info("logical_device-agent-stopped")
}
-// getLogicalDevice locks the logical device model and then retrieves the latest logical device information
-func (agent *LogicalDeviceAgent) getLogicalDevice() (*voltha.LogicalDevice, error) {
- log.Debug("getLogicalDevice")
+// GetLogicalDevice locks the logical device model and then retrieves the latest logical device information
+func (agent *LogicalDeviceAgent) GetLogicalDevice() (*voltha.LogicalDevice, error) {
+ log.Debug("GetLogicalDevice")
agent.lockLogicalDevice.Lock()
defer agent.lockLogicalDevice.Unlock()
logicalDevice := agent.clusterDataProxy.Get("/logical_devices/"+agent.logicalDeviceId, 1, false, "")
@@ -132,6 +154,45 @@
return nil, status.Errorf(codes.NotFound, "logical_device-%s", agent.logicalDeviceId)
}
+func (agent *LogicalDeviceAgent) ListLogicalDevicePorts() (*voltha.LogicalPorts, error) {
+ log.Debug("!!!!!ListLogicalDevicePorts")
+ agent.lockLogicalDevice.Lock()
+ defer agent.lockLogicalDevice.Unlock()
+ logicalDevice := agent.clusterDataProxy.Get("/logical_devices/"+agent.logicalDeviceId, 1, false, "")
+ if lDevice, ok := logicalDevice.(*voltha.LogicalDevice); ok {
+ lPorts := make([]*voltha.LogicalPort, 0)
+ for _, port := range lDevice.Ports {
+ lPorts = append(lPorts, proto.Clone(port).(*voltha.LogicalPort))
+ }
+ return &voltha.LogicalPorts{Items: lPorts}, nil
+ }
+ return nil, status.Errorf(codes.NotFound, "logical_device-%s", agent.logicalDeviceId)
+}
+
+// listFlows locks the logical device model and then retrieves the latest flow information
+func (agent *LogicalDeviceAgent) listFlows() []*ofp.OfpFlowStats {
+ log.Debug("listFlows")
+ agent.lockLogicalDevice.Lock()
+ defer agent.lockLogicalDevice.Unlock()
+ logicalDevice := agent.clusterDataProxy.Get("/logical_devices/"+agent.logicalDeviceId, 1, false, "")
+ if lDevice, ok := logicalDevice.(*voltha.LogicalDevice); ok {
+ return lDevice.Flows.Items
+ }
+ return nil
+}
+
+// listFlowGroups locks the logical device model and then retrieves the latest flow groups information
+func (agent *LogicalDeviceAgent) listFlowGroups() []*ofp.OfpGroupEntry {
+ log.Debug("listFlowGroups")
+ agent.lockLogicalDevice.Lock()
+ defer agent.lockLogicalDevice.Unlock()
+ logicalDevice := agent.clusterDataProxy.Get("/logical_devices/"+agent.logicalDeviceId, 1, false, "")
+ if lDevice, ok := logicalDevice.(*voltha.LogicalDevice); ok {
+ return lDevice.FlowGroups.Items
+ }
+ return nil
+}
+
// getLogicalDeviceWithoutLock retrieves a logical device from the model without locking it. This is used only by
// functions that have already acquired the logical device lock to the model
func (agent *LogicalDeviceAgent) getLogicalDeviceWithoutLock() (*voltha.LogicalDevice, error) {
@@ -145,12 +206,20 @@
}
// addUNILogicalPort creates a UNI port on the logical device that represents a child device
-func (agent *LogicalDeviceAgent) addUNILogicalPort(ctx context.Context, childDevice *voltha.Device, portNo uint32) error {
+func (agent *LogicalDeviceAgent) addUNILogicalPort(ctx context.Context, childDevice *voltha.Device) error {
log.Infow("addUNILogicalPort-start", log.Fields{"logicalDeviceId": agent.logicalDeviceId})
// Build the logical device based on information retrieved from the device adapter
var portCap *ca.PortCapability
var err error
- if portCap, err = agent.deviceMgr.getPortCapability(ctx, childDevice.Id, portNo); err != nil {
+
+ //Get UNI port number
+ var uniPort uint32
+ for _, port := range childDevice.Ports {
+ if port.Type == voltha.Port_ETHERNET_UNI {
+ uniPort = port.PortNo
+ }
+ }
+ if portCap, err = agent.deviceMgr.getPortCapability(ctx, childDevice.Id, uniPort); err != nil {
log.Errorw("error-creating-logical-port", log.Fields{"error": err})
return err
}
@@ -160,7 +229,16 @@
if ldevice, err := agent.getLogicalDeviceWithoutLock(); err != nil {
return status.Error(codes.NotFound, agent.logicalDeviceId)
} else {
+ log.Infow("!!!!!!!!!!!ADDING-UNI", log.Fields{"deviceId": childDevice.Id})
cloned := proto.Clone(ldevice).(*voltha.LogicalDevice)
+ portCap.Port.RootPort = false
+ //TODO: For now use the channel id assigned by the OLT as logical port number
+ lPortNo := childDevice.ProxyAddress.ChannelId
+ portCap.Port.Id = fmt.Sprintf("uni-%d", lPortNo)
+ portCap.Port.OfpPort.PortNo = lPortNo
+ portCap.Port.OfpPort.Name = portCap.Port.Id
+ portCap.Port.DeviceId = childDevice.Id
+ portCap.Port.DevicePortNo = uniPort
lp := proto.Clone(portCap.Port).(*voltha.LogicalPort)
lp.DeviceId = childDevice.Id
cloned.Ports = append(cloned.Ports, lp)
@@ -178,19 +256,382 @@
return nil
}
-// deleteLogicalPort removes the logical port associated with a child device
-func (agent *LogicalDeviceAgent) deleteLogicalPort(device *voltha.Device) error {
+//updateFlowTable updates the flow table of that logical device
+func (agent *LogicalDeviceAgent) updateFlowTable(ctx context.Context, flow *ofp.OfpFlowMod) error {
+ log.Debug("updateFlowTable")
+ if flow == nil {
+ return nil
+ }
+ switch flow.GetCommand() {
+ case ofp.OfpFlowModCommand_OFPFC_ADD:
+ return agent.flowAdd(flow)
+ case ofp.OfpFlowModCommand_OFPFC_DELETE:
+ return agent.flowDelete(flow)
+ case ofp.OfpFlowModCommand_OFPFC_DELETE_STRICT:
+ return agent.flowDeleteStrict(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())
+}
+
+//updateGroupTable updates the group table of that logical device
+func (agent *LogicalDeviceAgent) updateGroupTable(ctx context.Context, groupMod *ofp.OfpGroupMod) error {
+ log.Debug("updateGroupTable")
+ if groupMod == nil {
+ return nil
+ }
+ switch groupMod.GetCommand() {
+ case ofp.OfpGroupModCommand_OFPGC_ADD:
+ return agent.groupAdd(groupMod)
+ case ofp.OfpGroupModCommand_OFPGC_DELETE:
+ return agent.groupDelete(groupMod)
+ case ofp.OfpGroupModCommand_OFPGC_MODIFY:
+ return agent.groupModify(groupMod)
+ }
+ return status.Errorf(codes.Internal,
+ "unhandled-command: lDeviceId:%s, command:%s", agent.logicalDeviceId, groupMod.GetCommand())
+}
+
+//updateFlowsWithoutLock updates the flows in the logical device without locking the logical device. This function
+//must only be called by a function that is holding the lock on the logical device
+func (agent *LogicalDeviceAgent) updateFlowsWithoutLock(flows []*ofp.OfpFlowStats) error {
+ if ldevice, err := agent.getLogicalDeviceWithoutLock(); err != nil {
+ return status.Error(codes.NotFound, agent.logicalDeviceId)
+ } else {
+ flowsCloned := make([]*ofp.OfpFlowStats, len(flows))
+ copy(flowsCloned, flows)
+ ldevice.Flows.Items = flowsCloned
+ return agent.updateLogicalDeviceWithoutLock(ldevice)
+ }
+}
+
+//updateFlowGroupsWithoutLock updates the flows in the logical device without locking the logical device. This function
+//must only be called by a function that is holding the lock on the logical device
+func (agent *LogicalDeviceAgent) updateFlowGroupsWithoutLock(groups []*ofp.OfpGroupEntry) error {
+ if ldevice, err := agent.getLogicalDeviceWithoutLock(); err != nil {
+ return status.Error(codes.NotFound, agent.logicalDeviceId)
+ } else {
+ groupsCloned := make([]*ofp.OfpGroupEntry, len(groups))
+ copy(groupsCloned, groups)
+ ldevice.FlowGroups.Items = groupsCloned
+ return agent.updateLogicalDeviceWithoutLock(ldevice)
+ }
+}
+
+//flowAdd adds a flow to the flow table of that logical device
+func (agent *LogicalDeviceAgent) flowAdd(mod *ofp.OfpFlowMod) error {
+ log.Debug("flowAdd")
+ if mod == nil {
+ return nil
+ }
agent.lockLogicalDevice.Lock()
defer agent.lockLogicalDevice.Unlock()
+
+ var lDevice *voltha.LogicalDevice
+ var err error
+ if lDevice, err = agent.getLogicalDeviceWithoutLock(); err != nil {
+ log.Errorw("no-logical-device-present", log.Fields{"logicalDeviceId": agent.logicalDeviceId})
+ return errors.New(fmt.Sprintf("no-logical-device-present:%s", agent.logicalDeviceId))
+ }
+
+ var flows []*ofp.OfpFlowStats
+ if lDevice.Flows != nil && lDevice.Flows.Items != nil {
+ flows = lDevice.Flows.Items
+ }
+
+ oldData := proto.Clone(lDevice.Flows).(*voltha.Flows)
+ changed := false
+ 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?
+ log.Warnw("overlapped-flows", log.Fields{"logicaldeviceId": agent.logicalDeviceId})
+ } else {
+ // Add flow
+ flow := fd.FlowStatsEntryFromFlowModMessage(mod)
+ flows = append(flows, flow)
+ changed = true
+ }
+ } else {
+ flow := fd.FlowStatsEntryFromFlowModMessage(mod)
+ idx := fu.FindFlows(flows, flow)
+ if idx >= 0 {
+ oldFlow := flows[idx]
+ if (mod.Flags & uint32(ofp.OfpFlowModFlags_OFPFF_RESET_COUNTS)) != 0 {
+ flow.ByteCount = oldFlow.ByteCount
+ flow.PacketCount = oldFlow.PacketCount
+ }
+ flows[idx] = flow
+ } else {
+ flows = append(flows, flow)
+ }
+ changed = true
+ }
+ if changed {
+ // Update model
+ if lDevice.Flows == nil {
+ lDevice.Flows = &ofp.Flows{}
+ }
+ lDevice.Flows.Items = flows
+ if err := agent.updateLogicalDeviceWithoutLock(lDevice); err != nil {
+ log.Errorw("Cannot-update-logical-group", log.Fields{"logicalDeviceId": agent.logicalDeviceId})
+ return err
+ }
+ }
+ // For now, force the callback to occur
+ go agent.flowTableUpdated(oldData, lDevice.Flows)
+ return nil
+}
+
+//flowDelete deletes a flow from the flow table of that logical device
+func (agent *LogicalDeviceAgent) flowDelete(mod *ofp.OfpFlowMod) error {
+ log.Debug("flowDelete")
+ if mod == nil {
+ return nil
+ }
+ agent.lockLogicalDevice.Lock()
+ defer agent.lockLogicalDevice.Unlock()
+
+ var lDevice *voltha.LogicalDevice
+ var err error
+ if lDevice, err = agent.getLogicalDeviceWithoutLock(); err != nil {
+ log.Errorw("no-logical-device-present", log.Fields{"logicalDeviceId": agent.logicalDeviceId})
+ return errors.New(fmt.Sprintf("no-logical-device-present:%s", agent.logicalDeviceId))
+ }
+ flows := lDevice.Flows.Items
+
+ //build a list of what to keep vs what to delete
+ toKeep := make([]*ofp.OfpFlowStats, 0)
+ for _, f := range flows {
+ if !fu.FlowMatchesMod(f, mod) {
+ toKeep = append(toKeep, f)
+ }
+ }
+
+ //Update flows
+ if len(toKeep) < len(flows) {
+ lDevice.Flows.Items = toKeep
+ if err := agent.updateLogicalDeviceWithoutLock(lDevice); err != nil {
+ log.Errorw("Cannot-update-logical-group", log.Fields{"logicalDeviceId": agent.logicalDeviceId})
+ return err
+ }
+ }
+
+ //TODO: send announcement on delete
+ return nil
+}
+
+//flowStatsDelete deletes a flow from the flow table of that logical device
+func (agent *LogicalDeviceAgent) flowStatsDelete(flow *ofp.OfpFlowStats) error {
+ log.Debug("flowStatsDelete")
+ if flow == nil {
+ return nil
+ }
+ agent.lockLogicalDevice.Lock()
+ defer agent.lockLogicalDevice.Unlock()
+
+ var lDevice *voltha.LogicalDevice
+ var err error
+ if lDevice, err = agent.getLogicalDeviceWithoutLock(); err != nil {
+ log.Errorw("no-logical-device-present", log.Fields{"logicalDeviceId": agent.logicalDeviceId})
+ return errors.New(fmt.Sprintf("no-logical-device-present:%s", agent.logicalDeviceId))
+ }
+ flows := lDevice.Flows.Items
+
+ //build a list of what to keep vs what to delete
+ toKeep := make([]*ofp.OfpFlowStats, 0)
+ for _, f := range flows {
+ if !fu.FlowMatch(f, flow) {
+ toKeep = append(toKeep, f)
+ }
+ }
+
+ //Update flows
+ if len(toKeep) < len(flows) {
+ lDevice.Flows.Items = toKeep
+ if err := agent.updateLogicalDeviceWithoutLock(lDevice); err != nil {
+ log.Errorw("Cannot-update-logical-group", log.Fields{"logicalDeviceId": agent.logicalDeviceId})
+ return err
+ }
+ }
+ return nil
+}
+
+//flowDeleteStrict deletes a flow from the flow table of that logical device
+func (agent *LogicalDeviceAgent) flowDeleteStrict(mod *ofp.OfpFlowMod) error {
+ log.Debug("flowDeleteStrict")
+ if mod == nil {
+ return nil
+ }
+ agent.lockLogicalDevice.Lock()
+ defer agent.lockLogicalDevice.Unlock()
+
+ var lDevice *voltha.LogicalDevice
+ var err error
+ if lDevice, err = agent.getLogicalDeviceWithoutLock(); err != nil {
+ log.Errorw("no-logical-device-present", log.Fields{"logicalDeviceId": agent.logicalDeviceId})
+ return errors.New(fmt.Sprintf("no-logical-device-present:%s", agent.logicalDeviceId))
+ }
+ flows := lDevice.Flows.Items
+ changed := false
+ flow := fd.FlowStatsEntryFromFlowModMessage(mod)
+ idx := fu.FindFlows(flows, flow)
+ if idx >= 0 {
+ flows = append(flows[:idx], flows[idx+1:]...)
+ changed = true
+ } else {
+ return errors.New(fmt.Sprintf("Cannot delete flow - %s", flow))
+ }
+
+ if changed {
+ lDevice.Flows.Items = flows
+ if err := agent.updateLogicalDeviceWithoutLock(lDevice); err != nil {
+ log.Errorw("Cannot-update-logical-group", log.Fields{"logicalDeviceId": agent.logicalDeviceId})
+ return err
+ }
+ }
+
+ return nil
+}
+
+//flowModify modifies a flow from the flow table of that logical device
+func (agent *LogicalDeviceAgent) 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 *LogicalDeviceAgent) flowModifyStrict(mod *ofp.OfpFlowMod) error {
+ return errors.New("flowModifyStrict not implemented")
+}
+
+func (agent *LogicalDeviceAgent) groupAdd(groupMod *ofp.OfpGroupMod) error {
+ log.Debug("groupAdd")
+ if groupMod == nil {
+ return nil
+ }
+ agent.lockLogicalDevice.Lock()
+ defer agent.lockLogicalDevice.Unlock()
+
+ var lDevice *voltha.LogicalDevice
+ var err error
+ if lDevice, err = agent.getLogicalDeviceWithoutLock(); err != nil {
+ log.Errorw("no-logical-device-present", log.Fields{"logicalDeviceId": agent.logicalDeviceId})
+ return errors.New(fmt.Sprintf("no-logical-device-present:%s", agent.logicalDeviceId))
+ }
+ groups := lDevice.FlowGroups.Items
+ oldData := proto.Clone(lDevice.FlowGroups).(*voltha.FlowGroups)
+ if fu.FindGroup(groups, groupMod.GroupId) == -1 {
+ groups = append(groups, fd.GroupEntryFromGroupMod(groupMod))
+ lDevice.FlowGroups.Items = groups
+ if err := agent.updateLogicalDeviceWithoutLock(lDevice); err != nil {
+ log.Errorw("Cannot-update-logical-group", log.Fields{"logicalDeviceId": agent.logicalDeviceId})
+ return err
+ }
+ } else {
+ return errors.New(fmt.Sprintf("Groups %d already present", groupMod.GroupId))
+ }
+ // For now, force the callback to occur
+ go agent.groupTableUpdated(oldData, lDevice.FlowGroups)
+ return nil
+}
+
+func (agent *LogicalDeviceAgent) groupDelete(groupMod *ofp.OfpGroupMod) error {
+ log.Debug("groupDelete")
+ if groupMod == nil {
+ return nil
+ }
+ agent.lockLogicalDevice.Lock()
+ defer agent.lockLogicalDevice.Unlock()
+
+ var lDevice *voltha.LogicalDevice
+ var err error
+ if lDevice, err = agent.getLogicalDeviceWithoutLock(); err != nil {
+ log.Errorw("no-logical-device-present", log.Fields{"logicalDeviceId": agent.logicalDeviceId})
+ return errors.New(fmt.Sprintf("no-logical-device-present:%s", agent.logicalDeviceId))
+ }
+ groups := lDevice.FlowGroups.Items
+ flows := lDevice.Flows.Items
+ groupsChanged := false
+ flowsChanged := false
+ groupId := groupMod.GroupId
+ if groupId == uint32(ofp.OfpGroup_OFPG_ALL) {
+ //TODO we must delete all flows that point to this group and
+ //signal controller as requested by flow's flag
+ groups = []*ofp.OfpGroupEntry{}
+ groupsChanged = true
+ } else {
+ if idx := fu.FindGroup(groups, groupId); idx == -1 {
+ return nil // Valid case
+ } else {
+ flowsChanged, flows = fu.FlowsDeleteByGroupId(flows, groupId)
+ groups = append(groups[:idx], groups[idx+1:]...)
+ groupsChanged = true
+ }
+ }
+ if groupsChanged || flowsChanged {
+ lDevice.FlowGroups.Items = groups
+ lDevice.Flows.Items = flows
+ if err := agent.updateLogicalDeviceWithoutLock(lDevice); err != nil {
+ log.Errorw("Cannot-update-logical-group", log.Fields{"logicalDeviceId": agent.logicalDeviceId})
+ return err
+ }
+ }
+ return nil
+}
+
+func (agent *LogicalDeviceAgent) groupModify(groupMod *ofp.OfpGroupMod) error {
+ log.Debug("groupModify")
+ if groupMod == nil {
+ return nil
+ }
+ agent.lockLogicalDevice.Lock()
+ defer agent.lockLogicalDevice.Unlock()
+
+ var lDevice *voltha.LogicalDevice
+ var err error
+ if lDevice, err = agent.getLogicalDeviceWithoutLock(); err != nil {
+ log.Errorw("no-logical-device-present", log.Fields{"logicalDeviceId": agent.logicalDeviceId})
+ return errors.New(fmt.Sprintf("no-logical-device-present:%s", agent.logicalDeviceId))
+ }
+ groups := lDevice.FlowGroups.Items
+ groupsChanged := false
+ groupId := groupMod.GroupId
+ if idx := fu.FindGroup(groups, groupId); idx == -1 {
+ return errors.New(fmt.Sprintf("group-absent:%s", groupId))
+ } else {
+ //replace existing group entry with new group definition
+ groupEntry := fd.GroupEntryFromGroupMod(groupMod)
+ groups[idx] = groupEntry
+ groupsChanged = true
+ }
+ if groupsChanged {
+ lDevice.FlowGroups.Items = groups
+ if err := agent.updateLogicalDeviceWithoutLock(lDevice); err != nil {
+ log.Errorw("Cannot-update-logical-group", log.Fields{"logicalDeviceId": agent.logicalDeviceId})
+ return err
+ }
+ }
+ return nil
+}
+
+// deleteLogicalPort removes the logical port
+func (agent *LogicalDeviceAgent) deleteLogicalPort(lPort *voltha.LogicalPort) error {
+ agent.lockLogicalDevice.Lock()
+ defer agent.lockLogicalDevice.Unlock()
+
// Get the most up to date logical device
var logicaldevice *voltha.LogicalDevice
if logicaldevice, _ = agent.getLogicalDeviceWithoutLock(); logicaldevice == nil {
- log.Debugw("no-logical-device", log.Fields{"logicalDeviceId": agent.logicalDeviceId, "deviceId": device.Id})
+ log.Debugw("no-logical-device", log.Fields{"logicalDeviceId": agent.logicalDeviceId, "logicalPortId": lPort.Id})
return nil
}
index := -1
for i, logicalPort := range logicaldevice.Ports {
- if logicalPort.DeviceId == device.Id {
+ if logicalPort.Id == lPort.Id {
index = i
break
}
@@ -205,6 +646,58 @@
return nil
}
+// enableLogicalPort enables the logical port
+func (agent *LogicalDeviceAgent) enableLogicalPort(lPort *voltha.LogicalPort) error {
+ agent.lockLogicalDevice.Lock()
+ defer agent.lockLogicalDevice.Unlock()
+
+ // Get the most up to date logical device
+ var logicaldevice *voltha.LogicalDevice
+ if logicaldevice, _ = agent.getLogicalDeviceWithoutLock(); logicaldevice == nil {
+ log.Debugw("no-logical-device", log.Fields{"logicalDeviceId": agent.logicalDeviceId, "logicalPortId": lPort.Id})
+ return nil
+ }
+ index := -1
+ for i, logicalPort := range logicaldevice.Ports {
+ if logicalPort.Id == lPort.Id {
+ index = i
+ break
+ }
+ }
+ if index >= 0 {
+ logicaldevice.Ports[index].OfpPort.Config = logicaldevice.Ports[index].OfpPort.Config & ^uint32(ofp.OfpPortConfig_OFPPC_PORT_DOWN)
+ return agent.updateLogicalDeviceWithoutLock(logicaldevice)
+ }
+ //TODO: Trigger subsequent actions on the device
+ return nil
+}
+
+// disableLogicalPort disabled the logical port
+func (agent *LogicalDeviceAgent) disableLogicalPort(lPort *voltha.LogicalPort) error {
+ agent.lockLogicalDevice.Lock()
+ defer agent.lockLogicalDevice.Unlock()
+
+ // Get the most up to date logical device
+ var logicaldevice *voltha.LogicalDevice
+ if logicaldevice, _ = agent.getLogicalDeviceWithoutLock(); logicaldevice == nil {
+ log.Debugw("no-logical-device", log.Fields{"logicalDeviceId": agent.logicalDeviceId, "logicalPortId": lPort.Id})
+ return nil
+ }
+ index := -1
+ for i, logicalPort := range logicaldevice.Ports {
+ if logicalPort.Id == lPort.Id {
+ index = i
+ break
+ }
+ }
+ if index >= 0 {
+ logicaldevice.Ports[index].OfpPort.Config = (logicaldevice.Ports[index].OfpPort.Config & ^uint32(ofp.OfpPortConfig_OFPPC_PORT_DOWN)) | uint32(ofp.OfpPortConfig_OFPPC_PORT_DOWN)
+ return agent.updateLogicalDeviceWithoutLock(logicaldevice)
+ }
+ //TODO: Trigger subsequent actions on the device
+ return nil
+}
+
func isNNIPort(portNo uint32, nniPortsNo []uint32) bool {
for _, pNo := range nniPortsNo {
if pNo == portNo {
@@ -215,7 +708,9 @@
}
func (agent *LogicalDeviceAgent) getPreCalculatedRoute(ingress, egress uint32) []graph.RouteHop {
+ log.Debugw("ROUTE", log.Fields{"len": len(agent.deviceGraph.Routes)})
for routeLink, route := range agent.deviceGraph.Routes {
+ log.Debugw("ROUTELINKS", log.Fields{"ingress": ingress, "egress": egress, "routelink": routeLink})
if ingress == routeLink.Ingress && egress == routeLink.Egress {
return route
}
@@ -224,9 +719,7 @@
return nil
}
-func (agent *LogicalDeviceAgent) GetRoute(ingressPortNo *uint32, egressPortNo *uint32) []graph.RouteHop {
- agent.lockLogicalDevice.Lock()
- defer agent.lockLogicalDevice.Unlock()
+func (agent *LogicalDeviceAgent) GetRoute(ingressPortNo uint32, egressPortNo uint32) []graph.RouteHop {
log.Debugw("getting-route", log.Fields{"ingress-port": ingressPortNo, "egress-port": egressPortNo})
// Get the updated logical device
var ld *ca.LogicalDevice
@@ -245,10 +738,12 @@
log.Errorw("no-nni-ports", log.Fields{"LogicalDeviceId": ld.Id})
return nil
}
+ // Note: A port value of 0 is equivalent to a nil port
+
// Consider different possibilities
- if egressPortNo != nil && ((*egressPortNo & 0x7fffffff) == uint32(ofp.OfpPortNo_OFPP_CONTROLLER)) {
+ if egressPortNo != 0 && ((egressPortNo & 0x7fffffff) == uint32(ofp.OfpPortNo_OFPP_CONTROLLER)) {
log.Debugw("controller-flow", log.Fields{"ingressPortNo": ingressPortNo, "egressPortNo": egressPortNo, "nniPortsNo": nniLogicalPortsNo})
- if isNNIPort(*ingressPortNo, nniLogicalPortsNo) {
+ if isNNIPort(ingressPortNo, nniLogicalPortsNo) {
log.Debug("returning-half-route")
//This is a trap on the NNI Port
//Return a 'half' route to make the flow decomposer logic happy
@@ -263,13 +758,13 @@
return nil
}
//treat it as if the output port is the first NNI of the OLT
- egressPortNo = &nniLogicalPortsNo[0]
+ egressPortNo = nniLogicalPortsNo[0]
}
//If ingress port is not specified (nil), it may be a wildcarded
//route if egress port is OFPP_CONTROLLER or a nni logical port,
//in which case we need to create a half-route where only the egress
//hop is filled, the first hop is nil
- if ingressPortNo == nil && isNNIPort(*egressPortNo, nniLogicalPortsNo) {
+ if ingressPortNo == 0 && isNNIPort(egressPortNo, nniLogicalPortsNo) {
// We can use the 2nd hop of any upstream route, so just find the first upstream:
for routeLink, route := range agent.deviceGraph.Routes {
if isNNIPort(routeLink.Egress, nniLogicalPortsNo) {
@@ -282,9 +777,9 @@
return nil
}
//If egress port is not specified (nil), we can also can return a "half" route
- if egressPortNo == nil {
+ if egressPortNo == 0 {
for routeLink, route := range agent.deviceGraph.Routes {
- if routeLink.Ingress == *ingressPortNo {
+ if routeLink.Ingress == ingressPortNo {
routes = append(routes, route[0])
routes = append(routes, graph.RouteHop{})
return routes
@@ -295,14 +790,14 @@
}
// Return the pre-calculated route
- return agent.getPreCalculatedRoute(*ingressPortNo, *egressPortNo)
+ return agent.getPreCalculatedRoute(ingressPortNo, egressPortNo)
}
// updateRoutes updates the device routes whenever there is a device or port changes relevant to this
// logical device. TODO: Add more heuristics to this process to update the routes where a change has occurred
// instead of rebuilding the entire set of routes
func (agent *LogicalDeviceAgent) updateRoutes() {
- if ld, err := agent.getLogicalDevice(); err == nil {
+ if ld, err := agent.GetLogicalDevice(); err == nil {
agent.deviceGraph.ComputeRoutes(ld.Ports)
}
}
@@ -315,7 +810,7 @@
fg := fu.NewFlowsAndGroups()
var device *voltha.Device
var err error
- if device, err = agent.deviceMgr.getDevice(deviceId); err != nil {
+ if device, err = agent.deviceMgr.GetDevice(deviceId); err != nil {
return fg
}
//set the upstream and downstream ports
@@ -343,6 +838,7 @@
},
Actions: []*ofp.OfpAction{
fd.SetField(fd.VlanVid(uint32(ofp.OfpVlanId_OFPVID_PRESENT) | device.Vlan)),
+ fd.Output(upstreamPorts[0].PortNo),
},
}
fg.AddFlow(fd.MkFlowStat(fa))
@@ -381,7 +877,7 @@
rules := fu.NewDeviceRules()
var ld *voltha.LogicalDevice
var err error
- if ld, err = agent.getLogicalDevice(); err != nil {
+ if ld, err = agent.GetLogicalDevice(); err != nil {
log.Warnw("no-logical-device", log.Fields{"logicaldeviceId": agent.logicalDeviceId})
return rules
}
@@ -401,11 +897,11 @@
// Get latest
var lDevice *voltha.LogicalDevice
var err error
- if lDevice, err = agent.getLogicalDevice(); err != nil {
+ if lDevice, err = agent.GetLogicalDevice(); err != nil {
return fu.NewDeviceRules()
}
if agent.DefaultFlowRules == nil { // Nothing setup yet
- agent.deviceGraph = graph.NewDeviceGraph(agent.deviceMgr.getDevice)
+ agent.deviceGraph = graph.NewDeviceGraph(agent.deviceMgr.GetDevice)
agent.deviceGraph.ComputeRoutes(lDevice.Ports)
agent.DefaultFlowRules = agent.generateDefaultRules()
}
@@ -418,7 +914,7 @@
if len(excludePort) == 1 {
exclPort = excludePort[0]
}
- if lDevice, _ := agent.getLogicalDevice(); lDevice != nil {
+ if lDevice, _ := agent.GetLogicalDevice(); lDevice != nil {
for _, port := range lDevice.Ports {
if port.OfpPort.PortNo != exclPort {
lPorts = append(lPorts, port.OfpPort.PortNo)
@@ -427,3 +923,103 @@
}
return lPorts
}
+
+func (agent *LogicalDeviceAgent) GetDeviceGraph() *graph.DeviceGraph {
+ return agent.deviceGraph
+}
+
+//setupDeviceGraph creates the device graph if not done already
+func (agent *LogicalDeviceAgent) setupDeviceGraph() {
+ if agent.deviceGraph == nil {
+ agent.deviceGraph = graph.NewDeviceGraph(agent.deviceMgr.GetDevice)
+ agent.updateRoutes()
+ }
+}
+
+func (agent *LogicalDeviceAgent) flowTableUpdated(args ...interface{}) interface{} {
+ log.Debugw("flowTableUpdated-callback", log.Fields{"argsLen": len(args)})
+
+ //agent.lockLogicalDevice.Lock()
+ //defer agent.lockLogicalDevice.Unlock()
+
+ var previousData *ofp.Flows
+ var latestData *ofp.Flows
+
+ var ok bool
+ if previousData, ok = args[0].(*ofp.Flows); !ok {
+ log.Errorw("invalid-args", log.Fields{"args0": args[0]})
+ }
+ if latestData, ok = args[1].(*ofp.Flows); !ok {
+ log.Errorw("invalid-args", log.Fields{"args1": args[1]})
+ }
+
+ if reflect.DeepEqual(previousData.Items, latestData.Items) {
+ log.Debug("flow-update-not-required")
+ return nil
+ }
+
+ // Ensure the device graph has been setup
+ agent.setupDeviceGraph()
+
+ var groups *ofp.FlowGroups
+ lDevice, _ := agent.getLogicalDeviceWithoutLock()
+ groups = lDevice.FlowGroups
+ log.Debugw("flowsinfo", log.Fields{"flows": latestData, "groups": groups})
+ //groupsIf := agent.groupProxy.Get("/", 1, false, "")
+ //if groups, ok = groupsIf.(*ofp.FlowGroups); !ok {
+ // log.Errorw("cannot-retrieve-groups", log.Fields{"logicalDeviceId": agent.logicalDeviceId, "group": groupsIf})
+ // //return errors.New("cannot-retrieve-groups")
+ // groups = &ofp.FlowGroups{Items:nil}
+ //}
+ deviceRules := agent.flowDecomposer.DecomposeRules(agent, *latestData, *groups)
+ log.Debugw("rules", log.Fields{"rules": deviceRules.String()})
+ for deviceId, value := range deviceRules.GetRules() {
+ agent.deviceMgr.updateFlows(deviceId, value.ListFlows())
+ agent.deviceMgr.updateGroups(deviceId, value.ListGroups())
+ }
+ return nil
+}
+
+func (agent *LogicalDeviceAgent) groupTableUpdated(args ...interface{}) interface{} {
+ log.Debugw("groupTableUpdated-callback", log.Fields{"argsLen": len(args)})
+
+ //agent.lockLogicalDevice.Lock()
+ //defer agent.lockLogicalDevice.Unlock()
+
+ var previousData *ofp.FlowGroups
+ var latestData *ofp.FlowGroups
+
+ var ok bool
+ if previousData, ok = args[0].(*ofp.FlowGroups); !ok {
+ log.Errorw("invalid-args", log.Fields{"args0": args[0]})
+ }
+ if latestData, ok = args[1].(*ofp.FlowGroups); !ok {
+ log.Errorw("invalid-args", log.Fields{"args1": args[1]})
+ }
+
+ if reflect.DeepEqual(previousData.Items, latestData.Items) {
+ log.Debug("flow-update-not-required")
+ return nil
+ }
+
+ // Ensure the device graph has been setup
+ agent.setupDeviceGraph()
+
+ var flows *ofp.Flows
+ lDevice, _ := agent.getLogicalDeviceWithoutLock()
+ flows = lDevice.Flows
+ log.Debugw("groupsinfo", log.Fields{"groups": latestData, "flows": flows})
+ //flowsIf := agent.flowProxy.Get("/", 1, false, "")
+ //if flows, ok = flowsIf.(*ofp.Flows); !ok {
+ // log.Errorw("cannot-retrieve-flows", log.Fields{"logicalDeviceId": agent.logicalDeviceId, "flows": flows})
+ // //return errors.New("cannot-retrieve-groups")
+ // flows = &ofp.Flows{Items:nil}
+ //}
+ deviceRules := agent.flowDecomposer.DecomposeRules(agent, *flows, *latestData)
+ log.Debugw("rules", log.Fields{"rules": deviceRules.String()})
+ for deviceId, value := range deviceRules.GetRules() {
+ agent.deviceMgr.updateFlows(deviceId, value.ListFlows())
+ agent.deviceMgr.updateGroups(deviceId, value.ListGroups())
+ }
+ return nil
+}
diff --git a/rw_core/core/logical_device_manager.go b/rw_core/core/logical_device_manager.go
index 8f8548a..9d365aa 100644
--- a/rw_core/core/logical_device_manager.go
+++ b/rw_core/core/logical_device_manager.go
@@ -21,6 +21,7 @@
"github.com/opencord/voltha-go/common/log"
"github.com/opencord/voltha-go/db/model"
"github.com/opencord/voltha-go/kafka"
+ "github.com/opencord/voltha-go/protos/openflow_13"
"github.com/opencord/voltha-go/protos/voltha"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
@@ -60,6 +61,18 @@
log.Info("logical-device-manager-stopped")
}
+func sendAPIResponse(ctx context.Context, ch chan interface{}, result interface{}) {
+ if ctx.Err() == nil {
+ // Returned response only of the ctx has not been cancelled/timeout/etc
+ // Channel is automatically closed when a context is Done
+ ch <- result
+ log.Debugw("sendResponse", log.Fields{"result": result})
+ } else {
+ // Should the transaction be reverted back?
+ log.Debugw("sendResponse-context-error", log.Fields{"context-error": ctx.Err()})
+ }
+}
+
func (ldMgr *LogicalDeviceManager) addLogicalDeviceAgentToMap(agent *LogicalDeviceAgent) {
ldMgr.lockLogicalDeviceAgentsMap.Lock()
defer ldMgr.lockLogicalDeviceAgentsMap.Unlock()
@@ -83,11 +96,11 @@
delete(ldMgr.logicalDeviceAgents, logicalDeviceId)
}
-// getLogicalDevice provides a cloned most up to date logical device
+// GetLogicalDevice provides a cloned most up to date logical device
func (ldMgr *LogicalDeviceManager) getLogicalDevice(id string) (*voltha.LogicalDevice, error) {
log.Debugw("getlogicalDevice", log.Fields{"logicaldeviceid": id})
if agent := ldMgr.getLogicalDeviceAgent(id); agent != nil {
- return agent.getLogicalDevice()
+ return agent.GetLogicalDevice()
}
return nil, status.Errorf(codes.NotFound, "%s", id)
}
@@ -98,7 +111,7 @@
ldMgr.lockLogicalDeviceAgentsMap.Lock()
defer ldMgr.lockLogicalDeviceAgentsMap.Unlock()
for _, agent := range ldMgr.logicalDeviceAgents {
- if lDevice, err := agent.getLogicalDevice(); err == nil {
+ if lDevice, err := agent.GetLogicalDevice(); err == nil {
result.Items = append(result.Items, lDevice)
}
}
@@ -164,20 +177,67 @@
return nil, status.Errorf(codes.NotFound, "%s", device.Id)
}
-// DeleteLogicalDevice removes the logical port associated with a child device
-func (ldMgr *LogicalDeviceManager) deleteLogicalPort(ctx context.Context, device *voltha.Device) error {
- log.Debugw("deleting-logical-port", log.Fields{"deviceId": device.Id})
+func (ldMgr *LogicalDeviceManager) getLogicalPortId(device *voltha.Device) (*voltha.LogicalPortId, error) {
+ // Get the logical device where this device is attached
+ var lDeviceId *string
+ var err error
+ if lDeviceId, err = ldMgr.getLogicalDeviceId(device); err != nil {
+ return nil, err
+ }
+ var lDevice *voltha.LogicalDevice
+ if lDevice, err = ldMgr.getLogicalDevice(*lDeviceId); err != nil {
+ return nil, err
+ }
+ // Go over list of ports
+ for _, port := range lDevice.Ports {
+ if port.DeviceId == device.Id {
+ return &voltha.LogicalPortId{Id: *lDeviceId, PortId: port.Id}, nil
+ }
+ }
+ return nil, status.Errorf(codes.NotFound, "%s", device.Id)
+}
+
+func (ldMgr *LogicalDeviceManager) ListLogicalDevicePorts(ctx context.Context, id string) (*voltha.LogicalPorts, error) {
+ log.Debugw("ListLogicalDevicePorts", log.Fields{"logicaldeviceid": id})
+ if agent := ldMgr.getLogicalDeviceAgent(id); agent != nil {
+ return agent.ListLogicalDevicePorts()
+ }
+ return nil, status.Errorf(codes.NotFound, "%s", id)
+
+}
+
+func (ldMgr *LogicalDeviceManager) getLogicalPort(lPortId *voltha.LogicalPortId) (*voltha.LogicalPort, error) {
+ // Get the logical device where this device is attached
+ var err error
+ var lDevice *voltha.LogicalDevice
+ if lDevice, err = ldMgr.getLogicalDevice(lPortId.Id); err != nil {
+ return nil, err
+ }
+ // Go over list of ports
+ for _, port := range lDevice.Ports {
+ if port.Id == lPortId.PortId {
+ return port, nil
+ }
+ }
+ return nil, status.Errorf(codes.NotFound, "%s-$s", lPortId.Id, lPortId.PortId)
+}
+
+// deleteLogicalPort removes the logical port associated with a child device
+func (ldMgr *LogicalDeviceManager) deleteLogicalPort(ctx context.Context, lPortId *voltha.LogicalPortId) error {
+ log.Debugw("deleting-logical-port", log.Fields{"LDeviceId": lPortId.Id})
+ // Get logical port
+ var logicalPort *voltha.LogicalPort
+ var err error
+ if logicalPort, err = ldMgr.getLogicalPort(lPortId); err != nil {
+ log.Debugw("no-logical-device-port-present", log.Fields{"logicalPortId": lPortId.PortId})
+ return err
+ }
// Sanity check
- if device.Root {
+ if logicalPort.RootPort {
return errors.New("Device-root")
}
- logDeviceId, _ := ldMgr.getLogicalDeviceId(device)
- if logDeviceId == nil {
- log.Debugw("no-logical-device-present", log.Fields{"deviceId": device.Id})
- return nil
- }
- if agent := ldMgr.getLogicalDeviceAgent(*logDeviceId); agent != nil {
- agent.deleteLogicalPort(device)
+ if agent := ldMgr.getLogicalDeviceAgent(lPortId.Id); agent != nil {
+ agent.deleteLogicalPort(logicalPort)
}
log.Debug("deleting-logical-port-ends")
@@ -198,7 +258,69 @@
log.Debugw("AddUNILogicalPort", log.Fields{"logDeviceId": logDeviceId, "parentId": parentId})
if agent := ldMgr.getLogicalDeviceAgent(*logDeviceId); agent != nil {
- return agent.addUNILogicalPort(ctx, childDevice, childDevice.ProxyAddress.ChannelId)
+ return agent.addUNILogicalPort(ctx, childDevice)
}
return status.Errorf(codes.NotFound, "%s", childDevice.Id)
}
+
+func (ldMgr *LogicalDeviceManager) updateFlowTable(ctx context.Context, id string, flow *openflow_13.OfpFlowMod, ch chan interface{}) {
+ log.Debugw("updateFlowTable", log.Fields{"logicalDeviceId": id})
+ var res interface{}
+ if agent := ldMgr.getLogicalDeviceAgent(id); agent != nil {
+ res = agent.updateFlowTable(ctx, flow)
+ log.Debugw("updateFlowTable-result", log.Fields{"result": res})
+ } else {
+ res = status.Errorf(codes.NotFound, "%s", id)
+ }
+ sendAPIResponse(ctx, ch, res)
+}
+
+func (ldMgr *LogicalDeviceManager) updateGroupTable(ctx context.Context, id string, groupMod *openflow_13.OfpGroupMod, ch chan interface{}) {
+ log.Debugw("updateGroupTable", log.Fields{"logicalDeviceId": id})
+ var res interface{}
+ if agent := ldMgr.getLogicalDeviceAgent(id); agent != nil {
+ res = agent.updateGroupTable(ctx, groupMod)
+ log.Debugw("updateGroupTable-result", log.Fields{"result": res})
+ } else {
+ res = status.Errorf(codes.NotFound, "%s", id)
+ }
+ sendAPIResponse(ctx, ch, res)
+}
+
+func (ldMgr *LogicalDeviceManager) enableLogicalPort(ctx context.Context, id *voltha.LogicalPortId, ch chan interface{}) {
+ log.Debugw("enableLogicalPort", log.Fields{"logicalDeviceId": id})
+ var res interface{}
+ // Get logical port
+ var logicalPort *voltha.LogicalPort
+ var err error
+ if logicalPort, err = ldMgr.getLogicalPort(id); err != nil {
+ log.Debugw("no-logical-device-port-present", log.Fields{"logicalPortId": id.PortId})
+ res = err
+ }
+ if agent := ldMgr.getLogicalDeviceAgent(id.Id); agent != nil {
+ res = agent.enableLogicalPort(logicalPort)
+ log.Debugw("enableLogicalPort-result", log.Fields{"result": res})
+ } else {
+ res = status.Errorf(codes.NotFound, "%s", id.Id)
+ }
+ sendAPIResponse(ctx, ch, res)
+}
+
+func (ldMgr *LogicalDeviceManager) disableLogicalPort(ctx context.Context, id *voltha.LogicalPortId, ch chan interface{}) {
+ log.Debugw("disableLogicalPort", log.Fields{"logicalDeviceId": id})
+ var res interface{}
+ // Get logical port
+ var logicalPort *voltha.LogicalPort
+ var err error
+ if logicalPort, err = ldMgr.getLogicalPort(id); err != nil {
+ log.Debugw("no-logical-device-port-present", log.Fields{"logicalPortId": id.PortId})
+ res = err
+ }
+ if agent := ldMgr.getLogicalDeviceAgent(id.Id); agent != nil {
+ res = agent.disableLogicalPort(logicalPort)
+ log.Debugw("disableLogicalPort-result", log.Fields{"result": res})
+ } else {
+ res = status.Errorf(codes.NotFound, "%s", id.Id)
+ }
+ sendAPIResponse(ctx, ch, res)
+}