package org.opendaylight.openflowplugin.applications.statistics.manager;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.flow.node.SwitchFeatures;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.OpendaylightInventoryListener;
import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
* @param keyIdent
*/
void disconnectFlowCapableNode(InstanceIdentifier<Node> keyIdent);
+
+ /**
+ * Method returns if *this* instance of the stats-manager is owner of the node
+ * @param node Given Node
+ * @return true if owner, else false
+ */
+ boolean isFlowCapableNodeOwner(NodeId node);
}
import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
import org.opendaylight.controller.sal.binding.api.NotificationProviderService;
import org.opendaylight.openflowplugin.applications.statistics.manager.StatListeningCommiter;
+import org.opendaylight.openflowplugin.applications.statistics.manager.StatNodeRegistration;
import org.opendaylight.openflowplugin.applications.statistics.manager.StatisticsManager;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
import org.opendaylight.yangtools.concepts.ListenerRegistration;
import org.opendaylight.yangtools.yang.binding.DataObject;
private final DataBroker dataBroker;
+ protected final StatNodeRegistration nodeRegistrationManager;
+
private ReadOnlyTransaction currentReadTx;
private volatile boolean currentReadTxStale;
/* Constructor has to make a registration */
public StatAbstractListenCommit(final StatisticsManager manager, final DataBroker db,
- final NotificationProviderService nps, final Class<T> clazz) {
- super(manager,nps);
+ final NotificationProviderService nps, final Class<T> clazz, final StatNodeRegistration nodeRegistrationManager) {
+ super(manager,nps, nodeRegistrationManager);
this.clazz = Preconditions.checkNotNull(clazz, "Referenced Class can not be null");
Preconditions.checkArgument(db != null, "DataBroker can not be null!");
listenerRegistration = db.registerDataChangeListener(LogicalDatastoreType.CONFIGURATION,
getWildCardedRegistrationPath(), this, DataChangeScope.BASE);
this.dataBroker = db;
+ this.nodeRegistrationManager = nodeRegistrationManager;
}
/**
return Optional.absent();
}
+
}
import java.util.concurrent.TimeoutException;
import org.opendaylight.controller.sal.binding.api.NotificationProviderService;
+import org.opendaylight.openflowplugin.applications.statistics.manager.StatNodeRegistration;
import org.opendaylight.openflowplugin.applications.statistics.manager.StatNotifyCommiter;
import org.opendaylight.openflowplugin.applications.statistics.manager.StatRpcMsgManager.TransactionCacheContainer;
import org.opendaylight.openflowplugin.applications.statistics.manager.StatisticsManager;
protected final StatisticsManager manager;
private ListenerRegistration<NotificationListener> notifyListenerRegistration;
+ protected final StatNodeRegistration nodeRegistrationManager;
+
public StatAbstractNotifyCommit(final StatisticsManager manager,
- final NotificationProviderService nps) {
+ final NotificationProviderService nps,
+ final StatNodeRegistration nodeRegistrationManager) {
Preconditions.checkArgument(nps != null, "NotificationProviderService can not be null!");
this.manager = Preconditions.checkNotNull(manager, "StatisticManager can not be null!");
notifyListenerRegistration = nps.registerNotificationListener(getStatNotificationListener());
+ this.nodeRegistrationManager = nodeRegistrationManager;
}
@Override
import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
import org.opendaylight.controller.sal.binding.api.NotificationProviderService;
+import org.opendaylight.openflowplugin.applications.statistics.manager.StatNodeRegistration;
import org.opendaylight.openflowplugin.applications.statistics.manager.StatRpcMsgManager.TransactionCacheContainer;
import org.opendaylight.openflowplugin.applications.statistics.manager.StatisticsManager;
import org.opendaylight.openflowplugin.applications.statistics.manager.StatisticsManager.StatDataStoreOperation;
private final AtomicInteger unaccountedFlowsCounter = new AtomicInteger(0);
public StatListenCommitFlow (final StatisticsManager manager, final DataBroker db,
- final NotificationProviderService nps){
- super(manager, db, nps, Flow.class);
+ final NotificationProviderService nps,
+ final StatNodeRegistration nrm){
+ super(manager, db, nps, Flow.class,nrm);
}
@Override
if (( ! inputObj.isPresent()) || ( ! (inputObj.get() instanceof Table))) {
return;
}
+
+ if(!nodeRegistrationManager.isFlowCapableNodeOwner(nodeId)) { return; }
+
final Table table = (Table) inputObj.get();
final List<? extends TransactionAware> cacheNotifs = txContainer.get().getNotifications();
for (final TransactionAware notif : cacheNotifs) {
if (( ! txContainer.isPresent()) || txContainer.get().getNotifications() == null) {
return;
}
+ if(!nodeRegistrationManager.isFlowCapableNodeOwner(nodeId)) { return; }
+
final List<FlowAndStatisticsMapList> flowStats = new ArrayList<FlowAndStatisticsMapList>(10);
final InstanceIdentifier<Node> nodeIdent = InstanceIdentifier.create(Nodes.class)
.child(Node.class, new NodeKey(nodeId));
import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
import org.opendaylight.controller.sal.binding.api.NotificationProviderService;
+import org.opendaylight.openflowplugin.applications.statistics.manager.StatNodeRegistration;
import org.opendaylight.openflowplugin.applications.statistics.manager.StatPermCollector.StatCapabTypes;
import org.opendaylight.openflowplugin.applications.statistics.manager.StatRpcMsgManager.TransactionCacheContainer;
import org.opendaylight.openflowplugin.applications.statistics.manager.StatisticsManager;
private static final Logger LOG = LoggerFactory.getLogger(StatListenCommitMeter.class);
public StatListenCommitGroup(final StatisticsManager manager, final DataBroker db,
- final NotificationProviderService nps) {
- super(manager, db, nps, Group.class);
+ final NotificationProviderService nps,
+ final StatNodeRegistration nrm) {
+ super(manager, db, nps, Group.class,nrm);
}
@Override
if ( ! isTransactionCacheContainerValid(txContainer)) {
return;
}
+
+ if(!nodeRegistrationManager.isFlowCapableNodeOwner(nodeId)) { return; }
+
/* Prepare List actual Groups and not updated Groups will be removed */
final List<Group> existGroups = fNode.get().getGroup() != null
? fNode.get().getGroup() : Collections.<Group> emptyList();
return;
}
+ if(!nodeRegistrationManager.isFlowCapableNodeOwner(nodeId)) { return; }
+
final InstanceIdentifier<Node> nodeIdent = InstanceIdentifier
.create(Nodes.class).child(Node.class, new NodeKey(nodeId));
if ( ! isTransactionCacheContainerValid(txContainer)) {
return;
}
+
+ if(!nodeRegistrationManager.isFlowCapableNodeOwner(nodeId)) { return; }
+
final List<? extends TransactionAware> cacheNotifs = txContainer.get().getNotifications();
Optional<Group> notifGroup = Optional.absent();
import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
import org.opendaylight.controller.sal.binding.api.NotificationProviderService;
+import org.opendaylight.openflowplugin.applications.statistics.manager.StatNodeRegistration;
import org.opendaylight.openflowplugin.applications.statistics.manager.StatPermCollector.StatCapabTypes;
import org.opendaylight.openflowplugin.applications.statistics.manager.StatRpcMsgManager.TransactionCacheContainer;
import org.opendaylight.openflowplugin.applications.statistics.manager.StatisticsManager;
private static final Logger LOG = LoggerFactory.getLogger(StatListenCommitMeter.class);
public StatListenCommitMeter(final StatisticsManager manager, final DataBroker db,
- final NotificationProviderService nps) {
- super(manager, db, nps, Meter.class);
+ final NotificationProviderService nps,
+ final StatNodeRegistration nrm) {
+ super(manager, db, nps, Meter.class,nrm);
}
@Override
if ( ! isTransactionCacheContainerValid(txContainer)) {
return;
}
+
+ if(!nodeRegistrationManager.isFlowCapableNodeOwner(nodeId)) { return; }
+
/* Prepare List actual Meters and not updated Meters will be removed */
final List<Meter> existMeters = fNode.get().getMeter() != null
? fNode.get().getMeter() : Collections.<Meter> emptyList();
return;
}
+ if(!nodeRegistrationManager.isFlowCapableNodeOwner(nodeId)) { return; }
+
final InstanceIdentifier<Node> nodeIdent = InstanceIdentifier
.create(Nodes.class).child(Node.class, new NodeKey(nodeId));
if ( ! isTransactionCacheContainerValid(txContainer)) {
return;
}
+ if(!nodeRegistrationManager.isFlowCapableNodeOwner(nodeId)) { return; }
+
final List<? extends TransactionAware> cacheNotifs = txContainer.get().getNotifications();
Optional<Meter> notifMeter = Optional.absent();
import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
import org.opendaylight.controller.sal.binding.api.NotificationProviderService;
+import org.opendaylight.openflowplugin.applications.statistics.manager.StatNodeRegistration;
import org.opendaylight.openflowplugin.applications.statistics.manager.StatRpcMsgManager.TransactionCacheContainer;
import org.opendaylight.openflowplugin.applications.statistics.manager.StatisticsManager;
import org.opendaylight.openflowplugin.applications.statistics.manager.StatisticsManager.StatDataStoreOperation;
private static final Logger LOG = LoggerFactory.getLogger(StatListenCommitQueue.class);
public StatListenCommitQueue(final StatisticsManager manager, final DataBroker db,
- final NotificationProviderService nps) {
- super(manager, db, nps, Queue.class);
+ final NotificationProviderService nps,
+ final StatNodeRegistration nrm) {
+ super(manager, db, nps, Queue.class,nrm);
}
@Override
if ( ! isTransactionCacheContainerValid(txContainer)) {
return;
}
+
+ if(!nodeRegistrationManager.isFlowCapableNodeOwner(nodeId)) { return; }
+
/* Prepare List actual Queues and not updated Queues will be removed */
final List<NodeConnector> existConnectors = fNode.get().getNodeConnector() != null
? fNode.get().getNodeConnector() : Collections.<NodeConnector> emptyList();
final Map<QueueKey, NodeConnectorKey> existQueueKeys = new HashMap<>();
for (final NodeConnector connect : existConnectors) {
- final List<Queue> listQueues = connect.getAugmentation(FlowCapableNodeConnector.class).getQueue();
- if (listQueues != null) {
- for (final Queue queue : listQueues) {
- existQueueKeys.put(queue.getKey(), connect.getKey());
+ if(connect.getAugmentation(FlowCapableNodeConnector.class) != null){
+ final List<Queue> listQueues = connect.getAugmentation(FlowCapableNodeConnector.class).getQueue();
+ if (listQueues != null) {
+ for (final Queue queue : listQueues) {
+ existQueueKeys.put(queue.getKey(), connect.getKey());
+ }
}
}
}
import com.google.common.base.Optional;
import com.google.common.base.Preconditions;
+
+import java.math.BigInteger;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
import org.opendaylight.controller.md.sal.binding.api.DataBroker;
import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
import org.opendaylight.controller.md.sal.common.api.clustering.Entity;
import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
import org.opendaylight.yangtools.concepts.ListenerRegistration;
import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
+import org.opendaylight.yangtools.yang.common.QName;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifierWithPredicates;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
private static final Logger LOG = LoggerFactory.getLogger(StatNodeRegistrationImpl.class);
+ private static final QName ENTITY_QNAME =
+ org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.core.general.entity.rev150820.Entity.QNAME;
+ private static final QName ENTITY_NAME = QName.create(ENTITY_QNAME, "name");
+
private final StatisticsManager manager;
private ListenerRegistration<?> notifListenerRegistration;
//private DataBroker db;
private EntityOwnershipListenerRegistration ofListenerRegistration = null;
+ private final Map<NodeId, Boolean> nodeOwnershipState = new ConcurrentHashMap();
+
public StatNodeRegistrationImpl(final StatisticsManager manager, final DataBroker db,
final NotificationProviderService notificationService) {
manager.disconnectedNodeUnregistration(nodeIdent);
}
- private boolean preConfigurationCheck(final InstanceIdentifier<Node> nodeIdent) {
- Preconditions.checkNotNull(nodeIdent, "Node Instance Identifier can not be null!");
- NodeId nodeId = InstanceIdentifier.keyOf(nodeIdent).getId();
- final Entity entity = new Entity("openflow", nodeId.getValue());
+ private boolean preConfigurationCheck(final NodeId nodeId) {
+ Preconditions.checkNotNull(nodeId, "Node Instance Identifier can not be null!");
+ final Entity entity = getEntity(nodeId);
EntityOwnershipService ownershipService = manager.getOwnershipService();
if(ownershipService == null) {
LOG.error("preConfigurationCheck: EntityOwnershipService is null");
nodeRefIdent.firstIdentifierOf(Node.class);
if (nodeIdent != null) {
LOG.debug("Received onNodeRemoved for node:{} ", nodeIdent);
+ removeOwnership(InstanceIdentifier.keyOf(nodeIdent).getId());
disconnectFlowCapableNode(nodeIdent);
}
}
connectFlowCapableNode(swichFeaturesIdent, switchFeatures, nodeIdent);
//Send group/meter request to get addition details not present in switch feature response.
- if(preConfigurationCheck(nodeIdent)) {
+ NodeId nodeId = InstanceIdentifier.keyOf(nodeIdent).getId();
+ boolean ownershipState = preConfigurationCheck(nodeId);
+ setNodeOwnership(nodeId, ownershipState);
+ if(ownershipState) {
LOG.info("onNodeUpdated: Send group/meter feature request to the device {}",nodeIdent);
manager.getRpcMsgManager().getGroupFeaturesStat(nodeRef);
manager.getRpcMsgManager().getMeterFeaturesStat(nodeRef);
}
}
+ @Override
+ public boolean isFlowCapableNodeOwner(NodeId node) {
+ if(this.nodeOwnershipState.containsKey(node)){
+ return this.nodeOwnershipState.get(node).booleanValue();
+ }
+ return false;
+ }
+
+
@Override
public void ownershipChanged(EntityOwnershipChange ownershipChange) {
- //I believe the only scenario we need to handle here is
- // isOwner=false && hasOwner=false. E.g switch is connected to only
- // one controller and that goes down, all other controller will get
- // notification about ownership change with the flag set as above.
- // In this scenario, topology manager should remove the node from
- // operational data store, so no explict action is required here.
+
+ YangInstanceIdentifier yId = ownershipChange.getEntity().getId();
+ NodeIdentifierWithPredicates niWPredicates = (NodeIdentifierWithPredicates)yId.getLastPathArgument();
+ Map<QName, Object> keyValMap = niWPredicates.getKeyValues();
+ String nodeIdStr = (String)(keyValMap.get(ENTITY_NAME));
+ BigInteger dpId = new BigInteger(nodeIdStr.split(":")[1]);
+ NodeId nodeId = new NodeId(nodeIdStr);
+ setNodeOwnership(nodeId, ownershipChange.isOwner());
+ }
+
+ private void setNodeOwnership(NodeId node, boolean ownership) {
+ this.nodeOwnershipState.put(node,ownership);
+ }
+
+ private void removeOwnership(NodeId node) {
+ this.nodeOwnershipState.remove(node);
+ }
+
+ private Entity getEntity(NodeId nodeId) {
+ return new Entity("openflow", nodeId.getValue());
}
}
import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
import org.opendaylight.controller.sal.binding.api.NotificationProviderService;
+import org.opendaylight.openflowplugin.applications.statistics.manager.StatNodeRegistration;
import org.opendaylight.openflowplugin.applications.statistics.manager.StatRpcMsgManager.TransactionCacheContainer;
import org.opendaylight.openflowplugin.applications.statistics.manager.StatisticsManager;
import org.opendaylight.openflowplugin.applications.statistics.manager.StatisticsManager.StatDataStoreOperation;
private static final Logger LOG = LoggerFactory.getLogger(StatNotifyCommitPort.class);
public StatNotifyCommitPort(final StatisticsManager manager,
- final NotificationProviderService nps) {
- super(manager, nps);
+ final NotificationProviderService nps,
+ final StatNodeRegistration nrm) {
+ super(manager, nps,nrm);
}
@Override
if (( ! txContainer.isPresent()) || txContainer.get().getNotifications() == null) {
return;
}
+
+ if(!nodeRegistrationManager.isFlowCapableNodeOwner(nodeId)) { return; }
+
final List<NodeConnectorStatisticsAndPortNumberMap> portStats =
new ArrayList<NodeConnectorStatisticsAndPortNumberMap>(10);
final List<? extends TransactionAware> cachedNotifs = txContainer.get().getNotifications();
import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
import org.opendaylight.controller.sal.binding.api.NotificationProviderService;
+import org.opendaylight.openflowplugin.applications.statistics.manager.StatNodeRegistration;
import org.opendaylight.openflowplugin.applications.statistics.manager.StatRpcMsgManager.TransactionCacheContainer;
import org.opendaylight.openflowplugin.applications.statistics.manager.StatisticsManager;
import org.opendaylight.openflowplugin.applications.statistics.manager.StatisticsManager.StatDataStoreOperation;
private static final Logger LOG = LoggerFactory.getLogger(StatNotifyCommitTable.class);
public StatNotifyCommitTable(final StatisticsManager manager,
- final NotificationProviderService nps) {
- super(manager, nps);
+ final NotificationProviderService nps,
+ final StatNodeRegistration nrm) {
+ super(manager, nps, nrm);
}
@Override
if (( ! txContainer.isPresent()) || txContainer.get().getNodeId() == null) {
return;
}
+
+ if(!nodeRegistrationManager.isFlowCapableNodeOwner(nodeId)) { return; }
+
final List<? extends TransactionAware> cachedNotifs = txContainer.get().getNotifications();
for (final TransactionAware notif : cachedNotifs) {
if (notif instanceof FlowTableStatisticsUpdate) {
LOG.trace("STAT-MANAGER-collecting FLOW-STATS-ALL_FLOWS for NodeRef {}", actualNodeRef);
setActualTransactionId(manager.getRpcMsgManager().getAllFlowsStat(actualNodeRef).get());
waitingForNotification();
- LOG.trace("STAT-MANAGER-collecting FLOW-AGGREGATE-STATS for NodeRef {}", actualNodeRef);
+ /*LOG.trace("STAT-MANAGER-collecting FLOW-AGGREGATE-STATS for NodeRef {}", actualNodeRef);
for (short i = 0; i < maxTables; i++) {
final TableId tableId = new TableId(i);
manager.getRpcMsgManager().getAggregateFlowStat(actualNodeRef, tableId);
- }
+ }*/
break;
default:
/* Exception for programmers in implementation cycle */
}
private boolean isThisInstanceNodeOwner(NodeId nodeId) {
- final Entity deviceEntity = new Entity("openflow",nodeId.getValue());
- if(manager.getOwnershipService().isCandidateRegistered(deviceEntity)) {
- Optional<EntityOwnershipState> deviceOwnershipState = manager.getOwnershipService()
- .getOwnershipState(deviceEntity);
-
- if(deviceOwnershipState.isPresent()) {
- return deviceOwnershipState.get().isOwner();
- } else {
- LOG.error("Node {} is connected to the controller but ownership state is missing.");
- }
- } else {
- LOG.warn("Node {} is connected to the controller but it did not" +
- "registered for the device ownership.",nodeId);
- }
- return false;
+ return manager.getNodeRegistrator().isFlowCapableNodeOwner(nodeId);
}
private class StatNodeInfoHolder {
private AtomicInteger numNodesBeingCollected = new AtomicInteger(0);
- private final DataBroker dataBroker;
+ private final DataBroker dataBroker;
private final ExecutorService statRpcMsgManagerExecutor;
private final ExecutorService statDataStoreOperationServ;
private EntityOwnershipService ownershipService;
rpcMsgManager = new StatRpcMsgManagerImpl(this, rpcRegistry, statManagerConfig.getMaxNodesForCollector());
statCollectors = Collections.emptyList();
nodeRegistrator = new StatNodeRegistrationImpl(this, dataBroker, notifService);
- flowListeningCommiter = new StatListenCommitFlow(this, dataBroker, notifService);
- meterListeningCommiter = new StatListenCommitMeter(this, dataBroker, notifService);
- groupListeningCommiter = new StatListenCommitGroup(this, dataBroker, notifService);
- tableNotifCommiter = new StatNotifyCommitTable(this, notifService);
- portNotifyCommiter = new StatNotifyCommitPort(this, notifService);
- queueNotifyCommiter = new StatListenCommitQueue(this, dataBroker, notifService);
+ flowListeningCommiter = new StatListenCommitFlow(this, dataBroker, notifService, nodeRegistrator);
+ meterListeningCommiter = new StatListenCommitMeter(this, dataBroker, notifService, nodeRegistrator);
+ groupListeningCommiter = new StatListenCommitGroup(this, dataBroker, notifService, nodeRegistrator);
+ tableNotifCommiter = new StatNotifyCommitTable(this, notifService, nodeRegistrator);
+ portNotifyCommiter = new StatNotifyCommitPort(this, notifService, nodeRegistrator);
+ queueNotifyCommiter = new StatListenCommitQueue(this, dataBroker, notifService, nodeRegistrator);
statRpcMsgManagerExecutor.execute(rpcMsgManager);
statDataStoreOperationServ.execute(this);
public EntityOwnershipService getOwnershipService() {
return this.ownershipService;
}
+
}
import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
import org.opendaylight.controller.sal.binding.api.NotificationProviderService;
+import org.opendaylight.openflowplugin.applications.statistics.manager.StatNodeRegistration;
import org.opendaylight.openflowplugin.applications.statistics.manager.StatisticsManager;
import org.opendaylight.yangtools.yang.binding.DataObject;
import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
@Mock
private NotificationListener mockNotificationListener;
+ @Mock
+ private StatNodeRegistration statsNodeRegistration;
+
+
@SuppressWarnings("rawtypes")
private StatAbstractListenCommit statCommit;
MockitoAnnotations.initMocks(this);
statCommit = new StatAbstractListenCommit(mockStatisticsManager, mockDataBroker,
- mockNotificationProviderService, DataObject.class) {
+ mockNotificationProviderService, DataObject.class, statsNodeRegistration) {
@Override
protected InstanceIdentifier getWildCardedRegistrationPath() {
return InstanceIdentifier.create(DataObject.class);
import org.opendaylight.controller.md.sal.binding.api.ReadWriteTransaction;
import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
import org.opendaylight.controller.sal.binding.api.NotificationProviderService;
+import org.opendaylight.openflowplugin.applications.statistics.manager.StatNodeRegistration;
import org.opendaylight.openflowplugin.applications.statistics.manager.StatisticsManager;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNodeBuilder;
@Mock
private DataBroker mockDataBroker;
+ @Mock
+ private StatNodeRegistration statsNodeRegistration;
+
private StatListenCommitFlow statCommitFlow;
private TableKey tableKey = new TableKey((short) 12);
public void init() {
MockitoAnnotations.initMocks(this);
statCommitFlow = new StatListenCommitFlow(mockStatisticsManager, mockDataBroker,
- mockNotificationProviderService);
+ mockNotificationProviderService, statsNodeRegistration);
}
@Test
hsPool.purge();
conductorState = CONDUCTOR_STATE.WORKING;
QueueKeeperFactory.plugQueue(queueProcessor, queue);
- OFSessionUtil.setRole(sessionContext);
}
/**
package org.opendaylight.openflowplugin.openflow.md.core.role;
import java.math.BigInteger;
+import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import org.opendaylight.controller.md.sal.common.api.clustering.CandidateAlreadyRegisteredException;
import com.google.common.base.Optional;
entsession = new ConcurrentHashMap<>();
entRegistrationMap = new ConcurrentHashMap<>();
ThreadPoolLoggingExecutor delegate = new ThreadPoolLoggingExecutor(
- 1, 5, 0, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(5), "ofEntity");
+ 20, 20, 0, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), "ofEntity");
pool = MoreExecutors.listeningDecorator(delegate);
}
LOG.info("onDeviceOwnershipChanged: Set controller as a SLAVE controller because " +
"it's not the OWNER of the {}", entity);
entsession.get(entity).getOfSwitch().setEntityOwnership(newRole==OfpRole.BECOMEMASTER);
+ setSlaveRole(sessionContext);
sendNodeAddedNotification(entsession.get(entity));
if(ownershipChange.wasOwner()) {
deregisterRoutedRPCForSwitch(entsession.get(entity));
Futures.addCallback(JdkFutureAdapters.listenInPoolThread(roleReply), new FutureCallback<RpcResult<RoleRequestOutput>>() {
@Override
public void onSuccess(RpcResult<RoleRequestOutput> input) {
- result.set(input.getResult().getGenerationId());
+ if(input != null && input.getResult() != null) {
+ result.set(input.getResult().getGenerationId());
+ }
}
@Override
public void onFailure(Throwable t) {