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

   1 /**
   2  * Copyright (c) 2014 Cisco Systems, Inc. and others.  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
   9 package org.opendaylight.controller.md.statistics.manager.impl;
  10
  11 import java.util.ArrayList;
  12 import java.util.Collections;
  13 import java.util.HashMap;
  14 import java.util.Iterator;
  15 import java.util.LinkedList;
  16 import java.util.List;
  17 import java.util.Map;
  18 import java.util.Map.Entry;
  19 import java.util.concurrent.atomic.AtomicInteger;
  20
  21 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  22 import org.opendaylight.controller.md.sal.binding.api.ReadWriteTransaction;
  23 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  24 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
  25 import org.opendaylight.controller.md.statistics.manager.StatRpcMsgManager.TransactionCacheContainer;
  26 import org.opendaylight.controller.md.statistics.manager.StatisticsManager;
  27 import org.opendaylight.controller.md.statistics.manager.StatisticsManager.StatDataStoreOperation;
  28 import org.opendaylight.controller.md.statistics.manager.StatisticsManager.StatDataStoreOperation.StatsManagerOperationType;
  29 import org.opendaylight.controller.md.statistics.manager.impl.helper.FlowComparator;
  30 import org.opendaylight.controller.sal.binding.api.NotificationProviderService;
  31 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
  32 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowHashIdMapping;
  33 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowHashIdMappingBuilder;
  34 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowId;
  35 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.nodes.node.table.FlowHashIdMap;
  36 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.nodes.node.table.FlowHashIdMapBuilder;
  37 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.nodes.node.table.FlowHashIdMapKey;
  38 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
  39 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableBuilder;
  40 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableKey;
  41 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow;
  42 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowBuilder;
  43 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowKey;
  44 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.AggregateFlowStatisticsData;
  45 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.AggregateFlowStatisticsDataBuilder;
  46 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.AggregateFlowStatisticsUpdate;
  47 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.FlowStatisticsData;
  48 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.FlowStatisticsDataBuilder;
  49 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.FlowsStatisticsUpdate;
  50 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.OpendaylightFlowStatisticsListener;
  51 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.aggregate.flow.statistics.AggregateFlowStatisticsBuilder;
  52 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.flow.and.statistics.map.list.FlowAndStatisticsMapList;
  53 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.flow.and.statistics.map.list.FlowAndStatisticsMapListBuilder;
  54 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.flow.statistics.FlowStatisticsBuilder;
  55 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.transaction.rev131103.TransactionAware;
  56 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.transaction.rev131103.TransactionId;
  57 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
  58 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes;
  59 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
  60 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
  61 import org.opendaylight.yangtools.yang.binding.DataObject;
  62 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  63 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
  64 import org.slf4j.Logger;
  65 import org.slf4j.LoggerFactory;
  66
  67 import com.google.common.base.Optional;
  68 import com.google.common.collect.BiMap;
  69 import com.google.common.collect.HashBiMap;
  70
  71 /**
  72  * statistics-manager
  73  * org.opendaylight.controller.md.statistics.manager.impl
  74  *
  75  * StatListenCommitFlow
  76  * Class is a NotifyListener for FlowStatistics and DataChangeListener for Config/DataStore for Flow node.
  77  * All expected (registered) FlowStatistics will be builded and commit to Operational/DataStore.
  78  * DataChangeEven should call create/delete Flow in Operational/DS create process needs to pair
  79  * Device Flow HashCode and FlowId from Config/DS
  80  *
  81  * @author <a href="mailto:vdemcak@cisco.com">Vaclav Demcak</a>
  82  *
  83  */
  84 public class StatListenCommitFlow extends StatAbstractListenCommit<Flow, OpendaylightFlowStatisticsListener>
  85                                             implements OpendaylightFlowStatisticsListener {
  86
  87     protected static final Logger LOG = LoggerFactory.getLogger(StatListenCommitFlow.class);
  88
  89     private static final String ALIEN_SYSTEM_FLOW_ID = "#UF$TABLE*";
  90
  91     private static final Integer REMOVE_AFTER_MISSING_COLLECTION = 1;
  92
  93     private final AtomicInteger unaccountedFlowsCounter = new AtomicInteger(0);
  94
  95     public StatListenCommitFlow (final StatisticsManager manager, final DataBroker db,
  96             final NotificationProviderService nps){
  97         super(manager, db, nps, Flow.class);
  98     }
  99
 100     @Override
 101     protected OpendaylightFlowStatisticsListener getStatNotificationListener() {
 102         return this;
 103     }
 104
 105     @Override
 106     protected InstanceIdentifier<Flow> getWildCardedRegistrationPath() {
 107         return InstanceIdentifier.create(Nodes.class).child(Node.class)
 108                 .augmentation(FlowCapableNode.class).child(Table.class).child(Flow.class);
 109     }
 110
 111     @Override
 112     public void onAggregateFlowStatisticsUpdate(final AggregateFlowStatisticsUpdate notification) {
 113         final TransactionId transId = notification.getTransactionId();
 114         final NodeId nodeId = notification.getId();
 115         if ( ! isExpectedStatistics(transId, nodeId)) {
 116             LOG.debug("STAT-MANAGER - AggregateFlowStatisticsUpdate: unregistred notification detect TransactionId {}", transId);
 117             return;
 118         }
 119         manager.getRpcMsgManager().addNotification(notification, nodeId);
 120         if (notification.isMoreReplies()) {
 121             return;
 122         }
 123         /* check flow Capable Node and write statistics */
 124         manager.enqueue(new StatDataStoreOperation(StatsManagerOperationType.DATA_COMMIT_OPER_DS,nodeId) {
 125             @Override
 126             public void applyOperation(final ReadWriteTransaction tx) {
 127
 128                 final Optional<TransactionCacheContainer<?>> txContainer = getTransactionCacheContainer(transId, nodeId);
 129                 if (( ! txContainer.isPresent()) || txContainer.get().getNotifications() == null) {
 130                     return;
 131                 }
 132                 final Optional<? extends DataObject> inputObj = txContainer.get().getConfInput();
 133                 if (( ! inputObj.isPresent()) || ( ! (inputObj.get() instanceof Table))) {
 134                     return;
 135                 }
 136                 final Table table = (Table) inputObj.get();
 137                 final List<? extends TransactionAware> cacheNotifs = txContainer.get().getNotifications();
 138                 for (final TransactionAware notif : cacheNotifs) {
 139                     if (notif instanceof AggregateFlowStatisticsUpdate) {
 140                         final AggregateFlowStatisticsData stats = new AggregateFlowStatisticsDataBuilder()
 141                             .setAggregateFlowStatistics(new AggregateFlowStatisticsBuilder(notification).build()).build();
 142                         final InstanceIdentifier<FlowCapableNode> fNodeIdent = InstanceIdentifier.create(Nodes.class)
 143                                 .child(Node.class, new NodeKey(nodeId)).augmentation(FlowCapableNode.class);
 144                         final InstanceIdentifier<Table> tableRef = fNodeIdent.child(Table.class, table.getKey());
 145                         final InstanceIdentifier<AggregateFlowStatisticsData> tableStatRef = tableRef
 146                                 .augmentation(AggregateFlowStatisticsData.class);
 147                         Optional<FlowCapableNode> fNode = Optional.absent();
 148                         try {
 149                             fNode = tx.read(LogicalDatastoreType.OPERATIONAL, fNodeIdent).checkedGet();
 150                         } catch (final ReadFailedException e) {
 151                             LOG.debug("Read Operational/DS for FlowCapableNode fail! {}", fNodeIdent, e);
 152                             return;
 153                         }
 154                         if (fNode.isPresent()) {
 155                             ensureTable(tx, table.getId(), tableRef);
 156                             tx.put(LogicalDatastoreType.OPERATIONAL, tableStatRef, stats);
 157                         }
 158                     }
 159                 }
 160             }
 161         });
 162     }
 163
 164     public void ensureTable(final ReadWriteTransaction tx, final Short tableId, final InstanceIdentifier<Table> tableRef) {
 165         final Table tableNew = new TableBuilder().setId(tableId).build();
 166         tx.merge(LogicalDatastoreType.OPERATIONAL, tableRef, tableNew);
 167     }
 168
 169     @Override
 170     public void onFlowsStatisticsUpdate(final FlowsStatisticsUpdate notification) {
 171         final TransactionId transId = notification.getTransactionId();
 172         final NodeId nodeId = notification.getId();
 173         if ( ! isExpectedStatistics(transId, nodeId)) {
 174             LOG.debug("STAT-MANAGER - FlowsStatisticsUpdate: unregistred notification detect TransactionId {}", transId);
 175             return;
 176         }
 177         manager.getRpcMsgManager().addNotification(notification, nodeId);
 178         if (notification.isMoreReplies()) {
 179             LOG.trace("Next notification for join txId {}", transId);
 180             return;
 181         }
 182         /* add flow's statistics */
 183         manager.enqueue(new StatDataStoreOperation(StatsManagerOperationType.DATA_COMMIT_OPER_DS,nodeId) {
 184             @Override
 185             public void applyOperation(final ReadWriteTransaction tx) {
 186                 final Optional<TransactionCacheContainer<?>> txContainer = getTransactionCacheContainer(transId, nodeId);
 187                 if (( ! txContainer.isPresent()) || txContainer.get().getNotifications() == null) {
 188                     return;
 189                 }
 190                 final List<FlowAndStatisticsMapList> flowStats = new ArrayList<FlowAndStatisticsMapList>(10);
 191                 final InstanceIdentifier<Node> nodeIdent = InstanceIdentifier.create(Nodes.class)
 192                         .child(Node.class, new NodeKey(nodeId));
 193                 final List<? extends TransactionAware> cacheNotifs = txContainer.get().getNotifications();
 194                 for (final TransactionAware notif : cacheNotifs) {
 195                     if (notif instanceof FlowsStatisticsUpdate) {
 196                         final List<FlowAndStatisticsMapList> notifList =
 197                                 ((FlowsStatisticsUpdate) notif).getFlowAndStatisticsMapList();
 198                         if (notifList != null) {
 199                             flowStats.addAll(notifList);
 200                         }
 201                     }
 202                 }
 203
 204                 statsFlowCommitAll(flowStats, nodeIdent, tx);
 205                 /* cleaning all not cached hash collisions */
 206                 final Map<InstanceIdentifier<Flow>, Integer> listAliens = mapNodesForDelete.get(nodeIdent);
 207                 if (listAliens != null) {
 208                     for (final Entry<InstanceIdentifier<Flow>, Integer> nodeForDelete : listAliens.entrySet()) {
 209                         final Integer lifeIndex = nodeForDelete.getValue();
 210                         if (nodeForDelete.getValue() > 0) {
 211                             nodeForDelete.setValue(Integer.valueOf(lifeIndex.intValue() - 1));
 212                         } else {
 213                             final InstanceIdentifier<Flow> flowNodeIdent = nodeForDelete.getKey();
 214                             mapNodesForDelete.get(nodeIdent).remove(flowNodeIdent);
 215                             tx.delete(LogicalDatastoreType.OPERATIONAL, flowNodeIdent);
 216                         }
 217                     }
 218                 }
 219                 /* Notification for continue collecting statistics */
 220                 notifyToCollectNextStatistics(nodeIdent, transId);
 221             }
 222
 223         });
 224     }
 225
 226     private void statsFlowCommitAll(final List<FlowAndStatisticsMapList> list,
 227             final InstanceIdentifier<Node> nodeIdent, final ReadWriteTransaction tx) {
 228
 229         final InstanceIdentifier<FlowCapableNode> fNodeIdent = nodeIdent.augmentation(FlowCapableNode.class);
 230
 231         final Optional<FlowCapableNode> fNode;
 232         try {
 233             fNode = tx.read(LogicalDatastoreType.OPERATIONAL, fNodeIdent).checkedGet();
 234         }
 235         catch (final ReadFailedException e) {
 236             LOG.debug("Read FlowCapableNode {} in Operational/DS fail! Statistic scan not be updated.", nodeIdent, e);
 237             return;
 238         }
 239         if ( ! fNode.isPresent()) {
 240             LOG.trace("FlowCapableNode {} is not presented in Operational/DS. Statisticscan not be updated.", nodeIdent);
 241             return;
 242         }
 243
 244         final NodeUpdateState nodeState = new NodeUpdateState(fNodeIdent,fNode.get());
 245
 246         for (final FlowAndStatisticsMapList flowStat : list) {
 247             final TableKey tableKey = new TableKey(flowStat.getTableId());
 248             final TableFlowUpdateState tableState = nodeState.getTable(tableKey, tx);
 249             tableState.reportFlow(flowStat,tx);
 250         }
 251
 252         for (final TableFlowUpdateState table : nodeState.getTables()) {
 253             table.removeUnreportedFlows(tx);
 254         }
 255     }
 256
 257     /**
 258      * Method adds statistics to Flow
 259      *
 260      * @param flowBuilder
 261      * @param deviceFlow
 262      */
 263     private void addStatistics(final FlowBuilder flowBuilder, final FlowAndStatisticsMapList deviceFlow) {
 264         final FlowAndStatisticsMapListBuilder stats = new FlowAndStatisticsMapListBuilder(deviceFlow);
 265         final FlowStatisticsBuilder flowStatisticsBuilder = new FlowStatisticsBuilder(stats.build());
 266         final FlowStatisticsDataBuilder flowStatisticsData =new FlowStatisticsDataBuilder();
 267         flowStatisticsData.setFlowStatistics(flowStatisticsBuilder.build());
 268         flowBuilder.addAugmentation(FlowStatisticsData.class, flowStatisticsData.build());
 269     }
 270
 271     /**
 272      * build pseudoUnique hashCode for flow in table
 273      * for future easy identification
 274      *
 275      * FIXME: we expect same version for YANG models for all clusters and that has to be fix
 276      * FIXME: CREATE BETTER KEY - for flow (MATCH is the problem)
 277      */
 278     static String buildFlowIdOperKey(final FlowAndStatisticsMapList deviceFlow) {
 279         return new StringBuffer().append(deviceFlow.getMatch())
 280                 .append(deviceFlow.getPriority()).append(deviceFlow.getCookie().getValue()).toString();
 281     }
 282
 283     private class NodeUpdateState {
 284         private final InstanceIdentifier<FlowCapableNode> nodeIdentifier;
 285         private final Map<TableKey,TableFlowUpdateState> tables = new HashMap<>();
 286
 287         public NodeUpdateState(final InstanceIdentifier<FlowCapableNode> fNodeIdent, final FlowCapableNode flowCapableNode) {
 288             nodeIdentifier = fNodeIdent;
 289             final List<Table> tableList = flowCapableNode.getTable();
 290             if(tableList != null) {
 291             for (final Table table : tableList) {
 292                 final TableKey tableKey = table.getKey();
 293                     tables.put(tableKey, new TableFlowUpdateState(nodeIdentifier.child(Table.class,tableKey),table));
 294                 }
 295             }
 296         }
 297
 298         public Iterable<TableFlowUpdateState> getTables() {
 299             return tables.values();
 300         }
 301
 302         TableFlowUpdateState getTable(final TableKey key,final ReadWriteTransaction tx) {
 303             TableFlowUpdateState table = tables.get(key);
 304             if(table == null) {
 305                 table = new TableFlowUpdateState(nodeIdentifier.child(Table.class, key), null);
 306                 tables.put(key, table);
 307             }
 308             return table;
 309         }
 310     }
 311
 312     private class TableFlowUpdateState {
 313
 314         private boolean tableEnsured = false;
 315         final KeyedInstanceIdentifier<Table, TableKey> tableRef;
 316         final TableKey tableKey;
 317         final BiMap<FlowHashIdMapKey, FlowId> flowIdByHash;
 318         List<Flow> configFlows;
 319
 320         public TableFlowUpdateState(final KeyedInstanceIdentifier<Table, TableKey> tablePath, final Table table) {
 321             tableRef = tablePath;
 322             tableKey = tablePath.getKey();
 323             flowIdByHash = HashBiMap.create();
 324             if(table != null) {
 325                 final FlowHashIdMapping flowHashMapping = table.getAugmentation(FlowHashIdMapping.class);
 326                 if (flowHashMapping != null) {
 327                     final List<FlowHashIdMap>  flowHashMap = flowHashMapping.getFlowHashIdMap() != null
 328                             ? flowHashMapping.getFlowHashIdMap() : Collections.<FlowHashIdMap> emptyList();
 329                     for (final FlowHashIdMap flowHashId : flowHashMap) {
 330                         try {
 331                             flowIdByHash.put(flowHashId.getKey(), flowHashId.getFlowId());
 332                         } catch (final Exception e) {
 333                             LOG.warn("flow hashing hit a duplicate for {} -> {}", flowHashId.getKey(), flowHashId.getFlowId());
 334                         }
 335                     }
 336                 }
 337             }
 338         }
 339
 340         private void ensureTableFowHashIdMapping(final ReadWriteTransaction tx) {
 341             if( ! tableEnsured) {
 342                 ensureTable(tx, tableKey.getId(), tableRef);
 343                 final FlowHashIdMapping emptyMapping = new FlowHashIdMappingBuilder()
 344                     .setFlowHashIdMap(Collections.<FlowHashIdMap> emptyList()).build();
 345                 tx.merge(LogicalDatastoreType.OPERATIONAL, tableRef.augmentation(FlowHashIdMapping.class), emptyMapping);
 346                 tableEnsured = true;
 347             }
 348         }
 349
 350         private FlowKey searchInConfiguration(final FlowAndStatisticsMapList flowStat, final ReadWriteTransaction trans) {
 351             initConfigFlows(trans);
 352             final Iterator<Flow> it = configFlows.iterator();
 353             while(it.hasNext()) {
 354                 final Flow cfgFlow = it.next();
 355                 final FlowKey cfgKey = cfgFlow.getKey();
 356                 if(flowIdByHash.inverse().containsKey(cfgKey)) {
 357                     it.remove();
 358                 } else if(FlowComparator.flowEquals(flowStat, cfgFlow)) {
 359                     it.remove();
 360                     return cfgKey;
 361                 }
 362             }
 363             return null;
 364         }
 365
 366         private void initConfigFlows(final ReadWriteTransaction trans) {
 367             final Optional<Table> table = readLatestConfiguration(tableRef);
 368             List<Flow> localList = null;
 369             if(table.isPresent()) {
 370                 localList = table.get().getFlow();
 371             }
 372             if(localList == null) {
 373                 configFlows = Collections.emptyList();
 374             } else {
 375                 configFlows = new LinkedList<>(localList);
 376             }
 377         }
 378
 379         private FlowKey getFlowKeyAndRemoveHash(final FlowHashIdMapKey key) {
 380             final FlowId ret = flowIdByHash.get(key);
 381             if(ret != null) {
 382                 flowIdByHash.remove(key);
 383                 return new FlowKey(ret);
 384             }
 385             return null;
 386         }
 387
 388         /* Returns FlowKey which doesn't exist in any DataStore for now */
 389         private FlowKey makeAlienFlowKey() {
 390             final StringBuilder sBuilder = new StringBuilder(ALIEN_SYSTEM_FLOW_ID)
 391                 .append(tableKey.getId()).append("-").append(unaccountedFlowsCounter.incrementAndGet());
 392             final FlowId flowId = new FlowId(sBuilder.toString());
 393             return new FlowKey(flowId);
 394         }
 395
 396         private Map<FlowHashIdMapKey, FlowId> getRemovalList() {
 397             return flowIdByHash;
 398         }
 399
 400         void reportFlow(final FlowAndStatisticsMapList flowStat, final ReadWriteTransaction trans) {
 401             ensureTableFowHashIdMapping(trans);
 402             final FlowHashIdMapKey hashingKey = new FlowHashIdMapKey(buildFlowIdOperKey(flowStat));
 403             FlowKey flowKey = getFlowKeyAndRemoveHash(hashingKey);
 404             if (flowKey == null) {
 405                 flowKey = searchInConfiguration(flowStat, trans);
 406                 if ( flowKey == null) {
 407                     flowKey = makeAlienFlowKey();
 408                 }
 409                 updateHashCache(trans,flowKey,hashingKey);
 410             }
 411             final FlowBuilder flowBuilder = new FlowBuilder(flowStat);
 412             flowBuilder.setKey(flowKey);
 413             addStatistics(flowBuilder, flowStat);
 414             final InstanceIdentifier<Flow> flowIdent = tableRef.child(Flow.class, flowKey);
 415             trans.put(LogicalDatastoreType.OPERATIONAL, flowIdent, flowBuilder.build());
 416             /* check life for Alien flows */
 417             if (flowKey.getId().getValue().startsWith(ALIEN_SYSTEM_FLOW_ID)) {
 418                 removeData(flowIdent, REMOVE_AFTER_MISSING_COLLECTION);
 419             }
 420         }
 421
 422         /* Build and deploy new FlowHashId map */
 423         private void updateHashCache(final ReadWriteTransaction trans, final FlowKey flowKey, final FlowHashIdMapKey hashingKey) {
 424             final FlowHashIdMapBuilder flHashIdMap = new FlowHashIdMapBuilder();
 425             flHashIdMap.setFlowId(flowKey.getId());
 426             flHashIdMap.setKey(hashingKey);
 427             final KeyedInstanceIdentifier<FlowHashIdMap, FlowHashIdMapKey> flHashIdent = tableRef
 428                     .augmentation(FlowHashIdMapping.class).child(FlowHashIdMap.class, hashingKey);
 429             /* Add new FlowHashIdMap */
 430             trans.put(LogicalDatastoreType.OPERATIONAL, flHashIdent, flHashIdMap.build());
 431         }
 432
 433         void removeUnreportedFlows(final ReadWriteTransaction tx) {
 434             final InstanceIdentifier<Node> nodeIdent = tableRef.firstIdentifierOf(Node.class);
 435             final List<InstanceIdentifier<Flow>> listMissingConfigFlows = notStatReportedConfigFlows();
 436             final Map<InstanceIdentifier<Flow>, Integer> nodeDeleteMap = mapNodesForDelete.get(nodeIdent);
 437             final Map<FlowHashIdMapKey, FlowId> listForRemove = getRemovalList();
 438             for (final Entry<FlowHashIdMapKey, FlowId> entryForRemove : listForRemove.entrySet()) {
 439                 final FlowKey flowKey = new FlowKey(entryForRemove.getValue());
 440                 final InstanceIdentifier<Flow> flowRef = tableRef.child(Flow.class, flowKey);
 441                 if (nodeDeleteMap != null && flowKey.getId().getValue().startsWith(ALIEN_SYSTEM_FLOW_ID)) {
 442                     final Integer lifeIndex = nodeDeleteMap.get(flowRef);
 443                     if (lifeIndex > 0) {
 444                         break;
 445                     } else {
 446                         nodeDeleteMap.remove(flowRef);
 447                     }
 448                 } else {
 449                     if (listMissingConfigFlows.remove(flowRef)) {
 450                         break; // we probably lost some multipart msg
 451                     }
 452                 }
 453                 final InstanceIdentifier<FlowHashIdMap> flHashIdent =
 454                         tableRef.augmentation(FlowHashIdMapping.class).child(FlowHashIdMap.class, entryForRemove.getKey());
 455                 tx.delete(LogicalDatastoreType.OPERATIONAL, flowRef);
 456                 tx.delete(LogicalDatastoreType.OPERATIONAL, flHashIdent);
 457             }
 458         }
 459
 460         List<InstanceIdentifier<Flow>> notStatReportedConfigFlows() {
 461             if (configFlows != null) {
 462                 final List<InstanceIdentifier<Flow>> returnList = new ArrayList<>(configFlows.size());
 463                 for (final Flow confFlow : configFlows) {
 464                     final InstanceIdentifier<Flow> confFlowIdent = tableRef.child(Flow.class, confFlow.getKey());
 465                     returnList.add(confFlowIdent);
 466                 }
 467                 return returnList;
 468             }
 469             return Collections.emptyList();
 470         }
 471     }
 472 }
 473