import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
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.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.connectionmanager.IConnectionManager;
import org.opendaylight.controller.forwardingrulesmanager.FlowConfig;
import org.opendaylight.controller.forwardingrulesmanager.FlowEntry;
}
Node n = e.getNode();
- if (!connectionManager.isLocal(n)) {
+ if (connectionManager.getLocalityStatus(n) == ConnectionLocality.NOT_LOCAL) {
// Create the work order and distribute it
FlowEntryDistributionOrder fe =
new FlowEntryDistributionOrder(e, t, clusterContainerService.getMyAddress());
return ret;
}
- logsync.trace("LOCAL Node {} so processing Entry:{} UpdateType:{}", n, e, t);
-
+ logsync.trace("Node {} could be local. so processing Entry:{} UpdateType:{}", n, e, t);
return null;
}
EnumSet.of(IClusterServices.cacheMode.TRANSACTIONAL));
clusterContainerService.createCache(WORKSTATUSCACHE,
- EnumSet.of(IClusterServices.cacheMode.NON_TRANSACTIONAL));
+ EnumSet.of(IClusterServices.cacheMode.NON_TRANSACTIONAL));
clusterContainerService.createCache(WORKORDERCACHE,
- EnumSet.of(IClusterServices.cacheMode.NON_TRANSACTIONAL));
+ EnumSet.of(IClusterServices.cacheMode.NON_TRANSACTIONAL));
} catch (CacheConfigException cce) {
log.error("CacheConfigException");
}
}
if (target != null) {
- // Program the network node
- Status status = (target.installInHw()) ? this.uninstallFlowEntry(target.getFlowEntry()) : this
- .installFlowEntry(target.getFlowEntry());
+ Status status = target.validate(container);
+ if (!status.isSuccess()) {
+ log.warn(status.getDescription());
+ return status;
+ }
+ status = (target.installInHw()) ? this.uninstallFlowEntry(target.getFlowEntry()) : this
+ .installFlowEntry(target.getFlowEntry());
if (status.isSuccess()) {
// Update Configuration database
target.setStatus(SUCCESS);
/**
* Uninstall all the non-internal Flow Entries present in the software view.
- * A copy of each entry is stored in the inactive list so that it can be
- * re-applied when needed. This function is called on the global instance of
- * FRM only, when the first container is created
+ * 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)
+ *
+ * @param preserveFlowEntries
+ * if true, a copy of each original entry is stored in the
+ * inactive list
*/
- private void uninstallAllFlowEntries() {
+ private void uninstallAllFlowEntries(boolean preserveFlowEntries) {
log.info("Uninstalling all non-internal flows");
+ List<FlowEntryInstall> toRemove = new ArrayList<FlowEntryInstall>();
+
// Store entries / create target list
for (ConcurrentMap.Entry<FlowEntryInstall, FlowEntryInstall> mapEntry : installedSwView.entrySet()) {
FlowEntryInstall flowEntries = mapEntry.getValue();
// Skip internal generated static flows
if (!flowEntries.isInternal()) {
- inactiveFlows.put(flowEntries.getOriginal(), flowEntries.getOriginal());
+ toRemove.add(flowEntries);
+ // Store the original entries if requested
+ if (preserveFlowEntries) {
+ inactiveFlows.put(flowEntries.getOriginal(), flowEntries.getOriginal());
+ }
}
}
// Now remove the entries
- for (FlowEntry flowEntry : inactiveFlows.keySet()) {
- Status status = this.removeEntry(flowEntry, false);
+ for (FlowEntryInstall flowEntryHw : toRemove) {
+ Status status = this.removeEntryInternal(flowEntryHw, false);
if (!status.isSuccess()) {
- log.warn("Failed to remove entry: {}. The failure is: {}", flowEntry, status.getDescription());
+ log.warn("Failed to remove entry: {}. The failure is: {}", flowEntryHw, status.getDescription());
}
}
}
addStaticFlowInternal(allowARP, true); // skip validation on internal static flow name
}
+ /**
+ * (non-Javadoc)
+ *
+ * @see org.opendaylight.controller.switchmanager.ISwitchManagerAware#modeChangeNotify(org.opendaylight.controller.sal.core.Node,
+ * boolean)
+ *
+ * This method can be called from within the OSGi framework context,
+ * given the programming operation can take sometime, it not good
+ * pratice to have in it's context operations that can take time,
+ * hence moving off to a different thread for async processing.
+ */
+ private ExecutorService executor;
@Override
- public void modeChangeNotify(Node node, boolean proactive) {
- List<FlowConfig> defaultConfigs = new ArrayList<FlowConfig>();
-
- List<String> puntAction = new ArrayList<String>();
- puntAction.add(ActionType.CONTROLLER.toString());
-
- FlowConfig allowARP = new FlowConfig();
- allowARP.setInstallInHw(true);
- allowARP.setName(FlowConfig.INTERNALSTATICFLOWBEGIN + "Punt ARP" + FlowConfig.INTERNALSTATICFLOWEND);
- allowARP.setPriority("1");
- allowARP.setNode(node);
- allowARP.setEtherType("0x" + Integer.toHexString(EtherTypes.ARP.intValue()).toUpperCase());
- allowARP.setActions(puntAction);
- defaultConfigs.add(allowARP);
-
- FlowConfig allowLLDP = new FlowConfig();
- allowLLDP.setInstallInHw(true);
- allowLLDP.setName(FlowConfig.INTERNALSTATICFLOWBEGIN + "Punt LLDP" + FlowConfig.INTERNALSTATICFLOWEND);
- allowLLDP.setPriority("1");
- allowLLDP.setNode(node);
- allowLLDP.setEtherType("0x" + Integer.toHexString(EtherTypes.LLDP.intValue()).toUpperCase());
- allowLLDP.setActions(puntAction);
- defaultConfigs.add(allowLLDP);
-
- List<String> dropAction = new ArrayList<String>();
- dropAction.add(ActionType.DROP.toString());
-
- FlowConfig dropAllConfig = new FlowConfig();
- dropAllConfig.setInstallInHw(true);
- dropAllConfig.setName(FlowConfig.INTERNALSTATICFLOWBEGIN + "Catch-All Drop" + FlowConfig.INTERNALSTATICFLOWEND);
- dropAllConfig.setPriority("0");
- dropAllConfig.setNode(node);
- dropAllConfig.setActions(dropAction);
- defaultConfigs.add(dropAllConfig);
-
- log.info("Forwarding mode for node {} set to {}", node, (proactive ? "proactive" : "reactive"));
- for (FlowConfig fc : defaultConfigs) {
- Status status = (proactive) ? addStaticFlowInternal(fc, false) : removeStaticFlow(fc);
- if (status.isSuccess()) {
- log.info("{} Proactive Static flow: {}", (proactive ? "Installed" : "Removed"), fc.getName());
- } else {
- log.warn("Failed to {} Proactive Static flow: {}", (proactive ? "install" : "remove"), fc.getName());
+ public void modeChangeNotify(final Node node, final boolean proactive) {
+ Callable<Status> modeChangeCallable = new Callable<Status>() {
+ @Override
+ public Status call() throws Exception {
+ List<FlowConfig> defaultConfigs = new ArrayList<FlowConfig>();
+
+ List<String> puntAction = new ArrayList<String>();
+ puntAction.add(ActionType.CONTROLLER.toString());
+
+ FlowConfig allowARP = new FlowConfig();
+ allowARP.setInstallInHw(true);
+ allowARP.setName(FlowConfig.INTERNALSTATICFLOWBEGIN + "Punt ARP" + FlowConfig.INTERNALSTATICFLOWEND);
+ allowARP.setPriority("1");
+ allowARP.setNode(node);
+ allowARP.setEtherType("0x" + Integer.toHexString(EtherTypes.ARP.intValue())
+ .toUpperCase());
+ allowARP.setActions(puntAction);
+ defaultConfigs.add(allowARP);
+
+ FlowConfig allowLLDP = new FlowConfig();
+ allowLLDP.setInstallInHw(true);
+ allowLLDP.setName(FlowConfig.INTERNALSTATICFLOWBEGIN + "Punt LLDP" + FlowConfig.INTERNALSTATICFLOWEND);
+ allowLLDP.setPriority("1");
+ allowLLDP.setNode(node);
+ allowLLDP.setEtherType("0x" + Integer.toHexString(EtherTypes.LLDP.intValue())
+ .toUpperCase());
+ allowLLDP.setActions(puntAction);
+ defaultConfigs.add(allowLLDP);
+
+ List<String> dropAction = new ArrayList<String>();
+ dropAction.add(ActionType.DROP.toString());
+
+ FlowConfig dropAllConfig = new FlowConfig();
+ dropAllConfig.setInstallInHw(true);
+ dropAllConfig.setName(FlowConfig.INTERNALSTATICFLOWBEGIN + "Catch-All Drop"
+ + FlowConfig.INTERNALSTATICFLOWEND);
+ dropAllConfig.setPriority("0");
+ dropAllConfig.setNode(node);
+ dropAllConfig.setActions(dropAction);
+ defaultConfigs.add(dropAllConfig);
+
+ log.info("Forwarding mode for node {} set to {}", node, (proactive ? "proactive" : "reactive"));
+ for (FlowConfig fc : defaultConfigs) {
+ Status status = (proactive) ? addStaticFlowInternal(fc, false) : removeStaticFlow(fc);
+ if (status.isSuccess()) {
+ log.info("{} Proactive Static flow: {}", (proactive ? "Installed" : "Removed"), fc.getName());
+ } else {
+ log.warn("Failed to {} Proactive Static flow: {}", (proactive ? "install" : "remove"),
+ fc.getName());
+ }
+ }
+ return new Status(StatusCode.SUCCESS);
}
- }
+ };
+
+ /*
+ * Execute the work outside the caller context, this could be an
+ * expensive operation and we don't want to block the caller for it.
+ */
+ this.executor.submit(modeChangeCallable);
}
/**
} else {
log.warn("Not expected null WorkStatus", work);
}
+ } else if (event instanceof ContainerFlowChangeEvent) {
+ /*
+ * Whether it is an addition or removal, we have to
+ * recompute the merged flows entries taking into
+ * account all the current container flows because
+ * flow merging is not an injective function
+ */
+ updateFlowsContainerFlow();
} else {
log.warn("Dequeued unknown event {}", event.getClass()
.getSimpleName());
// Initialize graceful stop flag
stopping = false;
+ // Allocate the executor service
+ this.executor = Executors.newSingleThreadExecutor();
+
// Start event handler thread
frmEventHandler.start();
*/
void stop() {
stopping = true;
- uninstallAllFlowEntries();
+ uninstallAllFlowEntries(false);
+ // Shutdown executor
+ this.executor.shutdownNow();
}
public void setFlowProgrammerService(IFlowProgrammerService service) {
}
log.trace("Container {}: Updating installed flows because of container flow change: {} {}",
container.getName(), t, current);
- /*
- * Whether it is an addition or removal, we have to recompute the merged
- * flows entries taking into account all the current container flows
- * because flow merging is not an injective function
- */
- updateFlowsContainerFlow();
+ ContainerFlowChangeEvent ev = new ContainerFlowChangeEvent(previous, current, t);
+ pendingEvents.offer(ev);
}
@Override
}
switch (update) {
case ADDED:
+ /*
+ * Controller is moving to container mode. We are in the default
+ * container context, we need to remove all our non-internal flows
+ * to prevent any container isolation breakage. We also need to
+ * preserve our flow so that they can be re-installed if we move
+ * back to non container mode (no containers).
+ */
this.inContainerMode = true;
- this.uninstallAllFlowEntries();
+ this.uninstallAllFlowEntries(true);
break;
case REMOVED:
this.inContainerMode = false;
return newEntry;
}
}
+ private class ContainerFlowChangeEvent extends FRMEvent {
+ private final ContainerFlow previous;
+ private final ContainerFlow current;
+ private final UpdateType type;
+
+ public ContainerFlowChangeEvent(ContainerFlow previous, ContainerFlow current, UpdateType type) {
+ this.previous = previous;
+ this.current = current;
+ this.type = type;
+ }
+
+ public ContainerFlow getPrevious() {
+ return this.previous;
+ }
+
+ public ContainerFlow getCurrent() {
+ return this.current;
+ }
+
+ public UpdateType getType() {
+ return this.type;
+ }
+ }
+
private class WorkStatusCleanup extends FRMEvent {
private FlowEntryDistributionOrder fe;
return;
}
Node n = fei.getNode();
- if (connectionManager.isLocal(n)) {
+ if (connectionManager.getLocalityStatus(n) == ConnectionLocality.LOCAL) {
logsync.trace("workOrder for fe {} processed locally", fe);
// I'm the controller in charge for the request, queue it for
// processing
Code Review / controller.git / blobdiff