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