import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import org.eclipse.osgi.framework.console.CommandInterpreter;
import org.opendaylight.controller.clustering.services.IClusterContainerServices;
import org.opendaylight.controller.clustering.services.IClusterServices;
import org.opendaylight.controller.configuration.IConfigurationContainerAware;
-import org.opendaylight.controller.connectionmanager.ConnectionLocality;
+import org.opendaylight.controller.sal.connection.ConnectionLocality;
import org.opendaylight.controller.connectionmanager.IConnectionManager;
import org.opendaylight.controller.forwardingrulesmanager.FlowConfig;
import org.opendaylight.controller.forwardingrulesmanager.FlowEntry;
import org.opendaylight.controller.sal.action.Flood;
import org.opendaylight.controller.sal.action.Output;
import org.opendaylight.controller.sal.action.PopVlan;
+import org.opendaylight.controller.sal.core.Config;
import org.opendaylight.controller.sal.core.ContainerFlow;
import org.opendaylight.controller.sal.core.IContainer;
import org.opendaylight.controller.sal.core.IContainerListener;
IConfigurationContainerAware,
IInventoryListener,
IObjectReader,
- ICacheUpdateAware,
+ ICacheUpdateAware<Object,Object>,
CommandProvider,
IFlowProgrammerListener {
private static final String NODEDOWN = "Node is Down";
* @return a Future object for monitoring the progress of the result, or
* null in case the processing should take place locally
*/
- private Future<Status> distributeWorkOrder(FlowEntryInstall e, FlowEntryInstall u, UpdateType t) {
+ private FlowEntryDistributionOrderFutureTask distributeWorkOrder(FlowEntryInstall e, FlowEntryInstall u,
+ UpdateType t) {
// A null entry it's an unexpected condition, anyway it's safe to keep
// the handling local
if (e == null) {
* contain the unique id assigned to this request
*/
private Status modifyEntryInternal(FlowEntryInstall currentEntries, FlowEntryInstall newEntries, boolean async) {
- Future<Status> futureStatus = distributeWorkOrder(currentEntries, newEntries, UpdateType.CHANGED);
+ FlowEntryDistributionOrderFutureTask futureStatus =
+ distributeWorkOrder(currentEntries, newEntries, UpdateType.CHANGED);
if (futureStatus != null) {
Status retStatus = new Status(StatusCode.UNDEFINED);
try {
retStatus = futureStatus.get();
+ if (retStatus.getCode()
+ .equals(StatusCode.TIMEOUT)) {
+ // A timeout happened, lets cleanup the workMonitor
+ workMonitor.remove(futureStatus.getOrder());
+ }
} catch (InterruptedException e) {
log.error("", e);
} catch (ExecutionException e) {
* contain the unique id assigned to this request
*/
private Status removeEntryInternal(FlowEntryInstall entry, boolean async) {
- Future<Status> futureStatus = distributeWorkOrder(entry, null, UpdateType.REMOVED);
+ FlowEntryDistributionOrderFutureTask futureStatus = distributeWorkOrder(entry, null, UpdateType.REMOVED);
if (futureStatus != null) {
Status retStatus = new Status(StatusCode.UNDEFINED);
try {
retStatus = futureStatus.get();
+ if (retStatus.getCode()
+ .equals(StatusCode.TIMEOUT)) {
+ // A timeout happened, lets cleanup the workMonitor
+ workMonitor.remove(futureStatus.getOrder());
+ }
} catch (InterruptedException e) {
log.error("", e);
} catch (ExecutionException e) {
* contain the unique id assigned to this request
*/
private Status addEntriesInternal(FlowEntryInstall entry, boolean async) {
- Future<Status> futureStatus = distributeWorkOrder(entry, null, UpdateType.ADDED);
+ FlowEntryDistributionOrderFutureTask futureStatus = distributeWorkOrder(entry, null, UpdateType.ADDED);
if (futureStatus != null) {
Status retStatus = new Status(StatusCode.UNDEFINED);
try {
retStatus = futureStatus.get();
+ if (retStatus.getCode()
+ .equals(StatusCode.TIMEOUT)) {
+ // A timeout happened, lets cleanup the workMonitor
+ workMonitor.remove(futureStatus.getOrder());
+ }
} catch (InterruptedException e) {
log.error("", e);
} catch (ExecutionException e) {
return true;
}
+ private ConcurrentMap.Entry<Integer, FlowConfig> getStaticFlowEntry(String name, Node node) {
+ for (ConcurrentMap.Entry<Integer, FlowConfig> flowEntry : staticFlows.entrySet()) {
+ FlowConfig flowConfig = flowEntry.getValue();
+ if (flowConfig.isByNameAndNodeIdEqual(name, node)) {
+ return flowEntry;
+ }
+ }
+ return null;
+ }
+
private void updateLocalDatabase(FlowEntryInstall entry, boolean add) {
// Update the software view
updateSwViewes(entry, add);
retrieveCaches();
}
- @SuppressWarnings("deprecation")
private void allocateCaches() {
if (this.clusterContainerService == null) {
log.warn("Un-initialized clusterContainerService, can't create cache");
EnumSet.of(IClusterServices.cacheMode.TRANSACTIONAL));
clusterContainerService.createCache(WORKSTATUSCACHE,
- EnumSet.of(IClusterServices.cacheMode.NON_TRANSACTIONAL));
+ EnumSet.of(IClusterServices.cacheMode.NON_TRANSACTIONAL, IClusterServices.cacheMode.ASYNC));
clusterContainerService.createCache(WORKORDERCACHE,
- EnumSet.of(IClusterServices.cacheMode.NON_TRANSACTIONAL));
+ EnumSet.of(IClusterServices.cacheMode.NON_TRANSACTIONAL, IClusterServices.cacheMode.ASYNC));
} catch (CacheConfigException cce) {
log.error("CacheConfigException");
}
}
- @SuppressWarnings({ "unchecked", "deprecation" })
+ @SuppressWarnings({ "unchecked" })
private void retrieveCaches() {
ConcurrentMap<?, ?> map;
* If requested, a copy of each original flow entry will be stored in the
* inactive list so that it can be re-applied when needed (This is typically
* the case when running in the default container and controller moved to
- * container mode)
+ * container mode) NOTE WELL: The routine as long as does a bulk change will
+ * operate only on the entries for nodes locally attached so to avoid
+ * redundant operations initiated by multiple nodes
*
* @param preserveFlowEntries
* if true, a copy of each original entry is stored in the
// Now remove the entries
for (FlowEntryInstall flowEntryHw : toRemove) {
- Status status = this.removeEntryInternal(flowEntryHw, false);
- if (!status.isSuccess()) {
- log.warn("Failed to remove entry: {}. The failure is: {}", flowEntryHw, status.getDescription());
+ Node n = flowEntryHw.getNode();
+ if (n != null && connectionManager.getLocalityStatus(n) == ConnectionLocality.LOCAL) {
+ Status status = this.removeEntryInternal(flowEntryHw, false);
+ if (!status.isSuccess()) {
+ log.warn("Failed to remove entry: {}. The failure is: {}", flowEntryHw, status.getDescription());
+ }
+ } else {
+ log.debug("Not removing entry {} because not connected locally, the remote guy will do it's job",
+ flowEntryHw);
}
}
}
@Override
public FlowConfig getStaticFlow(String name, Node node) {
- for (ConcurrentMap.Entry<Integer, FlowConfig> entry : staticFlows.entrySet()) {
- if (entry.getValue().isByNameAndNodeIdEqual(name, node)) {
- return entry.getValue();
- }
+ ConcurrentMap.Entry<Integer, FlowConfig> entry = getStaticFlowEntry(name, node);
+ if(entry != null) {
+ return entry.getValue();
}
return null;
}
@Override
public void notifyNodeConnector(NodeConnector nodeConnector, UpdateType type, Map<String, Property> propMap) {
+ boolean updateStaticFlowCluster = false;
+ switch (type) {
+ case ADDED:
+ break;
+ case CHANGED:
+ Config config = (propMap == null) ? null : (Config) propMap.get(Config.ConfigPropName);
+ if (config != null) {
+ switch (config.getValue()) {
+ case Config.ADMIN_DOWN:
+ log.trace("Port {} is administratively down: uninstalling interested flows", nodeConnector);
+ updateStaticFlowCluster = removeFlowsOnNodeConnectorDown(nodeConnector);
+ break;
+ case Config.ADMIN_UP:
+ log.trace("Port {} is administratively up: installing interested flows", nodeConnector);
+ updateStaticFlowCluster = installFlowsOnNodeConnectorUp(nodeConnector);
+ break;
+ case Config.ADMIN_UNDEF:
+ break;
+ default:
+ }
+ }
+ break;
+ case REMOVED:
+ // This is the case where a switch port is removed from the SDN agent space
+ log.trace("Port {} was removed from our control: uninstalling interested flows", nodeConnector);
+ updateStaticFlowCluster = removeFlowsOnNodeConnectorDown(nodeConnector);
+ break;
+ default:
+
+ }
+
+ if (updateStaticFlowCluster) {
+ refreshClusterStaticFlowsStatus(nodeConnector.getNode());
+ }
+ }
+
+ /*
+ * It goes through the static flows configuration, it identifies the ones
+ * which have the specified node connector as input or output port and
+ * install them on the network node if they are marked to be installed in
+ * hardware and their status shows they were not installed yet
+ */
+ private boolean installFlowsOnNodeConnectorUp(NodeConnector nodeConnector) {
+ boolean updated = false;
+ List<FlowConfig> flowConfigForNode = getStaticFlows(nodeConnector.getNode());
+ for (FlowConfig flowConfig : flowConfigForNode) {
+ if (doesFlowContainNodeConnector(flowConfig.getFlow(), nodeConnector)) {
+ if (flowConfig.installInHw() && !flowConfig.getStatus().equals(SUCCESS)) {
+ Status status = this.installFlowEntry(flowConfig.getFlowEntry());
+ if (!status.isSuccess()) {
+ flowConfig.setStatus(status.getDescription());
+ } else {
+ flowConfig.setStatus(SUCCESS);
+ }
+ updated = true;
+ }
+ }
+ }
+ return updated;
+ }
+
+ /*
+ * Remove from the network node all the flows which have the specified node
+ * connector as input or output port. If any of the flow entry is a static
+ * flow, it updates the correspondent configuration.
+ */
+ private boolean removeFlowsOnNodeConnectorDown(NodeConnector nodeConnector) {
+ boolean updated = false;
+ List<FlowEntryInstall> nodeFlowEntries = nodeFlows.get(nodeConnector.getNode());
+ if (nodeFlowEntries == null) {
+ return updated;
+ }
+ for (FlowEntryInstall fei : new ArrayList<FlowEntryInstall>(nodeFlowEntries)) {
+ if (doesFlowContainNodeConnector(fei.getInstall().getFlow(), nodeConnector)) {
+ Status status = this.removeEntryInternal(fei, true);
+ if (!status.isSuccess()) {
+ continue;
+ }
+ /*
+ * If the flow entry is a static flow, then update its
+ * configuration
+ */
+ if (fei.getGroupName().equals(FlowConfig.STATICFLOWGROUP)) {
+ FlowConfig flowConfig = getStaticFlow(fei.getFlowName(), fei.getNode());
+ if (flowConfig != null) {
+ flowConfig.setStatus(PORTREMOVED);
+ updated = true;
+ }
+ }
+ }
+ }
+ return updated;
}
private FlowConfig getDerivedFlowConfig(FlowConfig original, String configName, Short port) {
uninstallAllFlowEntries(false);
// Shutdown executor
this.executor.shutdownNow();
+ // Now walk all the workMonitor and wake up the one sleeping because
+ // destruction is happening
+ for (FlowEntryDistributionOrder fe : workMonitor.keySet()) {
+ FlowEntryDistributionOrderFutureTask task = workMonitor.get(fe);
+ task.cancel(true);
+ }
}
public void setFlowProgrammerService(IFlowProgrammerService service) {
switch (t) {
case REMOVED:
-
- List<FlowEntryInstall> nodeFlowEntries = nodeFlows.get(nc.getNode());
- if (nodeFlowEntries == null) {
- return;
- }
- for (FlowEntryInstall fei : new ArrayList<FlowEntryInstall>(nodeFlowEntries)) {
- if (doesFlowContainNodeConnector(fei.getInstall().getFlow(), nc)) {
- Status status = this.removeEntryInternal(fei, true);
- if (!status.isSuccess()) {
- continue;
- }
- /*
- * If the flow entry is a static flow, then update its
- * configuration
- */
- if (fei.getGroupName().equals(FlowConfig.STATICFLOWGROUP)) {
- FlowConfig flowConfig = getStaticFlow(fei.getFlowName(), fei.getNode());
- if (flowConfig != null) {
- flowConfig.setStatus(PORTREMOVED);
- updateStaticFlowCluster = true;
- }
- }
- }
- }
- if (updateStaticFlowCluster) {
- refreshClusterStaticFlowsStatus(nc.getNode());
- }
+ log.trace("Port {} was removed from container: uninstalling interested flows", nc);
+ updateStaticFlowCluster = removeFlowsOnNodeConnectorDown(nc);
break;
case ADDED:
- List<FlowConfig> flowConfigForNode = getStaticFlows(nc.getNode());
- for (FlowConfig flowConfig : flowConfigForNode) {
- if (doesFlowContainNodeConnector(flowConfig.getFlow(), nc)) {
- if (flowConfig.installInHw()) {
- Status status = this.installFlowEntry(flowConfig.getFlowEntry());
- if (!status.isSuccess()) {
- flowConfig.setStatus(status.getDescription());
- } else {
- flowConfig.setStatus(SUCCESS);
- }
- updateStaticFlowCluster = true;
- }
- }
- }
- if (updateStaticFlowCluster) {
- refreshClusterStaticFlowsStatus(nc.getNode());
- }
+ log.trace("Port {} was added to container: reinstall interested flows", nc);
+ updateStaticFlowCluster = installFlowsOnNodeConnectorUp(nc);
+
break;
case CHANGED:
break;
default:
}
+
+ if (updateStaticFlowCluster) {
+ refreshClusterStaticFlowsStatus(nc.getNode());
+ }
}
@Override
}
}
+ public void _frmProcessErrorEvent(CommandInterpreter ci) throws UnknownHostException {
+ Node node = null;
+ long reqId = 0L;
+ String nodeId = ci.nextArgument();
+ if (nodeId == null) {
+ ci.print("Node id not specified");
+ return;
+ }
+ String requestId = ci.nextArgument();
+ if (requestId == null) {
+ ci.print("Request id not specified");
+ return;
+ }
+ try {
+ node = NodeCreator.createOFNode(Long.valueOf(nodeId));
+ } catch (NumberFormatException e) {
+ ci.print("Node id not a number");
+ return;
+ }
+ try {
+ reqId = Long.parseLong(requestId);
+ } catch (NumberFormatException e) {
+ ci.print("Request id not a number");
+ return;
+ }
+ // null for error object is good enough for now
+ ErrorReportedEvent event = new ErrorReportedEvent(reqId, node, null);
+ this.processErrorEvent(event);
+ }
+
@Override
public void flowRemoved(Node node, Flow flow) {
log.trace("Received flow removed notification on {} for {}", node, flow);
* mapping will have to be added in future
*/
FlowEntryInstall target = null;
- for (FlowEntryInstall index : nodeFlows.get(node)) {
- FlowEntryInstall entry = installedSwView.get(index);
- if (entry.getRequestId() == rid) {
- target = entry;
- break;
+ List<FlowEntryInstall> flowEntryInstallList = nodeFlows.get(node);
+ // flowEntryInstallList could be null.
+ // so check for it.
+ if(flowEntryInstallList != null) {
+ for (FlowEntryInstall index : flowEntryInstallList) {
+ FlowEntryInstall entry = installedSwView.get(index);
+ if(entry != null) {
+ if (entry.getRequestId() == rid) {
+ target = entry;
+ break;
+ }
+ }
}
}
if (target != null) {
// This was a flow install, update database
this.updateLocalDatabase(target, false);
+ // also update the config
+ if(FlowConfig.STATICFLOWGROUP.equals(target.getGroupName())) {
+ ConcurrentMap.Entry<Integer, FlowConfig> staticFlowEntry = getStaticFlowEntry(target.getFlowName(),target.getNode());
+ // staticFlowEntry should never be null.
+ // the null check is just an extra defensive check.
+ if(staticFlowEntry != null) {
+ staticFlows.remove(staticFlowEntry.getKey());
+ }
+ }
}
// Notify listeners
*/
if (fe.getRequestorController()
.equals(clusterContainerService.getMyAddress())) {
- FlowEntryDistributionOrderFutureTask fet = workMonitor.get(fe);
+ FlowEntryDistributionOrderFutureTask fet = workMonitor.remove(fe);
if (fet != null) {
logsync.trace("workStatus response is for us {}", fe);
// Signal we got the status
Code Review / controller.git / blobdiff