*/
package org.opendaylight.controller.md.statistics.manager;
+import java.math.BigInteger;
import java.util.Collection;
+import java.util.Collections;
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.FlowCookieMapping;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowId;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.nodes.node.table.FlowCookieMap;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.nodes.node.table.FlowCookieMapBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.nodes.node.table.FlowCookieMapKey;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableKey;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow;
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.types.rev131026.FlowCookie;
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.Optional;
+
final class FlowStatsTracker extends AbstractListeningStatsTracker<FlowAndStatisticsMapList, FlowStatsEntry> {
- private static final Logger logger = LoggerFactory.getLogger(FlowStatsTracker.class);
+ private static final Logger LOG = LoggerFactory.getLogger(FlowStatsTracker.class);
+ private static final String ALIEN_SYSTEM_FLOW_ID = "#UF$TABLE*";
private final OpendaylightFlowStatisticsService flowStatsService;
private FlowTableStatsTracker flowTableStats;
private int unaccountedFlowsCounter = 1;
+
FlowStatsTracker(final OpendaylightFlowStatisticsService flowStatsService, final FlowCapableContext context) {
super(context);
this.flowStatsService = flowStatsService;
FlowStatisticsDataBuilder flowStatisticsData = new FlowStatisticsDataBuilder();
- FlowBuilder flow = new FlowBuilder(map);
- if(map.getFlowId() != null) {
- flow.setId(new FlowId(map.getFlowId().getValue()));
+ FlowBuilder flowBuilder = new FlowBuilder(map);
+ if (map.getFlowId() != null) {
+ flowBuilder.setId(new FlowId(map.getFlowId().getValue()));
}
- if(map.getFlowId()!= null) {
- flow.setKey(new FlowKey(new FlowId(map.getKey().getFlowId().getValue())));
+ if (map.getFlowId() != null) {
+ flowBuilder.setKey(new FlowKey(new FlowId(map.getKey().getFlowId().getValue())));
}
- Flow flowRule = flow.build();
+ Flow flowRule = flowBuilder.build();
FlowAndStatisticsMapListBuilder stats = new FlowAndStatisticsMapListBuilder();
stats.setByteCount(map.getByteCount());
flowStatisticsData.setFlowStatistics(flowStatistics.build());
- logger.debug("Flow : {}",flowRule.toString());
- logger.debug("Statistics to augment : {}",flowStatistics.build().toString());
+ LOG.debug("Flow : {}",flowRule.toString());
+ LOG.debug("Statistics to augment : {}",flowStatistics.build().toString());
InstanceIdentifier<Table> tableRef = getNodeIdentifierBuilder()
- .augmentation(FlowCapableNode.class).child(Table.class, new TableKey(tableId)).toInstance();
-
- //TODO: Not a good way to do it, need to figure out better way.
- //TODO: major issue in any alternate approach is that flow key is incrementally assigned
- //to the flows stored in data store.
- // Augment same statistics to all the matching masked flow
- Table table= (Table)trans.readConfigurationData(tableRef);
- if(table != null){
- for(Flow existingFlow : table.getFlow()){
- logger.debug("Existing flow in data store : {}",existingFlow.toString());
- if(FlowComparator.flowEquals(flowRule,existingFlow)){
- InstanceIdentifier<Flow> flowRef = getNodeIdentifierBuilder()
- .augmentation(FlowCapableNode.class)
- .child(Table.class, new TableKey(tableId))
- .child(Flow.class,existingFlow.getKey()).toInstance();
- 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
- FlowStatsEntry flowStatsEntry = new FlowStatsEntry(tableId,flow.build());
- trans.putOperationalData(flowRef, flow.build());
- return flowStatsEntry;
- }
- }
- }
-
- table = (Table)trans.readOperationalData(tableRef);
- if(table != null){
- for(Flow existingFlow : table.getFlow()){
- FlowStatisticsData augmentedflowStatisticsData = existingFlow.getAugmentation(FlowStatisticsData.class);
- if(augmentedflowStatisticsData != null){
- FlowBuilder existingOperationalFlow = new FlowBuilder();
- existingOperationalFlow.fieldsFrom(augmentedflowStatisticsData.getFlowStatistics());
- logger.debug("Existing unaccounted flow in operational data store : {}",existingFlow.toString());
- if(FlowComparator.flowEquals(flowRule,existingOperationalFlow.build())){
- InstanceIdentifier<Flow> flowRef = getNodeIdentifierBuilder()
- .augmentation(FlowCapableNode.class)
- .child(Table.class, new TableKey(tableId))
- .child(Flow.class,existingFlow.getKey()).toInstance();
- 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
- FlowStatsEntry flowStatsEntry = new FlowStatsEntry(tableId,flow.build());
- trans.putOperationalData(flowRef, flow.build());
- return flowStatsEntry;
- }
- }
+ .augmentation(FlowCapableNode.class)
+ .child(Table.class, new TableKey(tableId)).toInstance();
+
+ final FlowCookie flowCookie = flowRule.getCookie() != null
+ ? flowRule.getCookie() : new FlowCookie(BigInteger.ZERO);
+ final InstanceIdentifier<FlowCookieMap> flowCookieRef = tableRef
+ .augmentation(FlowCookieMapping.class)
+ .child(FlowCookieMap.class, new FlowCookieMapKey(flowCookie));
+
+ FlowCookieMap cookieMap = (FlowCookieMap) trans.readOperationalData(flowCookieRef);
+
+ /* find flowKey in FlowCookieMap from DataStore/OPERATIONAL */
+ Optional<FlowKey> flowKey = this.getExistFlowKey(flowRule, tableRef, trans, cookieMap);
+ if ( ! flowKey.isPresent()) {
+ /* DataStore/CONFIG For every first statistic needs to be created */
+ flowKey = this.getFlowKeyFromExistFlow(flowRule, tableRef, trans);
+ if ( ! flowKey.isPresent()) {
+ /* Alien flow */
+ flowKey = this.makeAlienFlowKey(flowRule);
}
+ cookieMap = applyNewFlowKey(cookieMap, flowKey, flowCookie);
+ trans.putOperationalData(flowCookieRef, cookieMap);
}
- String flowKey = "#UF$TABLE*"+Short.toString(tableId)+"*"+Integer.toString(this.unaccountedFlowsCounter);
- this.unaccountedFlowsCounter++;
- FlowKey newFlowKey = new FlowKey(new FlowId(flowKey));
- InstanceIdentifier<Flow> flowRef = getNodeIdentifierBuilder().augmentation(FlowCapableNode.class)
- .child(Table.class, new TableKey(tableId))
- .child(Flow.class,newFlowKey).toInstance();
- 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",
- flow.build());
+ InstanceIdentifier<Flow> flowRef = getNodeIdentifierBuilder()
+ .augmentation(FlowCapableNode.class)
+ .child(Table.class, new TableKey(tableId))
+ .child(Flow.class, flowKey.get()).toInstance();
+ flowBuilder.setKey(flowKey.get());
+ flowBuilder.addAugmentation(FlowStatisticsData.class, flowStatisticsData.build());
// Update entry with timestamp of latest response
- flow.setKey(newFlowKey);
- FlowStatsEntry flowStatsEntry = new FlowStatsEntry(tableId,flow.build());
- trans.putOperationalData(flowRef, flow.build());
+ flowBuilder.setKey(flowKey.get());
+ FlowStatsEntry flowStatsEntry = new FlowStatsEntry(tableId, flowBuilder.build());
+ trans.putOperationalData(flowRef, flowBuilder.build());
return flowStatsEntry;
}
// 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());
+ LOG.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());
+ LOG.debug("Send aggregate stats request for flow table {} to node {}", key.getId(), this.getNodeRef());
this.requestAggregateFlows(key);
}
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);
+ LOG.debug("Key {} triggered request for flow {}", e.getKey(), flow);
requestFlow(flow);
} else {
- logger.debug("Ignoring key {}", e.getKey());
+ LOG.debug("Ignoring key {}", e.getKey());
}
}
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);
+ LOG.debug("Key {} triggered remove of Flow from operational space.", key);
trans.removeOperationalData(flow);
}
}
@Override
public void start(final DataBrokerService dbs) {
if (flowStatsService == null) {
- logger.debug("No Flow Statistics service, not subscribing to flows on node {}", getNodeIdentifier());
+ LOG.debug("No Flow Statistics service, not subscribing to flows on node {}", getNodeIdentifier());
return;
}
super.start(dbs);
}
+
+ /* Returns Exist FlowKey from exist FlowCookieMap identified by cookie
+ * and by switch flow identification (priority and match)*/
+ private Optional<FlowKey> getExistFlowKey(final Flow flowRule, final InstanceIdentifier<Table> tableRef,
+ final DataModificationTransaction trans, final FlowCookieMap cookieMap) {
+
+ if (cookieMap != null) {
+ for (FlowId flowId : cookieMap.getFlowIds()) {
+ InstanceIdentifier<Flow> flowIdent = tableRef.child(Flow.class, new FlowKey(flowId));
+ if (flowId.getValue().startsWith(ALIEN_SYSTEM_FLOW_ID)) {
+ LOG.debug("Search for flow in the operational datastore by flowID: {} ", flowIdent);
+ Flow readedFlow = (Flow) trans.readOperationalData(flowIdent);
+ if (FlowComparator.flowEquals(flowRule, readedFlow)) {
+ return Optional.<FlowKey> of(new FlowKey(flowId));
+ }
+ } else {
+ LOG.debug("Search for flow in the configuration datastore by flowID: {} ", flowIdent);
+ Flow readedFlow = (Flow) trans.readConfigurationData(flowIdent);
+ if (FlowComparator.flowEquals(flowRule, readedFlow)) {
+ return Optional.<FlowKey> of(new FlowKey(flowId));
+ }
+ }
+ }
+ LOG.debug("Flow was not found in the datastore. Flow {} ", flowRule);
+ }
+ return Optional.absent();
+ }
+
+ /* Returns FlowKey from existing Flow in DataStore/CONFIGURATIONAL which is identified by cookie
+ * and by switch flow identification (priority and match) */
+ private Optional<FlowKey> getFlowKeyFromExistFlow(final Flow flowRule, final InstanceIdentifier<Table> tableRef,
+ final DataModificationTransaction trans) {
+
+ /* Try to find it in DataSotre/CONFIG */
+ Table table= (Table)trans.readConfigurationData(tableRef);
+ if(table != null) {
+ for(Flow existingFlow : table.getFlow()) {
+ LOG.debug("Existing flow in data store : {}",existingFlow.toString());
+ if(FlowComparator.flowEquals(flowRule,existingFlow)){
+ return Optional.<FlowKey> of(new FlowKey(existingFlow.getId()));
+ }
+ }
+ }
+ return Optional.absent();
+ }
+
+ /* Returns FlowKey which doesn't exist in any DataStore for now */
+ private Optional<FlowKey> makeAlienFlowKey(final Flow flowRule) {
+
+ StringBuilder sBuilder = new StringBuilder(ALIEN_SYSTEM_FLOW_ID)
+ .append(flowRule.getTableId()).append("-").append(this.unaccountedFlowsCounter);
+ this.unaccountedFlowsCounter++;
+ final FlowId flowId = new FlowId(sBuilder.toString());
+ return Optional.<FlowKey> of(new FlowKey(flowId));
+ }
+
+ /* Build new whole FlowCookieMap or add new flowKey */
+ private FlowCookieMap applyNewFlowKey(FlowCookieMap flowCookieMap, final Optional<FlowKey> flowKey,
+ final FlowCookie flowCookie) {
+ if (flowCookieMap != null) {
+ flowCookieMap.getFlowIds().add(flowKey.get().getId());
+ } else {
+ final FlowCookieMapBuilder flowCookieMapBuilder = new FlowCookieMapBuilder();
+ flowCookieMapBuilder.setCookie(flowCookie);
+ flowCookieMapBuilder.setFlowIds(Collections.singletonList(flowKey.get().getId()));
+ flowCookieMap = flowCookieMapBuilder.build();
+ }
+ return flowCookieMap;
+ }
}