*/
package org.opendaylight.openflowplugin.applications.frm.impl;
+import static org.opendaylight.openflowplugin.api.openflow.ReconciliationState.ReconciliationStatus.COMPLETED;
+import static org.opendaylight.openflowplugin.api.openflow.ReconciliationState.ReconciliationStatus.FAILED;
+import static org.opendaylight.openflowplugin.api.openflow.ReconciliationState.ReconciliationStatus.STARTED;
+
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.common.util.concurrent.FluentFuture;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
import java.math.BigInteger;
+import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.opendaylight.mdsal.binding.api.WriteTransaction;
import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
import org.opendaylight.openflowplugin.api.OFConstants;
+import org.opendaylight.openflowplugin.api.openflow.FlowGroupCacheManager;
+import org.opendaylight.openflowplugin.api.openflow.ReconciliationState;
import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo;
import org.opendaylight.openflowplugin.applications.frm.FlowNodeReconciliation;
import org.opendaylight.openflowplugin.applications.frm.ForwardingRulesManager;
private final String serviceName;
private final int priority;
private final ResultState resultState;
- private final Map<DeviceInfo, ListenableFuture<Boolean>> futureMap = new HashMap<>();
+ private final Map<DeviceInfo, ListenableFuture<Boolean>> futureMap = new ConcurrentHashMap<>();
private final ExecutorService executor = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
private static final AtomicLong BUNDLE_ID = new AtomicLong();
private static final BundleFlags BUNDLE_FLAGS = new BundleFlags(true, true);
+ private Map<String, ReconciliationState> reconciliationStates;
public FlowNodeReconciliationImpl(final ForwardingRulesManager manager, final DataBroker db,
- final String serviceName, final int priority, final ResultState resultState) {
+ final String serviceName, final int priority, final ResultState resultState,
+ final FlowGroupCacheManager flowGroupCacheManager) {
this.provider = Preconditions.checkNotNull(manager, "ForwardingRulesManager can not be null!");
dataBroker = Preconditions.checkNotNull(db, "DataBroker can not be null!");
this.serviceName = serviceName;
this.resultState = resultState;
salBundleService = Preconditions.checkNotNull(manager.getSalBundleService(),
"salBundleService can not be null!");
+ reconciliationStates = flowGroupCacheManager.getReconciliationStates();
}
@Override
}
}
+ @Override
+ public void flowNodeDisconnected(InstanceIdentifier<FlowCapableNode> disconnectedNode) {
+ String node = disconnectedNode.firstKeyOf(Node.class).getId().getValue();
+ BigInteger dpnId = getDpnIdFromNodeName(node);
+ reconciliationStates.remove(dpnId.toString());
+ }
+
private class BundleBasedReconciliationTask implements Callable<Boolean> {
final InstanceIdentifier<FlowCapableNode> nodeIdentity;
} catch (ExecutionException | InterruptedException e) {
LOG.error("Error occurred while reading the configuration data store for node {}", nodeIdentity, e);
}
- try {
- if (flowNode.isPresent()) {
- LOG.debug("FlowNode present for Datapath ID {}", dpnId);
- OF_EVENT_LOG.debug("Bundle Reconciliation Start, Node: {}", dpnId);
- final NodeRef nodeRef = new NodeRef(nodeIdentity.firstIdentifierOf(Node.class));
-
- final ControlBundleInput closeBundleInput = new ControlBundleInputBuilder().setNode(nodeRef)
- .setBundleId(bundleIdValue).setFlags(BUNDLE_FLAGS)
- .setType(BundleControlType.ONFBCTCLOSEREQUEST).build();
-
- final ControlBundleInput openBundleInput = new ControlBundleInputBuilder().setNode(nodeRef)
- .setBundleId(bundleIdValue).setFlags(BUNDLE_FLAGS)
- .setType(BundleControlType.ONFBCTOPENREQUEST).build();
-
- final ControlBundleInput commitBundleInput = new ControlBundleInputBuilder().setNode(nodeRef)
- .setBundleId(bundleIdValue).setFlags(BUNDLE_FLAGS)
- .setType(BundleControlType.ONFBCTCOMMITREQUEST).build();
-
- final AddBundleMessagesInput deleteAllFlowGroupsInput = new AddBundleMessagesInputBuilder()
- .setNode(nodeRef).setBundleId(bundleIdValue).setFlags(BUNDLE_FLAGS)
- .setMessages(createMessages(nodeRef)).build();
-
- LOG.debug("Closing openflow bundle for device {}", dpnId);
- /* Close previously opened bundle on the openflow switch if any */
- ListenableFuture<RpcResult<ControlBundleOutput>> closeBundle
- = salBundleService.controlBundle(closeBundleInput);
-
- /* Open a new bundle on the switch */
- ListenableFuture<RpcResult<ControlBundleOutput>> openBundle =
- Futures.transformAsync(closeBundle,
- rpcResult -> salBundleService.controlBundle(openBundleInput),
- service);
+
+ if (flowNode.isPresent()) {
+ ReconciliationState reconciliationState = new ReconciliationState(
+ STARTED, LocalDateTime.now());
+ //put the dpn info into the map
+ reconciliationStates.put(dpnId.toString(), reconciliationState);
+ LOG.debug("FlowNode present for Datapath ID {}", dpnId);
+ OF_EVENT_LOG.debug("Bundle Reconciliation Start, Node: {}", dpnId);
+ final NodeRef nodeRef = new NodeRef(nodeIdentity.firstIdentifierOf(Node.class));
+
+ final ControlBundleInput closeBundleInput = new ControlBundleInputBuilder().setNode(nodeRef)
+ .setBundleId(bundleIdValue).setFlags(BUNDLE_FLAGS)
+ .setType(BundleControlType.ONFBCTCLOSEREQUEST).build();
+
+ final ControlBundleInput openBundleInput = new ControlBundleInputBuilder().setNode(nodeRef)
+ .setBundleId(bundleIdValue).setFlags(BUNDLE_FLAGS)
+ .setType(BundleControlType.ONFBCTOPENREQUEST).build();
+
+ final ControlBundleInput commitBundleInput = new ControlBundleInputBuilder().setNode(nodeRef)
+ .setBundleId(bundleIdValue).setFlags(BUNDLE_FLAGS)
+ .setType(BundleControlType.ONFBCTCOMMITREQUEST).build();
+
+ final AddBundleMessagesInput deleteAllFlowGroupsInput = new AddBundleMessagesInputBuilder()
+ .setNode(nodeRef).setBundleId(bundleIdValue).setFlags(BUNDLE_FLAGS)
+ .setMessages(createMessages(nodeRef)).build();
+
+ LOG.debug("Closing openflow bundle for device {}", dpnId);
+ /* Close previously opened bundle on the openflow switch if any */
+ ListenableFuture<RpcResult<ControlBundleOutput>> closeBundle
+ = salBundleService.controlBundle(closeBundleInput);
+
+ /* Open a new bundle on the switch */
+ ListenableFuture<RpcResult<ControlBundleOutput>> openBundle =
+ Futures.transformAsync(closeBundle,
+ rpcResult -> salBundleService.controlBundle(openBundleInput),
+ service);
/* Push groups and flows via bundle add messages */
- ListenableFuture<RpcResult<AddBundleMessagesOutput>> deleteAllFlowGroupsFuture =
- Futures.transformAsync(openBundle, rpcResult -> {
- if (rpcResult.isSuccessful()) {
- return salBundleService.addBundleMessages(deleteAllFlowGroupsInput);
- }
- return Futures.immediateFuture(null);
- }, service);
-
- /* Push flows and groups via bundle add messages */
- Optional<FlowCapableNode> finalFlowNode = flowNode;
- ListenableFuture<List<RpcResult<AddBundleMessagesOutput>>> addbundlesFuture =
- Futures.transformAsync(deleteAllFlowGroupsFuture, rpcResult -> {
- if (rpcResult.isSuccessful()) {
- LOG.debug("Adding delete all flow/group message is successful for device {}",dpnId);
- return Futures.allAsList(addBundleMessages(finalFlowNode.get(), bundleIdValue,
- nodeIdentity));
- }
- return Futures.immediateFuture(null);
- }, service);
+ ListenableFuture<RpcResult<AddBundleMessagesOutput>> deleteAllFlowGroupsFuture
+ = Futures.transformAsync(openBundle, rpcResult -> {
+ if (rpcResult.isSuccessful()) {
+ return salBundleService.addBundleMessages(deleteAllFlowGroupsInput);
+ }
+ return Futures.immediateFuture(null);
+ }, service);
+
+ /* Push flows and groups via bundle add messages */
+ Optional<FlowCapableNode> finalFlowNode = flowNode;
+ ListenableFuture<List<RpcResult<AddBundleMessagesOutput>>> addbundlesFuture
+ = Futures.transformAsync(deleteAllFlowGroupsFuture, rpcResult -> {
+ if (rpcResult.isSuccessful()) {
+ LOG.debug("Adding delete all flow/group message is successful for device {}", dpnId);
+ return Futures.allAsList(addBundleMessages(finalFlowNode.get(), bundleIdValue,
+ nodeIdentity));
+ }
+ return Futures.immediateFuture(null);
+ }, service);
/* Commit the bundle on the openflow switch */
- ListenableFuture<RpcResult<ControlBundleOutput>> commitBundleFuture = Futures.transformAsync(
- addbundlesFuture, rpcResult -> {
+ ListenableFuture<RpcResult<ControlBundleOutput>> commitBundleFuture
+ = Futures.transformAsync(addbundlesFuture, rpcResult -> {
LOG.debug("Adding bundle messages completed for device {}", dpnId);
return salBundleService.controlBundle(commitBundleInput);
}, service);
- /* Bundles not supported for meters */
- List<Meter> meters = flowNode.get().getMeter() != null ? flowNode.get().getMeter()
- : Collections.emptyList();
- Futures.transformAsync(commitBundleFuture,
- rpcResult -> {
- if (rpcResult.isSuccessful()) {
- for (Meter meter : meters) {
- final KeyedInstanceIdentifier<Meter, MeterKey> meterIdent = nodeIdentity
- .child(Meter.class, meter.key());
- provider.getMeterCommiter().add(meterIdent, meter, nodeIdentity);
- }
+ /* Bundles not supported for meters */
+ Collection<Meter> meters = flowNode.get().nonnullMeter().values();
+ Futures.transformAsync(commitBundleFuture,
+ rpcResult -> {
+ if (rpcResult.isSuccessful()) {
+ for (Meter meter : meters) {
+ final KeyedInstanceIdentifier<Meter, MeterKey> meterIdent = nodeIdentity
+ .child(Meter.class, meter.key());
+ provider.getMeterCommiter().add(meterIdent, meter, nodeIdentity);
}
- return Futures.immediateFuture(null);
- }, service);
- try {
- RpcResult<ControlBundleOutput> bundleFuture = commitBundleFuture.get();
- if (bundleFuture != null && bundleFuture.isSuccessful()) {
- LOG.debug("Completing bundle based reconciliation for device ID:{}", dpnId);
- OF_EVENT_LOG.debug("Bundle Reconciliation Finish, Node: {}", dpnId);
- return true;
- } else {
- LOG.error("commit bundle failed for device {} with error {}", dpnId,
- commitBundleFuture.get().getErrors());
- return false;
}
- } catch (InterruptedException | ExecutionException e) {
- LOG.error("Error while doing bundle based reconciliation for device ID:{}", dpnId, e);
+ return Futures.immediateFuture(null);
+ }, service);
+ try {
+ RpcResult<ControlBundleOutput> bundleFuture = commitBundleFuture.get();
+ if (bundleFuture != null && bundleFuture.isSuccessful()) {
+ reconciliationState.setState(COMPLETED, LocalDateTime.now());
+ LOG.debug("Completing bundle based reconciliation for device ID:{}", dpnId);
+ OF_EVENT_LOG.debug("Bundle Reconciliation Finish, Node: {}", dpnId);
+ return true;
+ } else {
+ reconciliationState.setState(FAILED, LocalDateTime.now());
+ LOG.error("commit bundle failed for device {} with error {}", dpnId,
+ commitBundleFuture.get().getErrors());
return false;
}
+ } catch (InterruptedException | ExecutionException e) {
+ reconciliationState.setState(FAILED, LocalDateTime.now());
+ LOG.error("commit bundle failed for device {} with error ", dpnId, e);
+ return false;
+ } finally {
+ service.shutdown();
}
- LOG.error("FlowNode not present for Datapath ID {}", dpnId);
- return false;
- } finally {
- service.shutdown();
}
+ LOG.error("FlowNode not present for Datapath ID {}", dpnId);
+ return false;
}
}
/* Tables - have to be pushed before groups */
// CHECK if while pushing the update, updateTableInput can be null to emulate a
// table add
- List<TableFeatures> tableList = flowNode.get().getTableFeatures() != null
- ? flowNode.get().getTableFeatures()
- : Collections.<TableFeatures>emptyList();
+ ReconciliationState reconciliationState = new ReconciliationState(
+ STARTED, LocalDateTime.now());
+ //put the dpn info into the map
+ reconciliationStates.put(dpnId.toString(), reconciliationState);
+ LOG.debug("Triggering reconciliation for node {} with state: {}", dpnId, STARTED);
+ Collection<TableFeatures> tableList = flowNode.get().nonnullTableFeatures().values();
for (TableFeatures tableFeaturesItem : tableList) {
TableFeaturesKey tableKey = tableFeaturesItem.key();
KeyedInstanceIdentifier<TableFeatures, TableFeaturesKey> tableFeaturesII = nodeIdentity
}
/* Groups - have to be first */
- List<Group> groups = flowNode.get().getGroup() != null ? flowNode.get().getGroup()
- : Collections.<Group>emptyList();
+ Collection<Group> groups = flowNode.get().nonnullGroup().values();
List<Group> toBeInstalledGroups = new ArrayList<>();
toBeInstalledGroups.addAll(groups);
// new list for suspected groups pointing to ports .. when the ports come up
Group group = iterator.next();
boolean okToInstall = true;
Buckets buckets = group.getBuckets();
- List<Bucket> bucketList = buckets == null ? null : buckets.getBucket();
+ Collection<Bucket> bucketList = buckets == null ? null : buckets.nonnullBucket().values();
if (bucketList == null) {
bucketList = Collections.<Bucket>emptyList();
}
for (Bucket bucket : bucketList) {
- List<Action> actions = bucket.getAction();
+ Collection<Action> actions = bucket.nonnullAction().values();
if (actions == null) {
actions = Collections.<Action>emptyList();
}
}
}
/* Meters */
- List<Meter> meters = flowNode.get().getMeter() != null ? flowNode.get().getMeter()
- : Collections.<Meter>emptyList();
+ Collection<Meter> meters = flowNode.get().nonnullMeter().values();
for (Meter meter : meters) {
final KeyedInstanceIdentifier<Meter, MeterKey> meterIdent = nodeIdentity.child(Meter.class,
meter.key());
awaitGroups(node, groupFutures.values());
/* Flows */
- List<Table> tables = flowNode.get().getTable() != null ? flowNode.get().getTable()
+ Collection<Table> tables = flowNode.get().getTable() != null ? flowNode.get().nonnullTable().values()
: Collections.<Table>emptyList();
int flowCount = 0;
for (Table table : tables) {
final KeyedInstanceIdentifier<Table, TableKey> tableIdent = nodeIdentity.child(Table.class,
table.key());
- List<Flow> flows = table.getFlow() != null ? table.getFlow() : Collections.<Flow>emptyList();
+ Collection<Flow> flows = table.nonnullFlow().values();
flowCount += flows.size();
for (Flow flow : flows) {
final KeyedInstanceIdentifier<Flow, FlowKey> flowIdent = tableIdent.child(Flow.class,
provider.getFlowCommiter().add(flowIdent, flow, nodeIdentity);
}
}
+ reconciliationState.setState(COMPLETED, LocalDateTime.now());
OF_EVENT_LOG.debug("Reconciliation Finish, Node: {}, flow count: {}", dpnId, flowCount);
}
return true;
LOG.debug("Proceeding with deletion of stale-marked Flows on switch {} using Openflow interface",
nodeIdent);
/* Stale-Flows - Stale-marked Flows have to be removed first for safety */
- List<Table> tables = flowNode.get().getTable() != null ? flowNode.get().getTable()
- : Collections.<Table>emptyList();
+ Collection<Table> tables = flowNode.get().nonnullTable().values();
for (Table table : tables) {
final KeyedInstanceIdentifier<Table, TableKey> tableIdent = nodeIdent.child(Table.class,
table.key());
- List<StaleFlow> staleFlows = table.getStaleFlow() != null ? table.getStaleFlow()
- : Collections.<StaleFlow>emptyList();
+ Collection<StaleFlow> staleFlows = table.nonnullStaleFlow().values();
for (StaleFlow staleFlow : staleFlows) {
FlowBuilder flowBuilder = new FlowBuilder(staleFlow);
// before attempting to delete groups - just in case there are references
/* Stale-marked Groups - Can be deleted after flows */
- List<StaleGroup> staleGroups = flowNode.get().getStaleGroup() != null ? flowNode.get().getStaleGroup()
- : Collections.<StaleGroup>emptyList();
+ Collection<StaleGroup> staleGroups = flowNode.get().nonnullStaleGroup().values();
for (StaleGroup staleGroup : staleGroups) {
GroupBuilder groupBuilder = new GroupBuilder(staleGroup);
LOG.debug("Proceeding with deletion of stale-marked Meters for switch {} using Openflow interface",
nodeIdent);
/* Stale-marked Meters - can be deleted anytime - so least priority */
- List<StaleMeter> staleMeters = flowNode.get().getStaleMeter() != null ? flowNode.get().getStaleMeter()
- : Collections.<StaleMeter>emptyList();
+ Collection<StaleMeter> staleMeters = flowNode.get().getStaleMeter().values();
for (StaleMeter staleMeter : staleMeters) {
final BundleId bundleIdValue,
final InstanceIdentifier<FlowCapableNode> nodeIdentity) {
List<ListenableFuture<RpcResult<AddBundleMessagesOutput>>> futureList = new ArrayList<>();
- for (Group group : flowNode.nonnullGroup()) {
+ for (Group group : flowNode.nonnullGroup().values()) {
final KeyedInstanceIdentifier<Group, GroupKey> groupIdent = nodeIdentity.child(Group.class, group.key());
futureList.add(provider.getBundleGroupListener().add(groupIdent, group, nodeIdentity, bundleIdValue));
}
- for (Table table : flowNode.nonnullTable()) {
+ for (Table table : flowNode.nonnullTable().values()) {
final KeyedInstanceIdentifier<Table, TableKey> tableIdent = nodeIdentity.child(Table.class, table.key());
- for (Flow flow : table.nonnullFlow()) {
+ for (Flow flow : table.nonnullFlow().values()) {
final KeyedInstanceIdentifier<Flow, FlowKey> flowIdent = tableIdent.child(Flow.class, flow.key());
futureList.add(provider.getBundleFlowListener().add(flowIdent, flow, nodeIdentity, bundleIdValue));
}
Code Review / openflowplugin.git / blobdiff