package org.opendaylight.controller.forwardingrulesmanager.consumer.impl;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.EnumSet;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import org.opendaylight.controller.md.sal.common.api.data.DataModification;
import org.opendaylight.controller.sal.common.util.Rpcs;
import org.opendaylight.controller.sal.core.IContainer;
-import org.opendaylight.controller.sal.utils.GlobalConstants;
import org.opendaylight.controller.sal.utils.ServiceHelper;
-import org.opendaylight.controller.sal.utils.Status;
-import org.opendaylight.controller.sal.utils.StatusCode;
-import org.opendaylight.controller.switchmanager.ISwitchManager;
-import org.opendaylight.controller.switchmanager.Switch;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.config.rev130819.Flows;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.config.rev130819.flows.Flow;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.config.rev130819.flows.FlowKey;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.AddFlowInput;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.AddFlowInputBuilder;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.AddFlowOutput;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowAdded;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowRemoved;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowUpdated;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.NodeErrorNotification;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.NodeExperimenterErrorNotification;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.NodeFlow;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.RemoveFlowInputBuilder;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.SalFlowListener;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.SalFlowService;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.SwitchFlowRemoved;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.UpdateFlowInputBuilder;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.flow.update.UpdatedFlowBuilder;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.instruction.list.Instruction;
import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes;
import org.opendaylight.yangtools.yang.binding.DataObject;
import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
import org.opendaylight.yangtools.yang.binding.NotificationListener;
+import org.opendaylight.yangtools.yang.common.RpcError;
import org.opendaylight.yangtools.yang.common.RpcResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class FlowConsumerImpl implements IForwardingRulesManager {
protected static final Logger logger = LoggerFactory.getLogger(FlowConsumerImpl.class);
- private final FlowEventListener flowEventListener = new FlowEventListener();
+ // private final FlowEventListener flowEventListener = new
+ // FlowEventListener();
private Registration<NotificationListener> listener1Reg;
private SalFlowService flowService;
// private FlowDataListener listener;
private boolean inContainerMode; // being used by global instance only
public FlowConsumerImpl() {
- InstanceIdentifier<? extends DataObject> path = InstanceIdentifier.builder(Flows.class).child(Flow.class)
- .toInstance();
+ InstanceIdentifier<? extends DataObject> path = InstanceIdentifier.builder(Flows.class).toInstance();
flowService = FRMConsumerImpl.getProviderSession().getRpcService(SalFlowService.class);
if (null == flowService) {
logger.error("Consumer SAL Service is down or NULL. FRM may not function as intended");
- System.out.println("Consumer SAL Service is down or NULL.");
return;
}
// }
// For switch events
- listener1Reg = FRMConsumerImpl.getNotificationService().registerNotificationListener(flowEventListener);
+ // listener1Reg =
+ // FRMConsumerImpl.getNotificationService().registerNotificationListener(flowEventListener);
if (null == listener1Reg) {
logger.error("Listener to listen on flow data modifcation events");
- System.out.println("Consumer SAL Service is down or NULL.");
return;
}
// addFlowTest();
- System.out.println("-------------------------------------------------------------------");
- allocateCaches();
commitHandler = new FlowDataCommitHandler();
FRMConsumerImpl.getDataProviderService().registerCommitHandler(path, commitHandler);
clusterContainerService = (IClusterContainerServices) ServiceHelper.getGlobalInstance(
IClusterContainerServices.class, this);
- container = (IContainer) ServiceHelper.getGlobalInstance(IContainer.class, this);
+ allocateCaches();
/*
* If we are not the first cluster node to come up, do not initialize
* the static flow entries ordinal
AddFlowInput firstMsg = input1.build();
if (null != flowService) {
- System.out.println(flowService.toString());
- } else {
- System.out.println("ConsumerFlowService is NULL");
+ logger.error("ConsumerFlowService is NULL");
}
@SuppressWarnings("unused")
Future<RpcResult<AddFlowOutput>> result1 = flowService.addFlow(firstMsg);
private void addFlow(InstanceIdentifier<?> path, Flow dataObject) {
AddFlowInputBuilder input = new AddFlowInputBuilder();
- List<Instruction> inst = (dataObject).getInstructions().getInstruction();
+
input.setNode((dataObject).getNode());
input.setPriority((dataObject).getPriority());
input.setMatch((dataObject).getMatch());
input.setCookie((dataObject).getCookie());
input.setInstructions((dataObject).getInstructions());
- dataObject.getMatch().getLayer3Match();
- for (int i = 0; i < inst.size(); i++) {
- System.out.println("i = " + i + inst.get(i).getInstruction().toString());
- System.out.println("i = " + i + inst.get(i).toString());
- }
-
- System.out.println("Instruction list" + (dataObject).getInstructions().getInstruction().toString());
+ input.setBufferId(dataObject.getBufferId());
+ input.setTableId(dataObject.getTableId());
+ input.setOutPort(dataObject.getOutPort());
+ input.setOutGroup(dataObject.getOutGroup());
+ input.setIdleTimeout(dataObject.getIdleTimeout());
+ input.setHardTimeout(dataObject.getHardTimeout());
+ input.setFlowName(dataObject.getFlowName());
+ input.setFlags(dataObject.getFlags());
+ input.setCookieMask(dataObject.getCookieMask());
+ input.setContainerName(dataObject.getContainerName());
+ input.setBarrier(dataObject.isBarrier());
+ input.setInstallHw(dataObject.isInstallHw());
+ input.setStrict(dataObject.isStrict());
// updating the staticflow cache
- Integer ordinal = staticFlowsOrdinal.get(0);
- staticFlowsOrdinal.put(0, ++ordinal);
- staticFlows.put(ordinal, dataObject);
+ /*
+ * Commented out... as in many other places... use of ClusteringServices
+ * is breaking things insufficient time to debug Integer ordinal =
+ * staticFlowsOrdinal.get(0); staticFlowsOrdinal.put(0, ++ordinal);
+ * staticFlows.put(ordinal, dataObject);
+ */
// We send flow to the sounthbound plugin
+
flowService.addFlow(input.build());
- updateLocalDatabase((NodeFlow) dataObject, true);
+
+ /*
+ * Commented out as this will also break due to improper use of
+ * ClusteringServices updateLocalDatabase((NodeFlow) dataObject, true);
+ */
}
/**
private void removeFlow(InstanceIdentifier<?> path, Flow dataObject) {
RemoveFlowInputBuilder input = new RemoveFlowInputBuilder();
- List<Instruction> inst = (dataObject).getInstructions().getInstruction();
input.setNode((dataObject).getNode());
input.setPriority((dataObject).getPriority());
input.setMatch((dataObject).getMatch());
input.setCookie((dataObject).getCookie());
input.setInstructions((dataObject).getInstructions());
- dataObject.getMatch().getLayer3Match();
- for (int i = 0; i < inst.size(); i++) {
- System.out.println("i = " + i + inst.get(i).getInstruction().toString());
- System.out.println("i = " + i + inst.get(i).toString());
- }
-
- System.out.println("Instruction list" + (dataObject).getInstructions().getInstruction().toString());
-
+ input.setBufferId(dataObject.getBufferId());
+ input.setTableId(dataObject.getTableId());
+ input.setOutPort(dataObject.getOutPort());
+ input.setOutGroup(dataObject.getOutGroup());
+ input.setIdleTimeout(dataObject.getIdleTimeout());
+ input.setHardTimeout(dataObject.getHardTimeout());
+ input.setFlowName(dataObject.getFlowName());
+ input.setFlags(dataObject.getFlags());
+ input.setCookieMask(dataObject.getCookieMask());
+ input.setContainerName(dataObject.getContainerName());
+ input.setBarrier(dataObject.isBarrier());
+ input.setInstallHw(dataObject.isInstallHw());
+ input.setStrict(dataObject.isStrict());
// updating the staticflow cache
- Integer ordinal = staticFlowsOrdinal.get(0);
- staticFlowsOrdinal.put(0, ++ordinal);
- staticFlows.put(ordinal, dataObject);
+ /*
+ * Commented out due to problems caused by improper use of
+ * ClusteringServices Integer ordinal = staticFlowsOrdinal.get(0);
+ * staticFlowsOrdinal.put(0, ++ordinal); staticFlows.put(ordinal,
+ * dataObject);
+ */
// We send flow to the sounthbound plugin
flowService.removeFlow(input.build());
- updateLocalDatabase((NodeFlow) dataObject, false);
+
+ /*
+ * Commented out due to problems caused by improper use of
+ * ClusteringServices updateLocalDatabase((NodeFlow) dataObject, false);
+ */
}
/**
UpdateFlowInputBuilder input = new UpdateFlowInputBuilder();
UpdatedFlowBuilder updatedflowbuilder = new UpdatedFlowBuilder();
updatedflowbuilder.fieldsFrom(dataObject);
+ input.setNode(dataObject.getNode());
input.setUpdatedFlow(updatedflowbuilder.build());
// updating the staticflow cache
- Integer ordinal = staticFlowsOrdinal.get(0);
- staticFlowsOrdinal.put(0, ++ordinal);
- staticFlows.put(ordinal, dataObject);
+ /*
+ * Commented out due to problems caused by improper use of
+ * ClusteringServices. Integer ordinal = staticFlowsOrdinal.get(0);
+ * staticFlowsOrdinal.put(0, ++ordinal); staticFlows.put(ordinal,
+ * dataObject);
+ */
// We send flow to the sounthbound plugin
flowService.updateFlow(input.build());
- updateLocalDatabase((NodeFlow) dataObject, true);
+
+ /*
+ * Commented out due to problems caused by improper use of
+ * ClusteringServices. updateLocalDatabase((NodeFlow) dataObject, true);
+ */
}
@SuppressWarnings("unchecked")
private void commitToPlugin(internalTransaction transaction) {
- for (Entry<InstanceIdentifier<?>, Flow> entry : transaction.additions.entrySet()) {
- System.out.println("Coming add cc in FlowDatacommitHandler");
- addFlow(entry.getKey(), entry.getValue());
+ Set<Entry<InstanceIdentifier<?>, DataObject>> createdEntries = transaction.getModification()
+ .getCreatedConfigurationData().entrySet();
+
+ /*
+ * This little dance is because updatedEntries contains both created and
+ * modified entries The reason I created a new HashSet is because the
+ * collections we are returned are immutable.
+ */
+ Set<Entry<InstanceIdentifier<?>, DataObject>> updatedEntries = new HashSet<Entry<InstanceIdentifier<?>, DataObject>>();
+ updatedEntries.addAll(transaction.getModification().getUpdatedConfigurationData().entrySet());
+ updatedEntries.removeAll(createdEntries);
+
+ Set<InstanceIdentifier<?>> removeEntriesInstanceIdentifiers = transaction.getModification()
+ .getRemovedConfigurationData();
+ transaction.getModification().getOriginalConfigurationData();
+ for (Entry<InstanceIdentifier<?>, DataObject> entry : createdEntries) {
+ if (entry.getValue() instanceof Flow) {
+ logger.debug("Coming add cc in FlowDatacommitHandler");
+ Flow flow = (Flow) entry.getValue();
+ boolean status = validate(flow);
+ if (!status) {
+ return;
+ }
+ addFlow(entry.getKey(), (Flow) entry.getValue());
+ }
}
for (@SuppressWarnings("unused")
- Entry<InstanceIdentifier<?>, Flow> entry : transaction.updates.entrySet()) {
- System.out.println("Coming update cc in FlowDatacommitHandler");
- updateFlow(entry.getKey(), entry.getValue());
+ Entry<InstanceIdentifier<?>, DataObject> entry : updatedEntries) {
+ if (entry.getValue() instanceof Flow) {
+ logger.debug("Coming update cc in FlowDatacommitHandler");
+ Flow flow = (Flow) entry.getValue();
+ boolean status = validate(flow);
+ if (!status) {
+ return;
+ }
+ updateFlow(entry.getKey(), (Flow) entry.getValue());
+ }
}
- for (Entry<InstanceIdentifier<?>, Flow> entry : transaction.removals.entrySet()) {
- System.out.println("Coming remove cc in FlowDatacommitHandler");
- removeFlow(entry.getKey(), entry.getValue());
+ for (InstanceIdentifier<?> instanceId : removeEntriesInstanceIdentifiers) {
+ DataObject removeValue = transaction.getModification().getOriginalConfigurationData().get(instanceId);
+ if (removeValue instanceof Flow) {
+ logger.debug("Coming remove cc in FlowDatacommitHandler");
+ Flow flow = (Flow) removeValue;
+ boolean status = validate(flow);
+ if (!status) {
+ return;
+ }
+ removeFlow(instanceId, (Flow) removeValue);
+
+ }
}
}
@Override
public DataCommitTransaction requestCommit(DataModification<InstanceIdentifier<?>, DataObject> modification) {
// We should verify transaction
- System.out.println("Coming in FlowDatacommitHandler");
+ logger.debug("Coming in FlowDatacommitHandler");
internalTransaction transaction = new internalTransaction(modification);
transaction.prepareUpdate();
return transaction;
Set<Entry<InstanceIdentifier<?>, DataObject>> puts = modification.getUpdatedConfigurationData().entrySet();
for (Entry<InstanceIdentifier<?>, DataObject> entry : puts) {
-
- // validating the DataObject
-
- Status status = validate(container, (NodeFlow) entry);
- if (!status.isSuccess()) {
- logger.warn("Invalid Configuration for flow {}. The failure is {}", entry, status.getDescription());
- String error = "Invalid Configuration (" + status.getDescription() + ")";
- logger.error(error);
- return;
- }
- // Presence check
- if (flowEntryExists((NodeFlow) entry)) {
- String error = "Entry with this name on specified table already exists";
- logger.warn("Entry with this name on specified table already exists: {}", entry);
- logger.error(error);
- return;
- }
- if (originalSwView.containsKey(entry)) {
- logger.warn("Operation Rejected: A flow with same match and priority exists on the target node");
- logger.trace("Aborting to install {}", entry);
- continue;
- }
- if (!FRMUtil.validateMatch((NodeFlow) entry)) {
- logger.error("Not a valid Match");
- return;
- }
- if (!FRMUtil.validateInstructions((NodeFlow) entry)) {
- logger.error("Not a valid Instruction");
- return;
- }
- if (entry.getValue() instanceof Flow) {
- Flow flow = (Flow) entry.getValue();
- preparePutEntry(entry.getKey(), flow);
- }
-
}
// removals = modification.getRemovedConfigurationData();
Flow original = originalSwView.get(key);
if (original != null) {
// It is update for us
- System.out.println("Coming update in FlowDatacommitHandler");
updates.put(key, flow);
} else {
// It is addition for us
- System.out.println("Coming add in FlowDatacommitHandler");
additions.put(key, flow);
}
}
commitToPlugin(this);
// We return true if internal transaction is successful.
// return Rpcs.getRpcResult(true, null, Collections.emptySet());
- return Rpcs.getRpcResult(true, null, null);
+ return Rpcs.getRpcResult(true, null, Collections.<RpcError> emptySet());
}
/**
// NOOP - we did not modified any internal state during
// requestCommit phase
// return Rpcs.getRpcResult(true, null, Collections.emptySet());
- return Rpcs.getRpcResult(true, null, null);
+ return Rpcs.getRpcResult(true, null, Collections.<RpcError> emptySet());
}
- public Status validate(IContainer container, NodeFlow dataObject) {
-
- // container validation
- Switch sw = null;
- Node node = null;
- String containerName = (container == null) ? GlobalConstants.DEFAULT.toString() : container.getName();
- ISwitchManager switchManager = (ISwitchManager) ServiceHelper.getInstance(ISwitchManager.class,
- containerName, this);
- // flow Name validation
- if (dataObject.getFlowName() == null || dataObject.getFlowName().trim().isEmpty()
- || !dataObject.getFlowName().matches(NAMEREGEX)) {
- return new Status(StatusCode.BADREQUEST, "Invalid Flow name");
- }
- // Node Validation
- if (dataObject.getNode() == null) {
- return new Status(StatusCode.BADREQUEST, "Node is null");
- }
-
- if (switchManager != null) {
- for (Switch device : switchManager.getNetworkDevices()) {
- node = (Node) device.getNode();
- if (device.getNode().equals(dataObject.getNode())) {
- sw = device;
- break;
- }
- }
- if (sw == null) {
- return new Status(StatusCode.BADREQUEST, String.format("Node %s not found", node));
- }
- } else {
- logger.debug("switchmanager is not set yet");
- }
-
- if (dataObject.getPriority() != null) {
- if (dataObject.getPriority() < 0 || dataObject.getPriority() > 65535) {
- return new Status(StatusCode.BADREQUEST, String.format("priority %s is not in the range 0 - 65535",
- dataObject.getPriority()));
- }
- }
-
- return new Status(StatusCode.SUCCESS);
- }
-
- private boolean flowEntryExists(NodeFlow config) {
+ private boolean flowEntryExists(Flow flow) {
// Flow name has to be unique on per table id basis
for (ConcurrentMap.Entry<FlowKey, Flow> entry : originalSwView.entrySet()) {
- if (entry.getValue().getFlowName().equals(config.getFlowName())
- && entry.getValue().getTableId().equals(config.getTableId())) {
+ if (entry.getValue().getFlowName().equals(flow.getFlowName())
+ && entry.getValue().getTableId().equals(flow.getTableId())) {
return true;
}
}
}
}
- final class FlowEventListener implements SalFlowListener {
-
- List<FlowAdded> addedFlows = new ArrayList<>();
- List<FlowRemoved> removedFlows = new ArrayList<>();
- List<FlowUpdated> updatedFlows = new ArrayList<>();
-
- @Override
- public void onFlowAdded(FlowAdded notification) {
- System.out.println("added flow..........................");
- addedFlows.add(notification);
- }
-
- @Override
- public void onFlowRemoved(FlowRemoved notification) {
- removedFlows.add(notification);
- };
-
- @Override
- public void onFlowUpdated(FlowUpdated notification) {
- updatedFlows.add(notification);
- }
-
- @Override
- public void onSwitchFlowRemoved(SwitchFlowRemoved notification) {
- // TODO
- }
-
- @Override
- public void onNodeErrorNotification(NodeErrorNotification notification) {
- // TODO Auto-generated method stub
-
- }
-
- @Override
- public void onNodeExperimenterErrorNotification(NodeExperimenterErrorNotification notification) {
- // TODO Auto-generated method stub
-
- };
-
- }
-
// Commented out DataChangeListene - to be used by Stats
// final class FlowDataListener implements DataChangeListener {
// }
// }
+ public boolean validate(Flow flow) {
+
+ String msg = ""; // Specific part of warn/error log
+
+ boolean result = true;
+ // flow Name validation
+ if (flow.getFlowName() == null || flow.getFlowName().trim().isEmpty() || !flow.getFlowName().matches(NAMEREGEX)) {
+ msg = "Invalid Flow name";
+ result = false;
+ }
+ // Node Validation
+ if (result == true && flow.getNode() == null) {
+ msg = "Node is null";
+ result = false;
+ }
+
+ // TODO: Validate we are seeking to program a flow against a valid
+ // Node
+
+ if (result == true && flow.getPriority() != null) {
+ if (flow.getPriority() < 0 || flow.getPriority() > 65535) {
+ msg = String.format("priority %s is not in the range 0 - 65535", flow.getPriority());
+ result = false;
+ }
+ }
+
+ // Presence check
+ /*
+ * This is breaking due to some improper use of caches...
+ *
+ * if (flowEntryExists(flow)) { String error =
+ * "Entry with this name on specified table already exists";
+ * logger.warn(
+ * "Entry with this name on specified table already exists: {}" ,
+ * entry); logger.error(error); return; } if
+ * (originalSwView.containsKey(entry)) { logger.warn(
+ * "Operation Rejected: A flow with same match and priority exists on the target node"
+ * ); logger.trace("Aborting to install {}", entry); continue; }
+ */
+ if (!FRMUtil.validateMatch(flow)) {
+ logger.error("Not a valid Match");
+ result = false;
+ }
+ if (!FRMUtil.validateInstructions(flow)) {
+ logger.error("Not a valid Instruction");
+ result = false;
+ }
+ if (result == false) {
+ logger.warn("Invalid Configuration for flow {}. The failure is {}", flow, msg);
+ logger.error("Invalid Configuration ({})", msg);
+ }
+ return result;
+ }
+
private static void updateLocalDatabase(NodeFlow entry, boolean add) {
updateSwViewes(entry, add);
Code Review / controller.git / blobdiff