Merge "fix parital multipart processing"
[openflowplugin.git] / openflowplugin-impl / src / main / java / org / opendaylight / openflowplugin / impl / statistics / StatisticsGatheringUtils.java
index fc530e341678989d727f5e5f590320bc1239c677..05ff9d495afab59dd6eebaf85a3e71ff1b384076 100644 (file)
@@ -8,23 +8,30 @@
 
 package org.opendaylight.openflowplugin.impl.statistics;
 
+import com.google.common.collect.Iterables;
+
+import java.util.Collections;
 import com.google.common.base.Function;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.JdkFutureAdapters;
 import com.google.common.util.concurrent.ListenableFuture;
 import java.util.List;
+import java.util.Map;
 import javax.annotation.Nullable;
 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
-import org.opendaylight.openflowplugin.api.openflow.flow.registry.FlowHash;
-import org.opendaylight.openflowplugin.api.openflow.flow.registry.FlowRegistryException;
-import org.opendaylight.openflowplugin.impl.flow.registry.FlowHashFactory;
+import org.opendaylight.openflowplugin.api.openflow.registry.flow.DeviceFlowRegistry;
+import org.opendaylight.openflowplugin.api.openflow.registry.flow.FlowDescriptor;
+import org.opendaylight.openflowplugin.api.openflow.registry.flow.FlowHash;
+import org.opendaylight.openflowplugin.impl.registry.flow.FlowDescriptorFactory;
+import org.opendaylight.openflowplugin.impl.registry.flow.FlowHashFactory;
 import org.opendaylight.openflowplugin.impl.statistics.services.dedicated.StatisticsGatheringService;
 import org.opendaylight.openflowplugin.impl.util.FlowUtil;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNodeConnector;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowId;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.meters.Meter;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.meters.MeterBuilder;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.meters.MeterKey;
 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;
@@ -41,22 +48,31 @@ import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev13
 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.port.rev130925.queues.Queue;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.port.rev130925.queues.QueueKey;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.GroupDescStatsUpdated;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.NodeGroupDescStats;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.NodeGroupDescStatsBuilder;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.group.desc.GroupDescBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.GroupStatisticsUpdated;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.NodeGroupStatistics;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.NodeGroupStatisticsBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.group.statistics.GroupStatistics;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.group.statistics.GroupStatisticsBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.GroupId;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.group.desc.stats.reply.GroupDescStats;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.group.statistics.reply.GroupStats;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.Group;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.GroupBuilder;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.GroupKey;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnector;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnectorKey;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.MeterConfigStatsUpdated;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.MeterStatisticsUpdated;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.NodeMeterStatistics;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.NodeMeterStatisticsBuilder;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.nodes.node.meter.MeterStatistics;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.nodes.node.meter.MeterStatisticsBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.types.rev130918.MeterId;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.types.rev130918.meter.config.stats.reply.MeterConfigStats;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.types.rev130918.meter.statistics.reply.MeterStats;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.common.types.rev130731.MultipartType;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.MultipartReply;
@@ -103,153 +119,264 @@ public final class StatisticsGatheringUtils {
         return transformAndStoreStatisticsData(statisticsDataInFuture, deviceContext);
     }
 
-    private static ListenableFuture<Boolean> transformAndStoreStatisticsData(final ListenableFuture<RpcResult<List<MultipartReply>>> statisticsDataInFuture, final DeviceContext deviceContext) {
+    private static ListenableFuture<Boolean> transformAndStoreStatisticsData(final ListenableFuture<RpcResult<List<MultipartReply>>> statisticsDataInFuture,
+                                                                             final DeviceContext deviceContext) {
         return Futures.transform(statisticsDataInFuture, new Function<RpcResult<List<MultipartReply>>, Boolean>() {
             @Nullable
             @Override
             public Boolean apply(final RpcResult<List<MultipartReply>> rpcResult) {
                 if (rpcResult.isSuccessful()) {
-                    final InstanceIdentifier basicIId = getInstanceIdentifier(deviceContext);
+                    boolean isMultipartProcessed = Boolean.TRUE;
+                    Iterable<? extends DataObject> allMultipartData = Collections.emptyList();
+                    DataObject multipartData = null;
                     for (final MultipartReply singleReply : rpcResult.getResult()) {
                         final List<? extends DataObject> multipartDataList = MULTIPART_REPLY_TRANSLATOR.translate(deviceContext, singleReply);
-                        for (final DataObject singleMultipartData : multipartDataList) {
-                            boolean logFirstTime = true;
-                            if (singleMultipartData instanceof GroupDescStatsUpdated) {
-                                final GroupDescStatsUpdated groupDescStatsUpdated = (GroupDescStatsUpdated) singleMultipartData;
-                                for (final GroupDescStats groupDescStats : groupDescStatsUpdated.getGroupDescStats()) {
-                                    final InstanceIdentifier<Node> nodeIdent = InstanceIdentifier
-                                            .create(Nodes.class).child(Node.class, new NodeKey(groupDescStatsUpdated.getId()));
-                                    final InstanceIdentifier<FlowCapableNode> fNodeIdent = nodeIdent.augmentation(FlowCapableNode.class);
-                                    final GroupKey groupKey = new GroupKey(groupDescStats.getGroupId());
-                                    final InstanceIdentifier<Group> groupRef = fNodeIdent.child(Group.class, groupKey);
-                                    final GroupBuilder groupBuilder = new GroupBuilder(groupDescStats);
-                                    final NodeGroupDescStatsBuilder groupDesc = new NodeGroupDescStatsBuilder();
-                                    groupDesc.setGroupDesc(new GroupDescBuilder(groupDescStats).build());
-                                    groupBuilder.addAugmentation(NodeGroupDescStats.class, groupDesc.build());
-                                    if (logFirstTime) {
-                                        logFirstTime = false;
-                                        LOG.info("Writing group statistics to operational DS.");
-                                    }
-                                    deviceContext.writeToTransaction(LogicalDatastoreType.OPERATIONAL, groupRef, groupBuilder.build());
-                                }
-                            }
-
-                            if (singleMultipartData instanceof MeterStatisticsUpdated) {
-                                final MeterStatisticsUpdated meterStatisticsUpdated = (MeterStatisticsUpdated) singleMultipartData;
-                                final InstanceIdentifier<Node> nodeIdent = InstanceIdentifier
-                                        .create(Nodes.class).child(Node.class, new NodeKey(meterStatisticsUpdated.getId()));
-                                final InstanceIdentifier<FlowCapableNode> fNodeIdent = nodeIdent.augmentation(FlowCapableNode.class);
-
-                                for (final MeterStats mStat : meterStatisticsUpdated.getMeterStats()) {
-                                    final MeterStatistics stats = new MeterStatisticsBuilder(mStat).build();
-
-                                    final InstanceIdentifier<Meter> meterIdent = fNodeIdent.child(Meter.class, new MeterKey(mStat.getMeterId()));
-                                    final InstanceIdentifier<NodeMeterStatistics> nodeMeterStatIdent = meterIdent
-                                            .augmentation(NodeMeterStatistics.class);
-                                    final InstanceIdentifier<MeterStatistics> msIdent = nodeMeterStatIdent.child(MeterStatistics.class);
-                                    if (logFirstTime) {
-                                        logFirstTime = false;
-                                        LOG.info("Writing meter statistics to operational DS.");
-                                    }
-                                    deviceContext.writeToTransaction(LogicalDatastoreType.OPERATIONAL, msIdent, stats);
-                                }
-                            }
-                            if (singleMultipartData instanceof NodeConnectorStatisticsUpdate) {
-                                final NodeConnectorStatisticsUpdate nodeConnectorStatisticsUpdate = (NodeConnectorStatisticsUpdate) singleMultipartData;
-                                final InstanceIdentifier<Node> nodeIdent = InstanceIdentifier.create(Nodes.class)
-                                        .child(Node.class, new NodeKey(nodeConnectorStatisticsUpdate.getId()));
-                                for (final NodeConnectorStatisticsAndPortNumberMap nConnectPort : nodeConnectorStatisticsUpdate.getNodeConnectorStatisticsAndPortNumberMap()) {
-                                    final FlowCapableNodeConnectorStatistics stats = new FlowCapableNodeConnectorStatisticsBuilder(nConnectPort).build();
-                                    final NodeConnectorKey key = new NodeConnectorKey(nConnectPort.getNodeConnectorId());
-                                    final InstanceIdentifier<NodeConnector> nodeConnectorIdent = nodeIdent.child(NodeConnector.class, key);
-                                    final InstanceIdentifier<FlowCapableNodeConnectorStatisticsData> nodeConnStatIdent = nodeConnectorIdent
-                                            .augmentation(FlowCapableNodeConnectorStatisticsData.class);
-                                    final InstanceIdentifier<FlowCapableNodeConnectorStatistics> flowCapNodeConnStatIdent =
-                                            nodeConnStatIdent.child(FlowCapableNodeConnectorStatistics.class);
-                                    if (logFirstTime) {
-                                        logFirstTime = false;
-                                        LOG.info("Writing port statistics to operational DS.");
-                                    }
-                                    deviceContext.writeToTransaction(LogicalDatastoreType.OPERATIONAL, flowCapNodeConnStatIdent, stats);
-                                }
-                            }
-                            if (singleMultipartData instanceof FlowTableStatisticsUpdate) {
-
-                                final FlowTableStatisticsUpdate flowTableStatisticsUpdate = (FlowTableStatisticsUpdate) singleMultipartData;
-                                final InstanceIdentifier<FlowCapableNode> fNodeIdent = InstanceIdentifier.create(Nodes.class)
-                                        .child(Node.class, new NodeKey(flowTableStatisticsUpdate.getId())).augmentation(FlowCapableNode.class);
-
-                                for (final FlowTableAndStatisticsMap tableStat : flowTableStatisticsUpdate.getFlowTableAndStatisticsMap()) {
-                                    final InstanceIdentifier<FlowTableStatistics> tStatIdent = fNodeIdent.child(Table.class, new TableKey(tableStat.getTableId().getValue()))
-                                            .augmentation(FlowTableStatisticsData.class).child(FlowTableStatistics.class);
-                                    final FlowTableStatistics stats = new FlowTableStatisticsBuilder(tableStat).build();
-                                    if (logFirstTime) {
-                                        logFirstTime = false;
-                                        LOG.info("Writing flowTable statistics to operational DS.");
-                                    }
-                                    deviceContext.writeToTransaction(LogicalDatastoreType.OPERATIONAL, tStatIdent, stats);
-                                }
-                            }
-                            if (singleMultipartData instanceof QueueStatisticsUpdate) {
-                                final QueueStatisticsUpdate queueStatisticsUpdate = (QueueStatisticsUpdate) singleMultipartData;
-                                final InstanceIdentifier<Node> nodeIdent = InstanceIdentifier.create(Nodes.class)
-                                        .child(Node.class, new NodeKey(queueStatisticsUpdate.getId()));
-                                for (final QueueIdAndStatisticsMap queueStat : queueStatisticsUpdate.getQueueIdAndStatisticsMap()) {
-                                    if (queueStat.getQueueId() != null) {
-                                        final FlowCapableNodeConnectorQueueStatistics statChild =
-                                                new FlowCapableNodeConnectorQueueStatisticsBuilder(queueStat).build();
-                                        final FlowCapableNodeConnectorQueueStatisticsDataBuilder statBuild =
-                                                new FlowCapableNodeConnectorQueueStatisticsDataBuilder();
-                                        statBuild.setFlowCapableNodeConnectorQueueStatistics(statChild);
-                                        final QueueKey qKey = new QueueKey(queueStat.getQueueId());
-                                        final InstanceIdentifier<Queue> queueIdent = nodeIdent
-                                                .child(NodeConnector.class, new NodeConnectorKey(queueStat.getNodeConnectorId()))
-                                                .augmentation(FlowCapableNodeConnector.class)
-                                                .child(Queue.class, qKey);
-                                        final InstanceIdentifier<FlowCapableNodeConnectorQueueStatisticsData> queueStatIdent = queueIdent.augmentation(FlowCapableNodeConnectorQueueStatisticsData.class);
-                                        if (logFirstTime) {
-                                            logFirstTime = false;
-                                            LOG.info("Writing queue statistics to operational DS.");
-                                        }
-                                        deviceContext.writeToTransaction(LogicalDatastoreType.OPERATIONAL, queueStatIdent, statBuild.build());
-                                    }
-                                }
-                            }
-                            if (singleMultipartData instanceof FlowsStatisticsUpdate) {
-                                final FlowsStatisticsUpdate flowsStatistics = (FlowsStatisticsUpdate) singleMultipartData;
-                                final InstanceIdentifier<Node> nodeIdent = InstanceIdentifier.create(Nodes.class)
-                                        .child(Node.class, new NodeKey(flowsStatistics.getId()));
-
-                                for (final FlowAndStatisticsMapList flowStat : flowsStatistics.getFlowAndStatisticsMapList()) {
-                                    final FlowBuilder flowBuilder = new FlowBuilder(flowStat);
-                                    FlowId flowId = null;
-                                    FlowHash flowHash = FlowHashFactory.create(flowBuilder.build());
-                                    try {
-                                        flowId = deviceContext.getDeviceFlowRegistry().retrieveIdForFlow(flowHash);
-                                    } catch (FlowRegistryException e) {
-                                        flowId = FlowUtil.createAlienFlowId(flowStat.getTableId());
-                                        deviceContext.getDeviceFlowRegistry().store(flowHash, flowId);
-                                    }
-                                    FlowKey flowKey = new FlowKey(flowId);
-                                    flowBuilder.setKey(flowKey);
-                                    final TableKey tableKey = new TableKey(flowStat.getTableId());
-                                    final InstanceIdentifier<FlowCapableNode> fNodeIdent = nodeIdent.augmentation(FlowCapableNode.class);
-                                    final InstanceIdentifier<Flow> flowIdent = fNodeIdent.child(Table.class, tableKey).child(Flow.class, flowKey);
-                                    if (logFirstTime) {
-                                        logFirstTime = false;
-                                        LOG.info("Writing flow statistics to operational DS.");
-                                    }
-                                    deviceContext.writeToTransaction(LogicalDatastoreType.OPERATIONAL, flowIdent, flowBuilder.build());
-                                }
-                            }
-
-                            //TODO : implement experimenter
-                        }
+                        multipartData = multipartDataList.get(0);
+                        allMultipartData = Iterables.concat(allMultipartData, multipartDataList);
+                    }
+
+                    if (deviceContext.getDeviceState().isValid()) {
+                        deviceContext.startGatheringOperationsToOneTransaction();
+                    }
+
+                    if (multipartData instanceof GroupStatisticsUpdated) {
+                        processGroupStatistics((Iterable<GroupStatisticsUpdated>)allMultipartData, deviceContext);
+                    } else if (multipartData instanceof MeterStatisticsUpdated) {
+                        processMetersStatistics((Iterable<MeterStatisticsUpdated>)allMultipartData, deviceContext);
+                    } else if (multipartData instanceof NodeConnectorStatisticsUpdate) {
+                        processNodeConnectorStatistics((Iterable<NodeConnectorStatisticsUpdate>) allMultipartData, deviceContext);
+                    } else if (multipartData instanceof FlowTableStatisticsUpdate) {
+                        processFlowTableStatistics((Iterable<FlowTableStatisticsUpdate>) allMultipartData, deviceContext);
+                    } else if (multipartData instanceof QueueStatisticsUpdate) {
+                        processQueueStatistics((Iterable<QueueStatisticsUpdate>) allMultipartData, deviceContext);
+                    } else if (multipartData instanceof FlowsStatisticsUpdate) {
+                        processFlowStatistics((Iterable<FlowsStatisticsUpdate>)allMultipartData, deviceContext);
+                    } else if (multipartData instanceof GroupDescStatsUpdated) {
+                        processGroupDescStats((Iterable<GroupDescStatsUpdated>) allMultipartData, deviceContext);
+                    } else if (multipartData instanceof MeterConfigStatsUpdated) {
+                        processMeterConfigStatsUpdated((Iterable<MeterConfigStatsUpdated>) allMultipartData, deviceContext);
+                    } else {
+                        isMultipartProcessed = Boolean.FALSE;
+                    }
+                    //TODO : implement experimenter
+
+                    if (deviceContext.getDeviceState().isValid()) {
+                        deviceContext.commitOperationsGatheredInOneTransaction();
                     }
-                    return Boolean.TRUE;
+                    return isMultipartProcessed;
                 }
                 return Boolean.FALSE;
             }
         });
     }
+
+    private static void processMeterConfigStatsUpdated(final Iterable<MeterConfigStatsUpdated> data, final DeviceContext deviceContext) {
+        for (final MeterConfigStatsUpdated meterConfigStatsUpdated : data) {
+            NodeId nodeId = meterConfigStatsUpdated.getId();
+            final InstanceIdentifier<FlowCapableNode> fNodeIdent = getFlowCapableNodeInstanceIdentifier(nodeId);
+            for (MeterConfigStats meterConfigStats : meterConfigStatsUpdated.getMeterConfigStats()) {
+                final MeterId meterId = meterConfigStats.getMeterId();
+                final InstanceIdentifier<Meter> meterInstanceIdentifier = fNodeIdent.child(Meter.class, new MeterKey(meterId));
+    
+                final MeterBuilder meterBuilder = new MeterBuilder(meterConfigStats);
+                meterBuilder.setKey(new MeterKey(meterId));
+                meterBuilder.addAugmentation(NodeMeterStatistics.class, new NodeMeterStatisticsBuilder().build());
+                deviceContext.getDeviceMeterRegistry().store(meterId);
+                deviceContext.writeToTransaction(LogicalDatastoreType.OPERATIONAL, meterInstanceIdentifier, meterBuilder.build());
+            }
+        }
+    }
+
+    private static void processFlowStatistics(final Iterable<FlowsStatisticsUpdate> data, final DeviceContext deviceContext) {
+        boolean isFirstLoop = true;
+        for (final FlowsStatisticsUpdate flowsStatistics : data) {
+            if (isFirstLoop) {
+                final InstanceIdentifier<Node> nodeIdent = InstanceIdentifier.create(Nodes.class)
+                        .child(Node.class, new NodeKey(flowsStatistics.getId()));
+                deleteAllKnownFlows(deviceContext, nodeIdent);
+                isFirstLoop = false;
+            }
+            for (final FlowAndStatisticsMapList flowStat : flowsStatistics.getFlowAndStatisticsMapList()) {
+                final FlowBuilder flowBuilder = new FlowBuilder(flowStat);
+                FlowId flowId = null;
+                FlowHash flowHash = FlowHashFactory.create(flowBuilder.build(), deviceContext.getPrimaryConnectionContext().getFeatures().getVersion());
+                short tableId = flowStat.getTableId();
+                final DeviceFlowRegistry deviceFlowRegistry = deviceContext.getDeviceFlowRegistry();
+                FlowDescriptor flowDescriptor;
+                flowDescriptor = deviceFlowRegistry.retrieveIdForFlow(flowHash);
+                if (null != flowDescriptor) {
+                    flowId = flowDescriptor.getFlowId();
+                } else {
+                    LOG.trace("Flow descriptor for flow hash {} wasn't found.", flowHash.hashCode());
+                    flowId = FlowUtil.createAlienFlowId(tableId);
+                    flowDescriptor = FlowDescriptorFactory.create(tableId, flowId);
+                    deviceFlowRegistry.store(flowHash, flowDescriptor);
+                }
+                FlowKey flowKey = new FlowKey(flowId);
+                flowBuilder.setKey(flowKey);
+                final TableKey tableKey = new TableKey(tableId);
+                final InstanceIdentifier<FlowCapableNode> fNodeIdent = getFlowCapableNodeInstanceIdentifier(flowsStatistics.getId());
+                final InstanceIdentifier<Flow> flowIdent = fNodeIdent.child(Table.class, tableKey).child(Flow.class, flowKey);
+                deviceContext.writeToTransaction(LogicalDatastoreType.OPERATIONAL, flowIdent, flowBuilder.build());
+            }
+        }
+    }
+
+    private static void deleteAllKnownFlows(final DeviceContext deviceContext, final InstanceIdentifier<Node> nodeIdent) {
+        if (deviceContext.getDeviceState().deviceSynchronized()) {
+            for (Map.Entry<FlowHash, FlowDescriptor> registryEntry : deviceContext.getDeviceFlowRegistry().getAllFlowDescriptors().entrySet()) {
+                FlowDescriptor flowDescriptor = registryEntry.getValue();
+
+                FlowId flowId = flowDescriptor.getFlowId();
+                FlowKey flowKey = new FlowKey(flowId);
+                final InstanceIdentifier<Flow> flowInstanceIdentifier = nodeIdent
+                        .augmentation(FlowCapableNode.class)
+                        .child(Table.class, flowDescriptor.getTableKey())
+                        .child(Flow.class, flowKey);
+
+                LOG.trace("Deleting flow with id {}", flowInstanceIdentifier);
+                deviceContext.addDeleteToTxChain(LogicalDatastoreType.OPERATIONAL, flowInstanceIdentifier);
+            }
+        }
+        deviceContext.getDeviceFlowRegistry().removeMarked();
+    }
+
+    private static void processQueueStatistics(final Iterable<QueueStatisticsUpdate> data, final DeviceContext deviceContext) {
+        for (final QueueStatisticsUpdate queueStatisticsUpdate : data) {
+            final InstanceIdentifier<Node> nodeIdent = InstanceIdentifier.create(Nodes.class)
+                    .child(Node.class, new NodeKey(queueStatisticsUpdate.getId()));
+            for (final QueueIdAndStatisticsMap queueStat : queueStatisticsUpdate.getQueueIdAndStatisticsMap()) {
+                if (queueStat.getQueueId() != null) {
+                    final FlowCapableNodeConnectorQueueStatistics statChild =
+                            new FlowCapableNodeConnectorQueueStatisticsBuilder(queueStat).build();
+                    final FlowCapableNodeConnectorQueueStatisticsDataBuilder statBuild =
+                            new FlowCapableNodeConnectorQueueStatisticsDataBuilder();
+                    statBuild.setFlowCapableNodeConnectorQueueStatistics(statChild);
+                    final QueueKey qKey = new QueueKey(queueStat.getQueueId());
+                    final InstanceIdentifier<Queue> queueIdent = nodeIdent
+                            .child(NodeConnector.class, new NodeConnectorKey(queueStat.getNodeConnectorId()))
+                            .augmentation(FlowCapableNodeConnector.class)
+                            .child(Queue.class, qKey);
+                    final InstanceIdentifier<FlowCapableNodeConnectorQueueStatisticsData> queueStatIdent = queueIdent.augmentation(FlowCapableNodeConnectorQueueStatisticsData.class);
+                    deviceContext.writeToTransaction(LogicalDatastoreType.OPERATIONAL, queueStatIdent, statBuild.build());
+                }
+            }
+        }
+    }
+
+    private static void processFlowTableStatistics(final Iterable<FlowTableStatisticsUpdate> data, final DeviceContext deviceContext) {
+        for(final FlowTableStatisticsUpdate flowTableStatisticsUpdate : data) {
+            final InstanceIdentifier<FlowCapableNode> fNodeIdent = getFlowCapableNodeInstanceIdentifier(flowTableStatisticsUpdate.getId());
+    
+            for (final FlowTableAndStatisticsMap tableStat : flowTableStatisticsUpdate.getFlowTableAndStatisticsMap()) {
+                final InstanceIdentifier<FlowTableStatistics> tStatIdent = fNodeIdent.child(Table.class, new TableKey(tableStat.getTableId().getValue()))
+                        .augmentation(FlowTableStatisticsData.class).child(FlowTableStatistics.class);
+                final FlowTableStatistics stats = new FlowTableStatisticsBuilder(tableStat).build();
+                deviceContext.writeToTransaction(LogicalDatastoreType.OPERATIONAL, tStatIdent, stats);
+            }
+        }
+    }
+
+    private static void processNodeConnectorStatistics(final Iterable<NodeConnectorStatisticsUpdate> data, final DeviceContext deviceContext) {
+        for (final NodeConnectorStatisticsUpdate nodeConnectorStatisticsUpdate : data) {
+            final InstanceIdentifier<Node> nodeIdent = InstanceIdentifier.create(Nodes.class)
+                    .child(Node.class, new NodeKey(nodeConnectorStatisticsUpdate.getId()));
+            for (final NodeConnectorStatisticsAndPortNumberMap nConnectPort : nodeConnectorStatisticsUpdate.getNodeConnectorStatisticsAndPortNumberMap()) {
+                final FlowCapableNodeConnectorStatistics stats = new FlowCapableNodeConnectorStatisticsBuilder(nConnectPort).build();
+                final NodeConnectorKey key = new NodeConnectorKey(nConnectPort.getNodeConnectorId());
+                final InstanceIdentifier<NodeConnector> nodeConnectorIdent = nodeIdent.child(NodeConnector.class, key);
+                final InstanceIdentifier<FlowCapableNodeConnectorStatisticsData> nodeConnStatIdent = nodeConnectorIdent
+                        .augmentation(FlowCapableNodeConnectorStatisticsData.class);
+                final InstanceIdentifier<FlowCapableNodeConnectorStatistics> flowCapNodeConnStatIdent =
+                        nodeConnStatIdent.child(FlowCapableNodeConnectorStatistics.class);
+                deviceContext.writeToTransaction(LogicalDatastoreType.OPERATIONAL, flowCapNodeConnStatIdent, stats);
+            }
+        }
+    }
+
+    private static void processMetersStatistics(final Iterable<MeterStatisticsUpdated> data,
+                                                final DeviceContext deviceContext) {
+        boolean isFirstLoop = true;
+        for (MeterStatisticsUpdated meterStatisticsUpdated : data) {
+            final InstanceIdentifier<FlowCapableNode> fNodeIdent = getFlowCapableNodeInstanceIdentifier(meterStatisticsUpdated.getId());
+            if (isFirstLoop) {
+                deleteAllKnownMeters(deviceContext, fNodeIdent);
+                isFirstLoop = false;
+            }
+            for (final MeterStats mStat : meterStatisticsUpdated.getMeterStats()) {
+                final MeterStatistics stats = new MeterStatisticsBuilder(mStat).build();
+                final MeterId meterId = mStat.getMeterId();
+                final InstanceIdentifier<Meter> meterIdent = fNodeIdent.child(Meter.class, new MeterKey(meterId));
+                final InstanceIdentifier<NodeMeterStatistics> nodeMeterStatIdent = meterIdent
+                        .augmentation(NodeMeterStatistics.class);
+                final InstanceIdentifier<MeterStatistics> msIdent = nodeMeterStatIdent.child(MeterStatistics.class);
+                deviceContext.writeToTransaction(LogicalDatastoreType.OPERATIONAL, msIdent, stats);
+            }
+        }
+    }
+
+    private static void deleteAllKnownMeters(final DeviceContext deviceContext, final InstanceIdentifier<FlowCapableNode> fNodeIdent) {
+        for (MeterId meterId : deviceContext.getDeviceMeterRegistry().getAllMeterIds()) {
+            final InstanceIdentifier<Meter> meterIdent = fNodeIdent.child(Meter.class, new MeterKey(meterId));
+            deviceContext.addDeleteToTxChain(LogicalDatastoreType.OPERATIONAL, meterIdent);
+        }
+        deviceContext.getDeviceMeterRegistry().removeMarked();
+    }
+
+    private static void processGroupDescStats(final Iterable<GroupDescStatsUpdated> data, final DeviceContext deviceContext) {
+        for ( GroupDescStatsUpdated groupDescStatsUpdated : data) {
+            NodeId nodeId = groupDescStatsUpdated.getId();
+            final InstanceIdentifier<FlowCapableNode> fNodeIdent = getFlowCapableNodeInstanceIdentifier(nodeId);
+    
+            for (GroupDescStats groupDescStats : groupDescStatsUpdated.getGroupDescStats()) {
+                final GroupId groupId = groupDescStats.getGroupId();
+    
+                final GroupBuilder groupBuilder = new GroupBuilder(groupDescStats);
+                groupBuilder.setKey(new GroupKey(groupId));
+                groupBuilder.addAugmentation(NodeGroupStatistics.class, new NodeGroupStatisticsBuilder().build());
+    
+                final InstanceIdentifier<Group> groupIdent = fNodeIdent.child(Group.class, new GroupKey(groupId));
+    
+                deviceContext.getDeviceGroupRegistry().store(groupId);
+                deviceContext.writeToTransaction(LogicalDatastoreType.OPERATIONAL, groupIdent, groupBuilder.build());
+            }
+        }
+    }
+
+    private static void deleteAllKnownGroups(final DeviceContext deviceContext, final InstanceIdentifier<FlowCapableNode> fNodeIdent) {
+        for (GroupId groupId : deviceContext.getDeviceGroupRegistry().getAllGroupIds()) {
+            final InstanceIdentifier<Group> groupIdent = fNodeIdent.child(Group.class, new GroupKey(groupId));
+            deviceContext.addDeleteToTxChain(LogicalDatastoreType.OPERATIONAL, groupIdent);
+        }
+        deviceContext.getDeviceGroupRegistry().removeMarked();
+    }
+
+    private static void processGroupStatistics(final Iterable<GroupStatisticsUpdated> data, final DeviceContext deviceContext) {
+        boolean isFirstLoop = true; 
+        for (GroupStatisticsUpdated groupStatistics : data) {
+            NodeId nodeId = groupStatistics.getId();
+            final InstanceIdentifier<FlowCapableNode> fNodeIdent = getFlowCapableNodeInstanceIdentifier(nodeId);
+            if (isFirstLoop) {
+                deleteAllKnownGroups(deviceContext, fNodeIdent);
+                isFirstLoop = false;
+            }
+    
+            for (final GroupStats groupStats : groupStatistics.getGroupStats()) {
+    
+                final InstanceIdentifier<Group> groupIdent = fNodeIdent.child(Group.class, new GroupKey(groupStats.getGroupId()));
+                final InstanceIdentifier<NodeGroupStatistics> nGroupStatIdent = groupIdent
+                        .augmentation(NodeGroupStatistics.class);
+    
+                final InstanceIdentifier<GroupStatistics> gsIdent = nGroupStatIdent.child(GroupStatistics.class);
+                final GroupStatistics stats = new GroupStatisticsBuilder(groupStats).build();
+                deviceContext.writeToTransaction(LogicalDatastoreType.OPERATIONAL, gsIdent, stats);
+            }
+        }
+    }
+
+    private static InstanceIdentifier<FlowCapableNode> getFlowCapableNodeInstanceIdentifier(final NodeId nodeId) {
+        final InstanceIdentifier<Node> nodeIdent = InstanceIdentifier
+                .create(Nodes.class).child(Node.class, new NodeKey(nodeId));
+        return nodeIdent.augmentation(FlowCapableNode.class);
+    }
 }