*/
package org.opendaylight.controller.md.statistics.manager;
+import java.util.Collection;
+import java.util.Map.Entry;
+
+import org.opendaylight.controller.md.sal.common.api.data.DataChangeEvent;
+import org.opendaylight.controller.sal.binding.api.data.DataBrokerService;
import org.opendaylight.controller.sal.binding.api.data.DataModificationTransaction;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowId;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.flow.and.statistics.map.list.FlowAndStatisticsMapList;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.flow.and.statistics.map.list.FlowAndStatisticsMapListBuilder;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.flow.statistics.FlowStatisticsBuilder;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.transaction.rev131103.TransactionId;
import org.opendaylight.yang.gen.v1.urn.opendaylight.model.statistics.types.rev130925.GenericStatistics;
+import org.opendaylight.yangtools.yang.binding.DataObject;
import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.common.base.Preconditions;
-import com.google.common.util.concurrent.ListenableFuture;
-
-final class FlowStatsTracker extends AbstractStatsTracker<FlowAndStatisticsMapList, FlowStatsEntry> {
+final class FlowStatsTracker extends AbstractListeningStatsTracker<FlowAndStatisticsMapList, FlowStatsEntry> {
private static final Logger logger = LoggerFactory.getLogger(FlowStatsTracker.class);
private final OpendaylightFlowStatisticsService flowStatsService;
+ private FlowTableStatsTracker flowTableStats;
private int unaccountedFlowsCounter = 1;
- FlowStatsTracker(OpendaylightFlowStatisticsService flowStatsService, final FlowCapableContext context, long lifetimeNanos) {
- super(context, lifetimeNanos);
- this.flowStatsService = Preconditions.checkNotNull(flowStatsService);
+ FlowStatsTracker(final OpendaylightFlowStatisticsService flowStatsService, final FlowCapableContext context) {
+ super(context);
+ this.flowStatsService = flowStatsService;
+ }
+ FlowStatsTracker(final OpendaylightFlowStatisticsService flowStatsService, final FlowCapableContext context, final FlowTableStatsTracker flowTableStats) {
+ this(flowStatsService, context);
+ this.flowTableStats = flowTableStats;
}
@Override
- protected void cleanupSingleStat(DataModificationTransaction trans, FlowStatsEntry item) {
+ protected void cleanupSingleStat(final DataModificationTransaction trans, final FlowStatsEntry item) {
InstanceIdentifier<?> flowRef = getNodeIdentifierBuilder()
.augmentation(FlowCapableNode.class)
.child(Table.class, new TableKey(item.getTableId()))
}
@Override
- protected FlowStatsEntry updateSingleStat(DataModificationTransaction trans, FlowAndStatisticsMapList map) {
+ protected FlowStatsEntry updateSingleStat(final DataModificationTransaction trans, final FlowAndStatisticsMapList map) {
short tableId = map.getTableId();
- FlowBuilder flowBuilder = new FlowBuilder();
-
FlowStatisticsDataBuilder flowStatisticsData = new FlowStatisticsDataBuilder();
- FlowBuilder flow = new FlowBuilder();
- flow.setContainerName(map.getContainerName());
- flow.setBufferId(map.getBufferId());
- flow.setCookie(map.getCookie());
- flow.setCookieMask(map.getCookieMask());
- flow.setFlags(map.getFlags());
- flow.setFlowName(map.getFlowName());
- flow.setHardTimeout(map.getHardTimeout());
- if(map.getFlowId() != null)
+ FlowBuilder flow = new FlowBuilder(map);
+ if(map.getFlowId() != null) {
flow.setId(new FlowId(map.getFlowId().getValue()));
- flow.setIdleTimeout(map.getIdleTimeout());
- flow.setInstallHw(map.isInstallHw());
- flow.setInstructions(map.getInstructions());
- if(map.getFlowId()!= null)
+ }
+ if(map.getFlowId()!= null) {
flow.setKey(new FlowKey(new FlowId(map.getKey().getFlowId().getValue())));
- flow.setMatch(map.getMatch());
- flow.setOutGroup(map.getOutGroup());
- flow.setOutPort(map.getOutPort());
- flow.setPriority(map.getPriority());
- flow.setStrict(map.isStrict());
- flow.setTableId(tableId);
+ }
Flow flowRule = flow.build();
flowStatistics.setByteCount(flowStats.getByteCount());
flowStatistics.setPacketCount(flowStats.getPacketCount());
flowStatistics.setDuration(flowStats.getDuration());
- flowStatistics.setContainerName(map.getContainerName());
- flowStatistics.setBufferId(map.getBufferId());
- flowStatistics.setCookie(map.getCookie());
- flowStatistics.setCookieMask(map.getCookieMask());
- flowStatistics.setFlags(map.getFlags());
- flowStatistics.setFlowName(map.getFlowName());
- flowStatistics.setHardTimeout(map.getHardTimeout());
- flowStatistics.setIdleTimeout(map.getIdleTimeout());
- flowStatistics.setInstallHw(map.isInstallHw());
- flowStatistics.setInstructions(map.getInstructions());
- flowStatistics.setMatch(map.getMatch());
- flowStatistics.setOutGroup(map.getOutGroup());
- flowStatistics.setOutPort(map.getOutPort());
- flowStatistics.setPriority(map.getPriority());
- flowStatistics.setStrict(map.isStrict());
- flowStatistics.setTableId(tableId);
flowStatisticsData.setFlowStatistics(flowStatistics.build());
.augmentation(FlowCapableNode.class)
.child(Table.class, new TableKey(tableId))
.child(Flow.class,existingFlow.getKey()).toInstance();
- flowBuilder.setKey(existingFlow.getKey());
- flowBuilder.addAugmentation(FlowStatisticsData.class, flowStatisticsData.build());
+ flow.setKey(existingFlow.getKey());
+ flow.addAugmentation(FlowStatisticsData.class, flowStatisticsData.build());
logger.debug("Found matching flow in the datastore, augmenting statistics");
// Update entry with timestamp of latest response
- flow.setKey(existingFlow.getKey());
FlowStatsEntry flowStatsEntry = new FlowStatsEntry(tableId,flow.build());
- trans.putOperationalData(flowRef, flowBuilder.build());
+ trans.putOperationalData(flowRef, flow.build());
return flowStatsEntry;
}
}
.augmentation(FlowCapableNode.class)
.child(Table.class, new TableKey(tableId))
.child(Flow.class,existingFlow.getKey()).toInstance();
- flowBuilder.setKey(existingFlow.getKey());
- flowBuilder.addAugmentation(FlowStatisticsData.class, flowStatisticsData.build());
+ flow.setKey(existingFlow.getKey());
+ flow.addAugmentation(FlowStatisticsData.class, flowStatisticsData.build());
logger.debug("Found matching unaccounted flow in the operational datastore, augmenting statistics");
// Update entry with timestamp of latest response
- flow.setKey(existingFlow.getKey());
FlowStatsEntry flowStatsEntry = new FlowStatsEntry(tableId,flow.build());
- trans.putOperationalData(flowRef, flowBuilder.build());
+ trans.putOperationalData(flowRef, flow.build());
return flowStatsEntry;
}
}
InstanceIdentifier<Flow> flowRef = getNodeIdentifierBuilder().augmentation(FlowCapableNode.class)
.child(Table.class, new TableKey(tableId))
.child(Flow.class,newFlowKey).toInstance();
- flowBuilder.setKey(newFlowKey);
- flowBuilder.addAugmentation(FlowStatisticsData.class, flowStatisticsData.build());
+ flow.setKey(newFlowKey);
+ flow.addAugmentation(FlowStatisticsData.class, flowStatisticsData.build());
logger.debug("Flow {} is not present in config data store, augmenting statistics as an unaccounted flow",
- flowBuilder.build());
+ flow.build());
// Update entry with timestamp of latest response
flow.setKey(newFlowKey);
FlowStatsEntry flowStatsEntry = new FlowStatsEntry(tableId,flow.build());
- trans.putOperationalData(flowRef, flowBuilder.build());
+ trans.putOperationalData(flowRef, flow.build());
return flowStatsEntry;
}
- public ListenableFuture<TransactionId> requestAllFlowsAllTables() {
- final GetAllFlowsStatisticsFromAllFlowTablesInputBuilder input = new GetAllFlowsStatisticsFromAllFlowTablesInputBuilder();
- input.setNode(getNodeRef());
+ @Override
+ protected InstanceIdentifier<?> listenPath() {
+ return getNodeIdentifierBuilder().augmentation(FlowCapableNode.class).child(Table.class).child(Flow.class).build();
+ }
+
+ @Override
+ protected String statName() {
+ return "Flow";
+ }
+
+ @Override
+ public void request() {
+ // FIXME: it does not make sense to trigger this before sendAllFlowTablesStatisticsRequest()
+ // comes back -- we do not have any tables anyway.
+ final Collection<TableKey> tables = flowTableStats.getTables();
+ logger.debug("Node {} supports {} table(s)", this.getNodeRef(), tables.size());
+ for (final TableKey key : tables) {
+ logger.debug("Send aggregate stats request for flow table {} to node {}", key.getId(), this.getNodeRef());
+ this.requestAggregateFlows(key);
+ }
+
+ this.requestAllFlowsAllTables();
+
+ }
+ public void requestAllFlowsAllTables() {
+ if (flowStatsService != null) {
+ final GetAllFlowsStatisticsFromAllFlowTablesInputBuilder input = new GetAllFlowsStatisticsFromAllFlowTablesInputBuilder();
+ input.setNode(getNodeRef());
+
+ requestHelper(flowStatsService.getAllFlowsStatisticsFromAllFlowTables(input.build()));
+ }
+ }
+
+ public void requestAggregateFlows(final TableKey key) {
+ if (flowStatsService != null) {
+ GetAggregateFlowStatisticsFromFlowTableForAllFlowsInputBuilder input =
+ new GetAggregateFlowStatisticsFromFlowTableForAllFlowsInputBuilder();
- return requestHelper(flowStatsService.getAllFlowsStatisticsFromAllFlowTables(input.build()));
+ input.setNode(getNodeRef());
+ input.setTableId(new org.opendaylight.yang.gen.v1.urn.opendaylight.table.types.rev131026.TableId(key.getId()));
+ requestHelper(flowStatsService.getAggregateFlowStatisticsFromFlowTableForAllFlows(input.build()));
+ }
}
- public ListenableFuture<TransactionId> requestAggregateFlows(final TableKey key) {
- GetAggregateFlowStatisticsFromFlowTableForAllFlowsInputBuilder input =
- new GetAggregateFlowStatisticsFromFlowTableForAllFlowsInputBuilder();
+ public void requestFlow(final Flow flow) {
+ if (flowStatsService != null) {
+ final GetFlowStatisticsFromFlowTableInputBuilder input =
+ new GetFlowStatisticsFromFlowTableInputBuilder(flow);
+ input.setNode(getNodeRef());
- input.setNode(getNodeRef());
- input.setTableId(new org.opendaylight.yang.gen.v1.urn.opendaylight.table.types.rev131026.TableId(key.getId()));
- return requestHelper(flowStatsService.getAggregateFlowStatisticsFromFlowTableForAllFlows(input.build()));
+ requestHelper(flowStatsService.getFlowStatisticsFromFlowTable(input.build()));
+ }
}
- public ListenableFuture<TransactionId> requestFlow(final Flow flow) {
- final GetFlowStatisticsFromFlowTableInputBuilder input =
- new GetFlowStatisticsFromFlowTableInputBuilder(flow);
- input.setNode(getNodeRef());
+ @Override
+ public void onDataChanged(final DataChangeEvent<InstanceIdentifier<?>, DataObject> change) {
+ for (Entry<InstanceIdentifier<?>, DataObject> e : change.getCreatedConfigurationData().entrySet()) {
+ if (Flow.class.equals(e.getKey().getTargetType())) {
+ final Flow flow = (Flow) e.getValue();
+ logger.debug("Key {} triggered request for flow {}", e.getKey(), flow);
+ requestFlow(flow);
+ } else {
+ logger.debug("Ignoring key {}", e.getKey());
+ }
+ }
+
+ final DataModificationTransaction trans = startTransaction();
+ for (InstanceIdentifier<?> key : change.getRemovedConfigurationData()) {
+ if (Flow.class.equals(key.getTargetType())) {
+ @SuppressWarnings("unchecked")
+ final InstanceIdentifier<Flow> flow = (InstanceIdentifier<Flow>)key;
+ logger.debug("Key {} triggered remove of Flow from operational space.", key);
+ trans.removeOperationalData(flow);
+ }
+ }
+ trans.commit();
+ }
+
+ @Override
+ public void start(final DataBrokerService dbs) {
+ if (flowStatsService == null) {
+ logger.debug("No Flow Statistics service, not subscribing to flows on node {}", getNodeIdentifier());
+ return;
+ }
- return requestHelper(flowStatsService.getFlowStatisticsFromFlowTable(input.build()));
+ super.start(dbs);
}
}
Code Review / controller.git / blobdiff