Add single layer deserialization support
[openflowplugin.git] / openflowplugin-impl / src / main / java / org / opendaylight / openflowplugin / impl / device / DeviceManagerImpl.java
index ef850d972f48018a4b172dc3666b430264a758cd..df49b1443769a6145fa754673ab673114b22c2b1 100644 (file)
@@ -9,8 +9,8 @@ package org.opendaylight.openflowplugin.impl.device;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
-import com.google.common.base.Verify;
 import com.google.common.collect.Iterators;
+import com.google.common.util.concurrent.CheckedFuture;
 import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
@@ -18,6 +18,8 @@ import io.netty.util.HashedWheelTimer;
 import io.netty.util.TimerTask;
 import java.util.Collections;
 import java.util.Iterator;
+import java.util.Objects;
+import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ExecutionException;
@@ -25,15 +27,18 @@ import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import javax.annotation.CheckForNull;
 import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
 import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
 import org.opendaylight.controller.md.sal.binding.api.WriteTransaction;
 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
+import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceProvider;
 import org.opendaylight.openflowjava.protocol.api.connection.ConnectionAdapter;
 import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueueHandlerRegistration;
 import org.opendaylight.openflowplugin.api.openflow.OFPContext;
 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
+import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionStatus;
 import org.opendaylight.openflowplugin.api.openflow.connection.OutboundQueueProvider;
 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
 import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo;
@@ -46,14 +51,17 @@ import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.Messa
 import org.opendaylight.openflowplugin.extension.api.ExtensionConverterProviderKeeper;
 import org.opendaylight.openflowplugin.extension.api.core.extension.ExtensionConverterProvider;
 import org.opendaylight.openflowplugin.impl.connection.OutboundQueueProviderImpl;
+import org.opendaylight.openflowplugin.impl.device.initialization.DeviceInitializerProvider;
 import org.opendaylight.openflowplugin.impl.device.listener.OpenflowProtocolListenerFullImpl;
 import org.opendaylight.openflowplugin.impl.lifecycle.LifecycleServiceImpl;
-import org.opendaylight.openflowplugin.impl.util.DeviceInitializationUtils;
+import org.opendaylight.openflowplugin.impl.services.sal.SalRoleServiceImpl;
 import org.opendaylight.openflowplugin.openflow.md.core.sal.convertor.ConvertorExecutor;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodesBuilder;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.SetRoleOutput;
 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
+import org.opendaylight.yangtools.yang.common.RpcResult;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -66,11 +74,12 @@ public class DeviceManagerImpl implements DeviceManager, ExtensionConverterProvi
 
     private final long globalNotificationQuota;
     private final boolean switchFeaturesMandatory;
-    private boolean isNotificationFlowRemovedOff;
-
+    private boolean isFlowRemovedNotificationOn;
+    private boolean skipTableFeatures;
     private static final int SPY_RATE = 10;
 
     private final DataBroker dataBroker;
+    private final DeviceInitializerProvider deviceInitializerProvider;
     private final ConvertorExecutor convertorExecutor;
     private TranslatorLibrary translatorLibrary;
     private DeviceInitializationPhaseHandler deviceInitPhaseHandler;
@@ -79,36 +88,37 @@ public class DeviceManagerImpl implements DeviceManager, ExtensionConverterProvi
     private final ConcurrentMap<DeviceInfo, DeviceContext> deviceContexts = new ConcurrentHashMap<>();
     private final ConcurrentMap<DeviceInfo, LifecycleService> lifecycleServices = new ConcurrentHashMap<>();
 
-    private final long barrierIntervalNanos;
-    private final int barrierCountLimit;
+    private long barrierIntervalNanos;
+    private int barrierCountLimit;
+
     private ExtensionConverterProvider extensionConverterProvider;
     private ScheduledThreadPoolExecutor spyPool;
     private final ClusterSingletonServiceProvider singletonServiceProvider;
     private final NotificationPublishService notificationPublishService;
     private final MessageSpy messageSpy;
     private final HashedWheelTimer hashedWheelTimer;
+    private final boolean useSingleLayerSerialization;
 
     public DeviceManagerImpl(@Nonnull final DataBroker dataBroker,
                              final long globalNotificationQuota,
                              final boolean switchFeaturesMandatory,
                              final long barrierInterval,
                              final int barrierCountLimit,
-                            
                              final MessageSpy messageSpy,
-                             final boolean isNotificationFlowRemovedOff,
+                             final boolean isFlowRemovedNotificationOn,
                              final ClusterSingletonServiceProvider singletonServiceProvider,
                              final NotificationPublishService notificationPublishService,
                              final HashedWheelTimer hashedWheelTimer,
-                            final ConvertorExecutor convertorExecutor) {
-        this.switchFeaturesMandatory = switchFeaturesMandatory;
-        this.globalNotificationQuota = globalNotificationQuota;
-        this.isNotificationFlowRemovedOff = isNotificationFlowRemovedOff;
-        this.dataBroker = Preconditions.checkNotNull(dataBroker);
-        this.convertorExecutor = convertorExecutor;
-        this.hashedWheelTimer = hashedWheelTimer;
+                             final ConvertorExecutor convertorExecutor,
+                             final boolean skipTableFeatures,
+                             final boolean useSingleLayerSerialization,
+                             final DeviceInitializerProvider deviceInitializerProvider) {
+
+        this.dataBroker = dataBroker;
+        this.deviceInitializerProvider = deviceInitializerProvider;
+
         /* merge empty nodes to oper DS to predict any problems with missing parent for Node */
         final WriteTransaction tx = dataBroker.newWriteOnlyTransaction();
-
         final NodesBuilder nodesBuilder = new NodesBuilder();
         nodesBuilder.setNode(Collections.<Node>emptyList());
         tx.merge(LogicalDatastoreType.OPERATIONAL, InstanceIdentifier.create(Nodes.class), nodesBuilder.build());
@@ -119,13 +129,19 @@ public class DeviceManagerImpl implements DeviceManager, ExtensionConverterProvi
             throw new IllegalStateException(e);
         }
 
+        this.switchFeaturesMandatory = switchFeaturesMandatory;
+        this.globalNotificationQuota = globalNotificationQuota;
+        this.isFlowRemovedNotificationOn = isFlowRemovedNotificationOn;
+        this.skipTableFeatures = skipTableFeatures;
+        this.convertorExecutor = convertorExecutor;
+        this.hashedWheelTimer = hashedWheelTimer;
         this.barrierIntervalNanos = TimeUnit.MILLISECONDS.toNanos(barrierInterval);
         this.barrierCountLimit = barrierCountLimit;
-
-        spyPool = new ScheduledThreadPoolExecutor(1);
+        this.spyPool = new ScheduledThreadPoolExecutor(1);
         this.singletonServiceProvider = singletonServiceProvider;
         this.notificationPublishService = notificationPublishService;
         this.messageSpy = messageSpy;
+        this.useSingleLayerSerialization = useSingleLayerSerialization;
     }
 
 
@@ -140,33 +156,42 @@ public class DeviceManagerImpl implements DeviceManager, ExtensionConverterProvi
         LOG.debug("Final phase of DeviceContextLevelUp for Node: {} ", deviceInfo.getNodeId());
         DeviceContext deviceContext = Preconditions.checkNotNull(deviceContexts.get(deviceInfo));
         deviceContext.onPublished();
+        lifecycleService.registerDeviceRemovedHandler(this);
         lifecycleService.registerService(this.singletonServiceProvider);
     }
 
     @Override
-    public boolean deviceConnected(@CheckForNull final ConnectionContext connectionContext) throws Exception {
+    public ConnectionStatus deviceConnected(@CheckForNull final ConnectionContext connectionContext) throws Exception {
         Preconditions.checkArgument(connectionContext != null);
+        final DeviceInfo deviceInfo = connectionContext.getDeviceInfo();
 
-        DeviceInfo deviceInfo = connectionContext.getDeviceInfo();
         /*
          * This part prevent destroy another device context. Throwing here an exception result to propagate close connection
          * in {@link org.opendaylight.openflowplugin.impl.connection.org.opendaylight.openflowplugin.impl.connection.HandshakeContextImpl}
          * If context already exist we are in state closing process (connection flapping) and we should not propagate connection close
          */
          if (deviceContexts.containsKey(deviceInfo)) {
-            LOG.warn("Rejecting connection from node which is already connected and there exist deviceContext for it: {}", connectionContext.getNodeId());
-             return false;
+             DeviceContext deviceContext = deviceContexts.get(deviceInfo);
+             LOG.warn("Node {} already connected disconnecting device. Rejecting connection", deviceInfo.getLOGValue());
+             if (!deviceContext.getState().equals(OFPContext.CONTEXT_STATE.TERMINATION)) {
+                 LOG.warn("Node {} context state not in TERMINATION state.",
+                         connectionContext.getDeviceInfo().getLOGValue());
+                 return ConnectionStatus.ALREADY_CONNECTED;
+             } else {
+                 return ConnectionStatus.CLOSING;
+             }
          }
 
         LOG.info("ConnectionEvent: Device connected to controller, Device:{}, NodeId:{}",
                 connectionContext.getConnectionAdapter().getRemoteAddress(), deviceInfo.getNodeId());
 
         // Add Disconnect handler
-        connectionContext.setDeviceDisconnectedHandler(DeviceManagerImpl.this);
+        connectionContext.setDeviceDisconnectedHandler(this);
+
         // Cache this for clarity
         final ConnectionAdapter connectionAdapter = connectionContext.getConnectionAdapter();
 
-        //FIXME: as soon as auxiliary connection are fully supported then this is needed only before device context published
+        // FIXME: as soon as auxiliary connection are fully supported then this is needed only before device context published
         connectionAdapter.setPacketInFiltering(true);
 
         final OutboundQueueProvider outboundQueueProvider = new OutboundQueueProviderImpl(deviceInfo.getVersion());
@@ -176,20 +201,29 @@ public class DeviceManagerImpl implements DeviceManager, ExtensionConverterProvi
                 connectionAdapter.registerOutboundQueueHandler(outboundQueueProvider, barrierCountLimit, barrierIntervalNanos);
         connectionContext.setOutboundQueueHandleRegistration(outboundQueueHandlerRegistration);
 
+        final LifecycleService lifecycleService = new LifecycleServiceImpl();
         final DeviceContext deviceContext = new DeviceContextImpl(
                 connectionContext,
                 dataBroker,
                 messageSpy,
                 translatorLibrary,
                 this,
-                convertorExecutor);
+                convertorExecutor,
+                skipTableFeatures,
+                hashedWheelTimer,
+                this,
+                useSingleLayerSerialization,
+                deviceInitializerProvider);
 
-        Verify.verify(deviceContexts.putIfAbsent(deviceInfo, deviceContext) == null, "DeviceCtx still not closed.");
+        deviceContext.setSalRoleService(new SalRoleServiceImpl(deviceContext, deviceContext));
+        deviceContexts.put(deviceInfo, deviceContext);
 
-        final LifecycleService lifecycleService = new LifecycleServiceImpl();
         lifecycleService.setDeviceContext(deviceContext);
+        deviceContext.putLifecycleServiceIntoTxChainManager(lifecycleService);
+
+        lifecycleServices.put(deviceInfo, lifecycleService);
 
-        lifecycleServices.putIfAbsent(deviceInfo, lifecycleService);
+        addCallbackToDeviceInitializeToSlave(deviceInfo, deviceContext, lifecycleService);
 
         deviceContext.setSwitchFeaturesMandatory(switchFeaturesMandatory);
 
@@ -203,23 +237,7 @@ public class DeviceManagerImpl implements DeviceManager, ExtensionConverterProvi
 
         connectionAdapter.setMessageListener(messageListener);
         deviceInitPhaseHandler.onDeviceContextLevelUp(connectionContext.getDeviceInfo(), lifecycleService);
-        return true;
-    }
-
-    private void updatePacketInRateLimiters() {
-        synchronized (deviceContexts) {
-            final int deviceContextsSize = deviceContexts.size();
-            if (deviceContextsSize > 0) {
-                long freshNotificationLimit = globalNotificationQuota / deviceContextsSize;
-                if (freshNotificationLimit < 100) {
-                    freshNotificationLimit = 100;
-                }
-                LOG.debug("fresh notification limit = {}", freshNotificationLimit);
-                for (final DeviceContext deviceContext : deviceContexts.values()) {
-                    deviceContext.updatePacketInRateLimit(freshNotificationLimit);
-                }
-            }
-        }
+        return ConnectionStatus.MAY_CONTINUE;
     }
 
     @Override
@@ -241,23 +259,15 @@ public class DeviceManagerImpl implements DeviceManager, ExtensionConverterProvi
             deviceCtx.shuttingDownDataStoreTransactions();
         }
 
-        if (spyPool != null) {
-            spyPool.shutdownNow();
-            spyPool = null;
-        }
+        Optional.ofNullable(spyPool).ifPresent(ScheduledThreadPoolExecutor::shutdownNow);
+        spyPool = null;
+
     }
 
     @Override
     public void onDeviceContextLevelDown(final DeviceInfo deviceInfo) {
-        LOG.debug("onDeviceContextClosed for Node {}", deviceInfo.getNodeId());
-        deviceContexts.remove(deviceInfo);
         updatePacketInRateLimiters();
-        LifecycleService lifecycleService = lifecycleServices.remove(deviceInfo);
-        try {
-            lifecycleService.close();
-        } catch (Exception e) {
-            LOG.warn("Closing service for node {} was unsuccessful ", deviceInfo.getNodeId().getValue(), e);
-        }
+        Optional.ofNullable(lifecycleServices.get(deviceInfo)).ifPresent(OFPContext::close);
     }
 
     @Override
@@ -286,40 +296,51 @@ public class DeviceManagerImpl implements DeviceManager, ExtensionConverterProvi
         final DeviceInfo deviceInfo = connectionContext.getDeviceInfo();
         final DeviceContext deviceCtx = this.deviceContexts.get(deviceInfo);
 
-        if (null == deviceCtx) {
-            LOG.info("DeviceContext for Node {} was not found. Connection is terminated without OFP context suite.", deviceInfo.getNodeId());
+        if (Objects.isNull(deviceCtx)) {
+            LOG.info("DeviceContext for Node {} was not found. Connection is terminated without OFP context suite.", deviceInfo.getLOGValue());
+            return;
+        }
+
+        if (deviceCtx.getState().equals(OFPContext.CONTEXT_STATE.TERMINATION)) {
+            LOG.info("Device context for node {} is already is termination state, waiting for close all context", deviceInfo.getLOGValue());
             return;
         }
 
+        deviceCtx.close();
+
         if (!connectionContext.equals(deviceCtx.getPrimaryConnectionContext())) {
-            /* Connection is not PrimaryConnection so try to remove from Auxiliary Connections */
+            LOG.debug("Node {} disconnected, but not primary connection.", connectionContext.getDeviceInfo().getLOGValue());
+            // Connection is not PrimaryConnection so try to remove from Auxiliary Connections
             deviceCtx.removeAuxiliaryConnectionContext(connectionContext);
-        } else {
-            /* Device is disconnected and so we need to close TxManager */
-            final ListenableFuture<Void> future = deviceCtx.shuttingDownDataStoreTransactions();
-            Futures.addCallback(future, new FutureCallback<Void>() {
-
-                @Override
-                public void onSuccess(final Void result) {
-                    LOG.debug("TxChainManager for device {} is closed successful.", deviceInfo.getNodeId());
-                    deviceTerminPhaseHandler.onDeviceContextLevelDown(deviceInfo);
-                }
-
-                @Override
-                public void onFailure(final Throwable t) {
-                    LOG.warn("TxChainManager for device {} failed by closing.", deviceInfo.getNodeId(), t);
-                    deviceTerminPhaseHandler.onDeviceContextLevelDown(deviceInfo);
-                }
-            });
-            /* Add timer for Close TxManager because it could fain ind cluster without notification */
-            final TimerTask timerTask = timeout -> {
-                if (!future.isDone()) {
-                    LOG.info("Shutting down TxChain for node {} not completed during 10 sec. Continue anyway.", deviceInfo.getNodeId());
-                    future.cancel(false);
-                }
-            };
-            hashedWheelTimer.newTimeout(timerTask, 10, TimeUnit.SECONDS);
         }
+
+        // TODO: Auxiliary connections supported ?
+        // Device is disconnected and so we need to close TxManager
+        final ListenableFuture<Void> future = deviceCtx.shuttingDownDataStoreTransactions();
+        Futures.addCallback(future, new FutureCallback<Void>() {
+            @Override
+            public void onSuccess(final Void result) {
+                LOG.debug("TxChainManager for device {} is closed successful.", deviceInfo.getLOGValue());
+                deviceTerminPhaseHandler.onDeviceContextLevelDown(deviceInfo);
+            }
+
+            @Override
+            public void onFailure(final Throwable t) {
+                LOG.warn("TxChainManager for device {} failed by closing.", deviceInfo.getLOGValue());
+                LOG.trace("TxChainManager failed by closing. ", t);
+                deviceTerminPhaseHandler.onDeviceContextLevelDown(deviceInfo);
+            }
+        });
+
+        // Add timer for Close TxManager because it could fail in cluster without notification
+        final TimerTask timerTask = timeout -> {
+            if (!future.isDone()) {
+                LOG.warn("Shutting down TxChain for node {} not completed during 10 sec. Continue anyway.", deviceInfo.getLOGValue());
+                future.cancel(false);
+            }
+        };
+
+        hashedWheelTimer.newTimeout(timerTask, 10, TimeUnit.SECONDS);
     }
 
     @VisibleForTesting
@@ -327,24 +348,96 @@ public class DeviceManagerImpl implements DeviceManager, ExtensionConverterProvi
         deviceContexts.put(deviceInfo, deviceContext);
     }
 
-    @VisibleForTesting
-    void removeDeviceContextFromMap(final DeviceInfo deviceInfo){
-        deviceContexts.remove(deviceInfo);
+    @Override
+    public void setFlowRemovedNotificationOn(boolean isNotificationFlowRemovedOff) {
+        this.isFlowRemovedNotificationOn = isNotificationFlowRemovedOff;
     }
 
     @Override
-    public <T extends OFPContext> T gainContext(final DeviceInfo deviceInfo) {
-        return (T) deviceContexts.get(deviceInfo);
+    public boolean isFlowRemovedNotificationOn() {
+        return this.isFlowRemovedNotificationOn;
     }
 
+
     @Override
-    public void setIsNotificationFlowRemovedOff(boolean isNotificationFlowRemovedOff) {
-        this.isNotificationFlowRemovedOff = isNotificationFlowRemovedOff;
+    public void setSkipTableFeatures(boolean skipTableFeaturesValue) {
+        skipTableFeatures = skipTableFeaturesValue;
     }
 
     @Override
-    public boolean getIsNotificationFlowRemovedOff() {
-        return this.isNotificationFlowRemovedOff;
+    public void setBarrierCountLimit(final int barrierCountLimit) {
+        this.barrierCountLimit = barrierCountLimit;
     }
 
+    @Override
+    public void setBarrierInterval(final long barrierTimeoutLimit) {
+        this.barrierIntervalNanos = TimeUnit.MILLISECONDS.toNanos(barrierTimeoutLimit);
+    }
+
+    @Override
+    public CheckedFuture<Void, TransactionCommitFailedException> removeDeviceFromOperationalDS(final DeviceInfo deviceInfo) {
+        final WriteTransaction delWtx = dataBroker.newWriteOnlyTransaction();
+        delWtx.delete(LogicalDatastoreType.OPERATIONAL, deviceInfo.getNodeInstanceIdentifier());
+        final CheckedFuture<Void, TransactionCommitFailedException> delFuture = delWtx.submit();
+
+        Futures.addCallback(delFuture, new FutureCallback<Void>() {
+            @Override
+            public void onSuccess(final Void result) {
+                if (LOG.isDebugEnabled()) {
+                    LOG.debug("Delete Node {} was successful", deviceInfo.getLOGValue());
+                }
+            }
+
+            @Override
+            public void onFailure(@Nonnull final Throwable t) {
+                LOG.warn("Delete node {} failed with exception {}", deviceInfo.getLOGValue(), t);
+            }
+        });
+
+        return delFuture;
+    }
+
+
+    private void addCallbackToDeviceInitializeToSlave(final DeviceInfo deviceInfo, final DeviceContext deviceContext, final LifecycleService lifecycleService) {
+        Futures.addCallback(deviceContext.makeDeviceSlave(), new FutureCallback<RpcResult<SetRoleOutput>>() {
+            @Override
+            public void onSuccess(@Nullable RpcResult<SetRoleOutput> setRoleOutputRpcResult) {
+                if (LOG.isDebugEnabled()) {
+                    LOG.debug("Role SLAVE was successfully propagated on device, node {}", deviceInfo.getLOGValue());
+                }
+            }
+
+            @Override
+            public void onFailure(Throwable throwable) {
+                LOG.warn("Was not able to set role SLAVE to device on node {} ",deviceInfo.getLOGValue());
+                lifecycleService.closeConnection();
+            }
+        });
+    }
+
+    private void updatePacketInRateLimiters() {
+        synchronized (deviceContexts) {
+            final int deviceContextsSize = deviceContexts.size();
+            if (deviceContextsSize > 0) {
+                long freshNotificationLimit = globalNotificationQuota / deviceContextsSize;
+                if (freshNotificationLimit < 100) {
+                    freshNotificationLimit = 100;
+                }
+                if (LOG.isDebugEnabled()) {
+                    LOG.debug("fresh notification limit = {}", freshNotificationLimit);
+                }
+                for (final DeviceContext deviceContext : deviceContexts.values()) {
+                    deviceContext.updatePacketInRateLimit(freshNotificationLimit);
+                }
+            }
+        }
+    }
+
+    public void onDeviceRemoved(DeviceInfo deviceInfo) {
+        deviceContexts.remove(deviceInfo);
+        LOG.debug("Device context removed for node {}", deviceInfo.getLOGValue());
+
+        lifecycleServices.remove(deviceInfo);
+        LOG.debug("Lifecycle service removed for node {}", deviceInfo.getLOGValue());
+    }
 }