opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/FlowStatsTracker.java

   1 /*
   2  * Copyright IBM Corporation, 2013.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.controller.md.statistics.manager;
   9
  10 import java.util.Collection;
  11 import java.util.Map.Entry;
  12
  13 import org.opendaylight.controller.md.sal.common.api.data.DataChangeEvent;
  14 import org.opendaylight.controller.sal.binding.api.data.DataBrokerService;
  15 import org.opendaylight.controller.sal.binding.api.data.DataModificationTransaction;
  16 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
  17 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowId;
  18 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
  19 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableKey;
  20 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow;
  21 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowBuilder;
  22 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowKey;
  23 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.FlowStatisticsData;
  24 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.FlowStatisticsDataBuilder;
  25 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.GetAggregateFlowStatisticsFromFlowTableForAllFlowsInputBuilder;
  26 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.GetAllFlowsStatisticsFromAllFlowTablesInputBuilder;
  27 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.GetFlowStatisticsFromFlowTableInputBuilder;
  28 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.OpendaylightFlowStatisticsService;
  29 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.flow.and.statistics.map.list.FlowAndStatisticsMapList;
  30 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.flow.and.statistics.map.list.FlowAndStatisticsMapListBuilder;
  31 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.flow.statistics.FlowStatisticsBuilder;
  32 import org.opendaylight.yang.gen.v1.urn.opendaylight.model.statistics.types.rev130925.GenericStatistics;
  33 import org.opendaylight.yangtools.yang.binding.DataObject;
  34 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  35 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
  36 import org.slf4j.Logger;
  37 import org.slf4j.LoggerFactory;
  38
  39 final class FlowStatsTracker extends AbstractListeningStatsTracker<FlowAndStatisticsMapList, FlowStatsEntry> {
  40     private static final Logger logger = LoggerFactory.getLogger(FlowStatsTracker.class);
  41     private final OpendaylightFlowStatisticsService flowStatsService;
  42     private FlowTableStatsTracker flowTableStats;
  43     private int unaccountedFlowsCounter = 1;
  44
  45     FlowStatsTracker(final OpendaylightFlowStatisticsService flowStatsService, final FlowCapableContext context) {
  46         super(context);
  47         this.flowStatsService = flowStatsService;
  48     }
  49     FlowStatsTracker(final OpendaylightFlowStatisticsService flowStatsService, final FlowCapableContext context, final FlowTableStatsTracker flowTableStats) {
  50         this(flowStatsService, context);
  51         this.flowTableStats = flowTableStats;
  52     }
  53
  54     @Override
  55     protected void cleanupSingleStat(final DataModificationTransaction trans, final FlowStatsEntry item) {
  56         KeyedInstanceIdentifier<Flow, FlowKey> flowRef = getNodeIdentifier()
  57                 .augmentation(FlowCapableNode.class)
  58                 .child(Table.class, new TableKey(item.getTableId()))
  59                 .child(Flow.class, item.getFlow().getKey());
  60         trans.removeOperationalData(flowRef);
  61     }
  62
  63     @Override
  64     protected FlowStatsEntry updateSingleStat(final DataModificationTransaction trans, final FlowAndStatisticsMapList map) {
  65         short tableId = map.getTableId();
  66
  67         FlowStatisticsDataBuilder flowStatisticsData = new FlowStatisticsDataBuilder();
  68
  69         FlowBuilder flow = new FlowBuilder(map);
  70         if(map.getFlowId() != null) {
  71             flow.setId(new FlowId(map.getFlowId().getValue()));
  72         }
  73         if(map.getFlowId()!= null) {
  74             flow.setKey(new FlowKey(new FlowId(map.getKey().getFlowId().getValue())));
  75         }
  76
  77         Flow flowRule = flow.build();
  78
  79         FlowAndStatisticsMapListBuilder stats = new FlowAndStatisticsMapListBuilder();
  80         stats.setByteCount(map.getByteCount());
  81         stats.setPacketCount(map.getPacketCount());
  82         stats.setDuration(map.getDuration());
  83
  84         GenericStatistics flowStats = stats.build();
  85
  86         //Augment the data to the flow node
  87
  88         FlowStatisticsBuilder flowStatistics = new FlowStatisticsBuilder();
  89         flowStatistics.setByteCount(flowStats.getByteCount());
  90         flowStatistics.setPacketCount(flowStats.getPacketCount());
  91         flowStatistics.setDuration(flowStats.getDuration());
  92
  93         flowStatisticsData.setFlowStatistics(flowStatistics.build());
  94
  95         logger.debug("Flow : {}",flowRule.toString());
  96         logger.debug("Statistics to augment : {}",flowStatistics.build().toString());
  97
  98         InstanceIdentifier<Table> tableRef = getNodeIdentifierBuilder()
  99                 .augmentation(FlowCapableNode.class).child(Table.class, new TableKey(tableId)).toInstance();
 100
 101         //TODO: Not a good way to do it, need to figure out better way.
 102         //TODO: major issue in any alternate approach is that flow key is incrementally assigned
 103         //to the flows stored in data store.
 104         // Augment same statistics to all the matching masked flow
 105         Table table= (Table)trans.readConfigurationData(tableRef);
 106         if(table != null){
 107             for(Flow existingFlow : table.getFlow()){
 108                 logger.debug("Existing flow in data store : {}",existingFlow.toString());
 109                 if(FlowComparator.flowEquals(flowRule,existingFlow)){
 110                     InstanceIdentifier<Flow> flowRef = getNodeIdentifierBuilder()
 111                             .augmentation(FlowCapableNode.class)
 112                             .child(Table.class, new TableKey(tableId))
 113                             .child(Flow.class,existingFlow.getKey()).toInstance();
 114                     flow.setKey(existingFlow.getKey());
 115                     flow.addAugmentation(FlowStatisticsData.class, flowStatisticsData.build());
 116                     logger.debug("Found matching flow in the datastore, augmenting statistics");
 117                     // Update entry with timestamp of latest response
 118                     FlowStatsEntry flowStatsEntry = new FlowStatsEntry(tableId,flow.build());
 119                     trans.putOperationalData(flowRef, flow.build());
 120                     return flowStatsEntry;
 121                 }
 122             }
 123         }
 124
 125         table = (Table)trans.readOperationalData(tableRef);
 126         if(table != null){
 127             for(Flow existingFlow : table.getFlow()){
 128                 FlowStatisticsData augmentedflowStatisticsData = existingFlow.getAugmentation(FlowStatisticsData.class);
 129                 if(augmentedflowStatisticsData != null){
 130                     FlowBuilder existingOperationalFlow = new FlowBuilder();
 131                     existingOperationalFlow.fieldsFrom(augmentedflowStatisticsData.getFlowStatistics());
 132                     logger.debug("Existing unaccounted flow in operational data store : {}",existingFlow.toString());
 133                     if(FlowComparator.flowEquals(flowRule,existingOperationalFlow.build())){
 134                         InstanceIdentifier<Flow> flowRef = getNodeIdentifierBuilder()
 135                                 .augmentation(FlowCapableNode.class)
 136                                 .child(Table.class, new TableKey(tableId))
 137                                 .child(Flow.class,existingFlow.getKey()).toInstance();
 138                         flow.setKey(existingFlow.getKey());
 139                         flow.addAugmentation(FlowStatisticsData.class, flowStatisticsData.build());
 140                         logger.debug("Found matching unaccounted flow in the operational datastore, augmenting statistics");
 141                         // Update entry with timestamp of latest response
 142                         FlowStatsEntry flowStatsEntry = new FlowStatsEntry(tableId,flow.build());
 143                         trans.putOperationalData(flowRef, flow.build());
 144                         return flowStatsEntry;
 145                     }
 146                 }
 147             }
 148         }
 149
 150         String flowKey = "#UF$TABLE*"+Short.toString(tableId)+"*"+Integer.toString(this.unaccountedFlowsCounter);
 151         this.unaccountedFlowsCounter++;
 152         FlowKey newFlowKey = new FlowKey(new FlowId(flowKey));
 153         InstanceIdentifier<Flow> flowRef = getNodeIdentifierBuilder().augmentation(FlowCapableNode.class)
 154                     .child(Table.class, new TableKey(tableId))
 155                     .child(Flow.class,newFlowKey).toInstance();
 156         flow.setKey(newFlowKey);
 157         flow.addAugmentation(FlowStatisticsData.class, flowStatisticsData.build());
 158         logger.debug("Flow {} is not present in config data store, augmenting statistics as an unaccounted flow",
 159                     flow.build());
 160
 161         // Update entry with timestamp of latest response
 162         flow.setKey(newFlowKey);
 163         FlowStatsEntry flowStatsEntry = new FlowStatsEntry(tableId,flow.build());
 164         trans.putOperationalData(flowRef, flow.build());
 165         return flowStatsEntry;
 166     }
 167
 168     @Override
 169     protected InstanceIdentifier<?> listenPath() {
 170         return getNodeIdentifierBuilder().augmentation(FlowCapableNode.class).child(Table.class).child(Flow.class).build();
 171     }
 172
 173     @Override
 174     protected String statName() {
 175         return "Flow";
 176     }
 177
 178     @Override
 179     public void request() {
 180         // FIXME: it does not make sense to trigger this before sendAllFlowTablesStatisticsRequest()
 181         //        comes back -- we do not have any tables anyway.
 182         final Collection<TableKey> tables = flowTableStats.getTables();
 183         logger.debug("Node {} supports {} table(s)", this.getNodeRef(), tables.size());
 184         for (final TableKey key : tables) {
 185             logger.debug("Send aggregate stats request for flow table {} to node {}", key.getId(), this.getNodeRef());
 186             this.requestAggregateFlows(key);
 187         }
 188
 189         this.requestAllFlowsAllTables();
 190
 191     }
 192     public void requestAllFlowsAllTables() {
 193         if (flowStatsService != null) {
 194             final GetAllFlowsStatisticsFromAllFlowTablesInputBuilder input = new GetAllFlowsStatisticsFromAllFlowTablesInputBuilder();
 195             input.setNode(getNodeRef());
 196
 197             requestHelper(flowStatsService.getAllFlowsStatisticsFromAllFlowTables(input.build()));
 198         }
 199     }
 200
 201     public void requestAggregateFlows(final TableKey key) {
 202         if (flowStatsService != null) {
 203             GetAggregateFlowStatisticsFromFlowTableForAllFlowsInputBuilder input =
 204                     new GetAggregateFlowStatisticsFromFlowTableForAllFlowsInputBuilder();
 205
 206             input.setNode(getNodeRef());
 207             input.setTableId(new org.opendaylight.yang.gen.v1.urn.opendaylight.table.types.rev131026.TableId(key.getId()));
 208             requestHelper(flowStatsService.getAggregateFlowStatisticsFromFlowTableForAllFlows(input.build()));
 209         }
 210     }
 211
 212     public void requestFlow(final Flow flow) {
 213         if (flowStatsService != null) {
 214             final GetFlowStatisticsFromFlowTableInputBuilder input =
 215                     new GetFlowStatisticsFromFlowTableInputBuilder(flow);
 216             input.setNode(getNodeRef());
 217
 218             requestHelper(flowStatsService.getFlowStatisticsFromFlowTable(input.build()));
 219         }
 220     }
 221
 222     @Override
 223     public void onDataChanged(final DataChangeEvent<InstanceIdentifier<?>, DataObject> change) {
 224         for (Entry<InstanceIdentifier<?>, DataObject> e : change.getCreatedConfigurationData().entrySet()) {
 225             if (Flow.class.equals(e.getKey().getTargetType())) {
 226                 final Flow flow = (Flow) e.getValue();
 227                 logger.debug("Key {} triggered request for flow {}", e.getKey(), flow);
 228                 requestFlow(flow);
 229             } else {
 230                 logger.debug("Ignoring key {}", e.getKey());
 231             }
 232         }
 233
 234         final DataModificationTransaction trans = startTransaction();
 235         for (InstanceIdentifier<?> key : change.getRemovedConfigurationData()) {
 236             if (Flow.class.equals(key.getTargetType())) {
 237                 @SuppressWarnings("unchecked")
 238                 final InstanceIdentifier<Flow> flow = (InstanceIdentifier<Flow>)key;
 239                 logger.debug("Key {} triggered remove of Flow from operational space.", key);
 240                 trans.removeOperationalData(flow);
 241             }
 242         }
 243         trans.commit();
 244     }
 245
 246     @Override
 247     public void start(final DataBrokerService dbs) {
 248         if (flowStatsService == null) {
 249             logger.debug("No Flow Statistics service, not subscribing to flows on node {}", getNodeIdentifier());
 250             return;
 251         }
 252
 253         super.start(dbs);
 254     }
 255 }