import org.opendaylight.controller.clustering.services.ICacheUpdateAware;
import org.opendaylight.controller.clustering.services.IClusterContainerServices;
import org.opendaylight.controller.clustering.services.IClusterServices;
+import org.opendaylight.controller.configuration.ConfigurationObject;
import org.opendaylight.controller.configuration.IConfigurationContainerAware;
+import org.opendaylight.controller.configuration.IConfigurationContainerService;
import org.opendaylight.controller.connectionmanager.IConnectionManager;
import org.opendaylight.controller.containermanager.IContainerManager;
import org.opendaylight.controller.forwardingrulesmanager.FlowConfig;
import org.opendaylight.controller.sal.utils.EtherTypes;
import org.opendaylight.controller.sal.utils.GlobalConstants;
import org.opendaylight.controller.sal.utils.IObjectReader;
-import org.opendaylight.controller.sal.utils.ObjectReader;
-import org.opendaylight.controller.sal.utils.ObjectWriter;
import org.opendaylight.controller.sal.utils.Status;
import org.opendaylight.controller.sal.utils.StatusCode;
import org.opendaylight.controller.switchmanager.IInventoryListener;
private static final String PORT_REMOVED = "Port removed";
private static final String NODE_DOWN = "Node is Down";
private static final String INVALID_FLOW_ENTRY = "Invalid FlowEntry";
- private String frmFileName;
- private String portGroupFileName;
+ private static final String STATIC_FLOWS_FILE_NAME = "frm_staticflows.conf";
+ private static final String PORT_GROUP_FILE_NAME = "portgroup.conf";
private ConcurrentMap<Integer, FlowConfig> staticFlows;
private ConcurrentMap<Integer, Integer> staticFlowsOrdinal;
private ConcurrentMap<String, PortGroupConfig> portGroupConfigs;
private ConcurrentMap<PortGroupConfig, Map<Node, PortGroup>> portGroupData;
private ConcurrentMap<String, Object> TSPolicies;
private IContainerManager containerManager;
+ private IConfigurationContainerService configurationService;
private boolean inContainerMode; // being used by global instance only
protected boolean stopping;
for (FlowEntryInstall installEntry : toInstallSafe) {
// Install and update database
- Status ret = addEntriesInternal(installEntry, async);
+ Status ret = addEntryInternal(installEntry, async);
if (ret.isSuccess()) {
oneSucceded = true;
Status succeeded = null;
boolean decouple = false;
if (installedList.size() != toInstallList.size()) {
- log.info("Modify: New flow entry does not satisfy the same "
+ log.trace("Modify: New flow entry does not satisfy the same "
+ "number of container flows as the original entry does");
decouple = true;
}
*/
FlowEntryInstall sameMatchEntry = installedSwView.get(installEntry);
if (sameMatchEntry != null && !sameMatchEntry.getOriginal().equals(currentFlowEntry)) {
- log.info("Modify: new container flow merged flow entry clashes with existing flow");
+ log.trace("Modify: new container flow merged flow entry clashes with existing flow");
decouple = true;
} else {
toInstallSafe.add(installEntry);
}
// Install new entries
for (FlowEntryInstall newEntry : toInstallSafe) {
- succeeded = this.addEntriesInternal(newEntry, async);
+ succeeded = this.addEntryInternal(newEntry, async);
}
} else {
/*
/**
* This is the function that modifies the final container flows merged
* entries on the network node and update the database. It expects that all
- * the validity checks are passed
+ * the validity checks are passed.
+ * This function is supposed to be called only on the controller on which
+ * the IFRM call is executed.
*
* @param currentEntries
* @param newEntries
* contain the unique id assigned to this request
*/
private Status modifyEntryInternal(FlowEntryInstall currentEntries, FlowEntryInstall newEntries, boolean async) {
+ Status status = new Status(StatusCode.UNDEFINED);
FlowEntryDistributionOrderFutureTask futureStatus =
distributeWorkOrder(currentEntries, newEntries, UpdateType.CHANGED);
if (futureStatus != null) {
- Status retStatus = new Status(StatusCode.UNDEFINED);
try {
- retStatus = futureStatus.get();
- if (retStatus.getCode()
+ status = futureStatus.get();
+ if (status.getCode()
.equals(StatusCode.TIMEOUT)) {
// A timeout happened, lets cleanup the workMonitor
workMonitor.remove(futureStatus.getOrder());
} catch (ExecutionException e) {
log.error("", e);
}
- return retStatus;
} else {
// Modify the flow on the network node
- Status status = async ? programmer.modifyFlowAsync(currentEntries.getNode(), currentEntries.getInstall()
- .getFlow(), newEntries.getInstall()
- .getFlow()) : programmer.modifyFlow(currentEntries.getNode(), currentEntries.getInstall()
- .getFlow(), newEntries.getInstall()
- .getFlow());
+ status = modifyEntryInHw(currentEntries, newEntries, async);
+ }
- if (!status.isSuccess()) {
- log.trace("SDN Plugin failed to program the flow: {}. The failure is: {}", newEntries.getInstall(),
- status.getDescription());
- return status;
- }
+ if (!status.isSuccess()) {
+ log.trace("{} SDN Plugin failed to program the flow: {}. The failure is: {}",
+ (futureStatus != null) ? "Remote" : "Local", newEntries.getInstall(), status.getDescription());
+ return status;
+ }
- log.trace("Modified {} => {}", currentEntries.getInstall(), newEntries.getInstall());
+ log.trace("Modified {} => {}", currentEntries.getInstall(), newEntries.getInstall());
- // Update DB
- newEntries.setRequestId(status.getRequestId());
- updateSwViews(currentEntries, false);
- updateSwViews(newEntries, true);
+ // Update DB
+ newEntries.setRequestId(status.getRequestId());
+ updateSwViews(currentEntries, false);
+ updateSwViews(newEntries, true);
- return status;
- }
+ return status;
+ }
+
+ private Status modifyEntryInHw(FlowEntryInstall currentEntries, FlowEntryInstall newEntries, boolean async) {
+ return async ? programmer.modifyFlowAsync(currentEntries.getNode(), currentEntries.getInstall().getFlow(),
+ newEntries.getInstall().getFlow()) : programmer.modifyFlow(currentEntries.getNode(), currentEntries
+ .getInstall().getFlow(), newEntries.getInstall().getFlow());
}
/**
* This is the function that removes the final container flows merged entry
* from the network node and update the database. It expects that all the
* validity checks are passed
+ * This function is supposed to be called only on the controller on which
+ * the IFRM call is executed.
*
* @param entry
* the flow entry to remove
* contain the unique id assigned to this request
*/
private Status removeEntryInternal(FlowEntryInstall entry, boolean async) {
+ Status status = new Status(StatusCode.UNDEFINED);
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)) {
+ status = futureStatus.get();
+ if (status.getCode().equals(StatusCode.TIMEOUT)) {
// A timeout happened, lets cleanup the workMonitor
workMonitor.remove(futureStatus.getOrder());
}
} catch (ExecutionException e) {
log.error("", e);
}
- return retStatus;
} else {
// Mark the entry to be deleted (for CC just in case we fail)
entry.toBeDeleted();
// Remove from node
- Status status = async ? programmer.removeFlowAsync(entry.getNode(), entry.getInstall()
- .getFlow()) : programmer.removeFlow(entry.getNode(), entry.getInstall()
- .getFlow());
-
- if (!status.isSuccess()) {
- log.trace("SDN Plugin failed to remove the flow: {}. The failure is: {}", entry.getInstall(),
- status.getDescription());
- return status;
- }
- log.trace("Removed {}", entry.getInstall());
-
- // Update DB
- updateSwViews(entry, false);
+ status = removeEntryInHw(entry, async);
+ }
+ if (!status.isSuccess()) {
+ log.trace("{} SDN Plugin failed to remove the flow: {}. The failure is: {}",
+ (futureStatus != null) ? "Remote" : "Local", entry.getInstall(), status.getDescription());
return status;
}
+
+ log.trace("Removed {}", entry.getInstall());
+
+ // Update DB
+ updateSwViews(entry, false);
+
+ return status;
+ }
+
+ private Status removeEntryInHw(FlowEntryInstall entry, boolean async) {
+ return async ? programmer.removeFlowAsync(entry.getNode(), entry.getInstall().getFlow()) : programmer
+ .removeFlow(entry.getNode(), entry.getInstall().getFlow());
}
/**
* on the network node and updates the database. It expects that all the
* validity and conflict checks are passed. That means it does not check
* whether this flow would conflict or overwrite an existing one.
+ * This function is supposed to be called only on the controller on which
+ * the IFRM call is executed.
*
* @param entry
* the flow entry to install
* @return the status of this request. In case of asynchronous call, it will
* contain the unique id assigned to this request
*/
- private Status addEntriesInternal(FlowEntryInstall entry, boolean async) {
+ private Status addEntryInternal(FlowEntryInstall entry, boolean async) {
+ Status status = new Status(StatusCode.UNDEFINED);
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)) {
+ status = futureStatus.get();
+ if (status.getCode().equals(StatusCode.TIMEOUT)) {
// A timeout happened, lets cleanup the workMonitor
workMonitor.remove(futureStatus.getOrder());
}
} catch (ExecutionException e) {
log.error("", e);
}
- return retStatus;
} else {
- // Install the flow on the network node
- Status status = async ? programmer.addFlowAsync(entry.getNode(), entry.getInstall()
- .getFlow()) : programmer.addFlow(entry.getNode(), entry.getInstall()
- .getFlow());
+ status = addEntryInHw(entry, async);
+ }
- if (!status.isSuccess()) {
- log.trace("SDN Plugin failed to program the flow: {}. The failure is: {}", entry.getInstall(),
- status.getDescription());
- return status;
- }
+ if (!status.isSuccess()) {
+ log.trace("{} SDN Plugin failed to program the flow: {}. The failure is: {}",
+ (futureStatus != null) ? "Remote" : "Local", entry.getInstall(), status.getDescription());
+ return status;
+ }
- log.trace("Added {}", entry.getInstall());
+ log.trace("Added {}", entry.getInstall());
- // Update DB
- entry.setRequestId(status.getRequestId());
- updateSwViews(entry, true);
+ // Update DB
+ entry.setRequestId(status.getRequestId());
+ updateSwViews(entry, true);
- return status;
- }
+ return status;
+ }
+
+ private Status addEntryInHw(FlowEntryInstall entry, boolean async) {
+ // Install the flow on the network node
+ return async ? programmer.addFlowAsync(entry.getNode(), entry.getInstall().getFlow()) : programmer.addFlow(
+ entry.getNode(), entry.getInstall().getFlow());
}
/**
}
Status error = modifyEntry(currentFlowEntry, newFlowEntry, false);
if (error.isSuccess()) {
- log.info("Ports {} added to FlowEntry {}", portList, flowName);
+ log.trace("Ports {} added to FlowEntry {}", portList, flowName);
} else {
log.warn("Failed to add ports {} to Flow entry {}. The failure is: {}", portList,
currentFlowEntry.toString(), error.getDescription());
}
Status status = modifyEntry(currentFlowEntry, newFlowEntry, false);
if (status.isSuccess()) {
- log.info("Ports {} removed from FlowEntry {}", portList, flowName);
+ log.trace("Ports {} removed from FlowEntry {}", portList, flowName);
} else {
log.warn("Failed to remove ports {} from Flow entry {}. The failure is: {}", portList,
currentFlowEntry.toString(), status.getDescription());
Status status = modifyEntry(currentFlowEntry, newFlowEntry, false);
if (status.isSuccess()) {
- log.info("Output port replaced with {} for flow {} on node {}", outPort, flowName, node);
+ log.trace("Output port replaced with {} for flow {} on node {}", outPort, flowName, node);
} else {
log.warn("Failed to replace output port for flow {} on node {}. The failure is: {}", flowName, node,
status.getDescription());
// Do not attempt to reinstall the flow, warn user
if (newFlowConfig.equals(oldFlowConfig)) {
String msg = "No modification detected";
- log.info("Static flow modification skipped. New flow and old flow are the same: {}", newFlowConfig);
+ log.trace("Static flow modification skipped. New flow and old flow are the same: {}", newFlowConfig);
return new Status(StatusCode.SUCCESS, msg);
}
* inactive list
*/
private void uninstallAllFlowEntries(boolean preserveFlowEntries) {
- log.info("Uninstalling all non-internal flows");
+ log.trace("Uninstalling all non-internal flows");
List<FlowEntryInstall> toRemove = new ArrayList<FlowEntryInstall>();
* default container instance of FRM only when the last container is deleted
*/
private void reinstallAllFlowEntries() {
- log.info("Reinstalling all inactive flows");
+ log.trace("Reinstalling all inactive flows");
for (FlowEntry flowEntry : this.inactiveFlows.keySet()) {
this.addEntry(flowEntry, false);
@Override
public List<FlowConfig> getStaticFlows() {
- return getStaticFlowsOrderedList(staticFlows, staticFlowsOrdinal.get(0).intValue());
- }
-
- // TODO: need to come out with a better algorithm for maintaining the order
- // of the configuration entries
- // with actual one, index associated to deleted entries cannot be reused and
- // map grows...
- private List<FlowConfig> getStaticFlowsOrderedList(ConcurrentMap<Integer, FlowConfig> flowMap, int maxKey) {
- List<FlowConfig> orderedList = new ArrayList<FlowConfig>();
- for (int i = 0; i <= maxKey; i++) {
- FlowConfig entry = flowMap.get(i);
- if (entry != null) {
- orderedList.add(entry);
- }
- }
- return orderedList;
+ return new ArrayList<FlowConfig>(staticFlows.values());
}
@Override
return new ArrayList<Node>(set);
}
- @SuppressWarnings("unchecked")
private void loadFlowConfiguration() {
- ObjectReader objReader = new ObjectReader();
- ConcurrentMap<Integer, FlowConfig> confList = (ConcurrentMap<Integer, FlowConfig>) objReader.read(this,
- frmFileName);
-
- ConcurrentMap<String, PortGroupConfig> pgConfig = (ConcurrentMap<String, PortGroupConfig>) objReader.read(this,
- portGroupFileName);
-
- if (pgConfig != null) {
- for (ConcurrentMap.Entry<String, PortGroupConfig> entry : pgConfig.entrySet()) {
- addPortGroupConfig(entry.getKey(), entry.getValue().getMatchString(), true);
- }
+ for (ConfigurationObject conf : configurationService.retrieveConfiguration(this, PORT_GROUP_FILE_NAME)) {
+ addPortGroupConfig(((PortGroupConfig) conf).getName(), ((PortGroupConfig) conf).getMatchString(), true);
}
- if (confList == null) {
- return;
- }
-
- int maxKey = 0;
- for (Integer key : confList.keySet()) {
- if (key.intValue() > maxKey) {
- maxKey = key.intValue();
- }
- }
-
- for (FlowConfig conf : getStaticFlowsOrderedList(confList, maxKey)) {
- addStaticFlowInternal(conf, true);
+ for (ConfigurationObject conf : configurationService.retrieveConfiguration(this, STATIC_FLOWS_FILE_NAME)) {
+ addStaticFlowInternal((FlowConfig) conf, true);
}
}
}
private Status saveConfigInternal() {
- ObjectWriter objWriter = new ObjectWriter();
- ConcurrentMap<Integer, FlowConfig> nonDynamicFlows = new ConcurrentHashMap<Integer, FlowConfig>();
+ List<ConfigurationObject> nonDynamicFlows = new ArrayList<ConfigurationObject>();
+
for (Integer ordinal : staticFlows.keySet()) {
FlowConfig config = staticFlows.get(ordinal);
// Do not save dynamic and controller generated static flows
if (config.isDynamic() || config.isInternalFlow()) {
continue;
}
- nonDynamicFlows.put(ordinal, config);
+ nonDynamicFlows.add(config);
}
- objWriter.write(nonDynamicFlows, frmFileName);
- objWriter.write(new ConcurrentHashMap<String, PortGroupConfig>(portGroupConfigs), portGroupFileName);
- return new Status(StatusCode.SUCCESS, null);
+
+ configurationService.persistConfiguration(nonDynamicFlows, STATIC_FLOWS_FILE_NAME);
+ configurationService.persistConfiguration(new ArrayList<ConfigurationObject>(portGroupConfigs.values()),
+ PORT_GROUP_FILE_NAME);
+
+ return new Status(StatusCode.SUCCESS);
}
@Override
public void subnetNotify(Subnet sub, boolean add) {
}
+ private boolean programInternalFlow(boolean proactive, FlowConfig fc) {
+ boolean retVal = true; // program flows unless determined otherwise
+ if(proactive) {
+ // if the flow already exists do not program
+ if(flowConfigExists(fc)) {
+ retVal = false;
+ }
+ } else {
+ // if the flow does not exist do not program
+ if(!flowConfigExists(fc)) {
+ retVal = false;
+ }
+ }
+ return retVal;
+ }
+
/**
* (non-Javadoc)
*
dropAllConfig.setActions(dropAction);
defaultConfigs.add(dropAllConfig);
- log.info("Forwarding mode for node {} set to {}", node, (proactive ? "proactive" : "reactive"));
+ log.trace("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());
+ // check if the frm really needs to act on the notification.
+ // this is to check against duplicate notifications
+ if(programInternalFlow(proactive, fc)) {
+ Status status = (proactive) ? addStaticFlowInternal(fc, false) : removeStaticFlow(fc);
+ if (status.isSuccess()) {
+ log.trace("{} Proactive Static flow: {}", (proactive ? "Installed" : "Removed"), fc.getName());
+ } else {
+ log.warn("Failed to {} Proactive Static flow: {}", (proactive ? "install" : "remove"),
+ fc.getName());
+ }
} else {
- log.warn("Failed to {} Proactive Static flow: {}", (proactive ? "install" : "remove"),
- fc.getName());
+ log.debug("Got redundant install request for internal flow: {} on node: {}. Request not sent to FRM.", fc.getName(), node);
}
}
return new Status(StatusCode.SUCCESS);
* @param node
*/
private void cleanDatabaseForNode(Node node) {
- log.info("Cleaning Flow database for Node {}", node);
+ log.trace("Cleaning Flow database for Node {}", node);
if (nodeFlows.containsKey(node)) {
List<FlowEntryInstall> toRemove = new ArrayList<FlowEntryInstall>(nodeFlows.get(node));
@Override
public void portGroupChanged(PortGroupConfig config, Map<Node, PortGroup> data, boolean add) {
- log.info("PortGroup Changed for: {} Data: {}", config, portGroupData);
+ log.trace("PortGroup Changed for: {} Data: {}", config, portGroupData);
Map<Node, PortGroup> existingData = portGroupData.get(config);
if (existingData != null) {
for (Map.Entry<Node, PortGroup> entry : data.entrySet()) {
}
}
+ public void setConfigurationContainerService(IConfigurationContainerService service) {
+ log.trace("Got configuration service set request {}", service);
+ this.configurationService = service;
+ }
+
+ public void unsetConfigurationContainerService(IConfigurationContainerService service) {
+ log.trace("Got configuration service UNset request");
+ this.configurationService = null;
+ }
+
@Override
public PortGroupProvider getPortGroupProvider() {
return portGroupProvider;
*
*/
void init() {
- frmFileName = GlobalConstants.STARTUPHOME.toString() + "frm_staticflows_" + this.getContainerName() + ".conf";
- portGroupFileName = GlobalConstants.STARTUPHOME.toString() + "portgroup_" + this.getContainerName() + ".conf";
inContainerMode = false;
FlowEntryInstall feiNew = workOrder.get(fe);
switch (fe.getUpType()) {
case ADDED:
- gotStatus = addEntriesInternal(feiCurrent, false);
+ gotStatus = addEntryInHw(feiCurrent, false);
break;
case CHANGED:
- gotStatus = modifyEntryInternal(feiCurrent, feiNew, false);
+ gotStatus = modifyEntryInHw(feiCurrent, feiNew, false);
break;
case REMOVED:
- gotStatus = removeEntryInternal(feiCurrent, false);
+ gotStatus = removeEntryInHw(feiCurrent, false);
break;
}
// Remove the Order
}
/*
- * Read startup and build database if we have not already gotten the
- * configurations synced from another node
+ * Read startup and build database if we are the coordinator
*/
- if (staticFlows.isEmpty()) {
- loadFlowConfiguration();
- }
+ loadFlowConfiguration();
+ }
+
+ /**
+ * Function called by the dependency manager before Container is Stopped and Destroyed.
+ */
+ public void containerStop() {
+ uninstallAllFlowEntries(false);
}
/**
*/
void stop() {
stopping = true;
- uninstallAllFlowEntries(false);
// Shutdown executor
this.executor.shutdownNow();
// Now walk all the workMonitor and wake up the one sleeping because
// staticFlowEntry should never be null.
// the null check is just an extra defensive check.
if(staticFlowEntry != null) {
- staticFlows.remove(staticFlowEntry.getKey());
+ // Modify status and update cluster cache
+ log.debug("Updating static flow configuration on async error event");
+ String status = String.format("Cannot be installed on node. reason: %s", errorString);
+ staticFlowEntry.getValue().setStatus(status);
+ refreshClusterStaticFlowsStatus(node);
}
}
}
Code Review / controller.git / blobdiff