Merge "Bug 5976 - Thread leak when connecting/disconnecting ovs nodes"
[ovsdb.git] / southbound / southbound-impl / src / main / java / org / opendaylight / ovsdb / southbound / OvsdbConnectionManager.java
index 2ac0f496e23754a2b5c78f6c7035f9d88ba788cb..5740ab573feeb40235a005df9193f1ff055e101f 100644 (file)
@@ -16,9 +16,14 @@ import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
 
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
 import org.opendaylight.controller.md.sal.binding.api.ReadOnlyTransaction;
 import org.opendaylight.controller.md.sal.binding.api.ReadWriteTransaction;
@@ -42,7 +47,12 @@ import org.opendaylight.ovsdb.lib.schema.DatabaseSchema;
 import org.opendaylight.ovsdb.lib.schema.GenericTableSchema;
 import org.opendaylight.ovsdb.lib.schema.typed.TyperUtils;
 import org.opendaylight.ovsdb.schema.openvswitch.OpenVSwitch;
+import org.opendaylight.ovsdb.southbound.reconciliation.ReconciliationManager;
+import org.opendaylight.ovsdb.southbound.reconciliation.ReconciliationTask;
+import org.opendaylight.ovsdb.southbound.reconciliation.configuration.BridgeConfigReconciliationTask;
+import org.opendaylight.ovsdb.southbound.reconciliation.connection.ConnectionReconciliationTask;
 import org.opendaylight.ovsdb.southbound.transactions.md.OvsdbNodeRemoveCommand;
+import org.opendaylight.ovsdb.southbound.transactions.md.TransactionCommand;
 import org.opendaylight.ovsdb.southbound.transactions.md.TransactionInvoker;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.ovsdb.rev150105.OvsdbBridgeAttributes;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.ovsdb.rev150105.OvsdbBridgeAugmentation;
@@ -74,6 +84,7 @@ public class OvsdbConnectionManager implements OvsdbConnectionListener, AutoClos
     private EntityOwnershipService entityOwnershipService;
     private OvsdbDeviceEntityOwnershipListener ovsdbDeviceEntityOwnershipListener;
     private OvsdbConnection ovsdbConnection;
+    private final ReconciliationManager reconciliationManager;
 
     public OvsdbConnectionManager(DataBroker db,TransactionInvoker txInvoker,
                                   EntityOwnershipService entityOwnershipService,
@@ -83,6 +94,7 @@ public class OvsdbConnectionManager implements OvsdbConnectionListener, AutoClos
         this.entityOwnershipService = entityOwnershipService;
         this.ovsdbDeviceEntityOwnershipListener = new OvsdbDeviceEntityOwnershipListener(this, entityOwnershipService);
         this.ovsdbConnection = ovsdbConnection;
+        this.reconciliationManager = new ReconciliationManager(db);
     }
 
     @Override
@@ -93,10 +105,20 @@ public class OvsdbConnectionManager implements OvsdbConnectionListener, AutoClos
                 externalClient.getConnectionInfo().getRemotePort(),
                 externalClient.getConnectionInfo().getLocalAddress(),
                 externalClient.getConnectionInfo().getLocalPort());
-        OvsdbConnectionInstance client = connectedButCallBacksNotRegistered(externalClient);
+        List<String> databases = new ArrayList<>();
+        try {
+            databases = externalClient.getDatabases().get(1000, TimeUnit.MILLISECONDS);
+            if(databases.contains(SouthboundConstants.OPEN_V_SWITCH)) {
+                OvsdbConnectionInstance client = connectedButCallBacksNotRegistered(externalClient);
+                // Register Cluster Ownership for ConnectionInfo
+                registerEntityForOwnership(client);
+            }
+        } catch (InterruptedException | ExecutionException | TimeoutException e) {
+            LOG.warn("Unable to fetch Database list from device {}. Disconnecting from the device.",
+                    externalClient.getConnectionInfo().getRemoteAddress(), e);
+            externalClient.disconnect();
+        }
 
-        // Register Cluster Ownership for ConnectionInfo
-        registerEntityForOwnership(client);
     }
 
     public OvsdbConnectionInstance connectedButCallBacksNotRegistered(final OvsdbClient externalClient) {
@@ -121,6 +143,8 @@ public class OvsdbConnectionManager implements OvsdbConnectionListener, AutoClos
             ovsdbConnectionInstance.disconnect();
 
             removeConnectionInstance(key);
+
+            stopBridgeConfigReconciliationIfActive(ovsdbConnectionInstance.getInstanceIdentifier());
         }
 
         ovsdbConnectionInstance = new OvsdbConnectionInstance(key, externalClient, txInvoker,
@@ -149,6 +173,14 @@ public class OvsdbConnectionManager implements OvsdbConnectionListener, AutoClos
             txInvoker.invoke(new OvsdbNodeRemoveCommand(ovsdbConnectionInstance, null, null));
 
             removeConnectionInstance(key);
+
+            //Controller initiated connection can be terminated from switch side.
+            //So cleanup the instance identifier cache.
+            removeInstanceIdentifier(key);
+            stopBridgeConfigReconciliationIfActive(ovsdbConnectionInstance.getInstanceIdentifier());
+            retryConnection(ovsdbConnectionInstance.getInstanceIdentifier(),
+                    ovsdbConnectionInstance.getOvsdbNodeAugmentation(),
+                    ConnectionReconciliationTriggers.ON_DISCONNECT);
         } else {
             LOG.warn("disconnected : Connection instance not found for OVSDB Node {} ", key);
         }
@@ -190,6 +222,8 @@ public class OvsdbConnectionManager implements OvsdbConnectionListener, AutoClos
             client.disconnect();
 
             removeInstanceIdentifier(ovsdbNode.getConnectionInfo());
+
+            stopBridgeConfigReconciliationIfActive(client.getInstanceIdentifier());
         } else {
             LOG.debug("disconnect : connection instance not found for {}",ovsdbNode.getConnectionInfo());
         }
@@ -210,7 +244,7 @@ public class OvsdbConnectionManager implements OvsdbConnectionListener, AutoClos
     }
 */
     @Override
-    public void close() throws Exception {
+    public void close() {
         if (ovsdbDeviceEntityOwnershipListener != null) {
             ovsdbDeviceEntityOwnershipListener.close();
         }
@@ -312,6 +346,30 @@ public class OvsdbConnectionManager implements OvsdbConnectionListener, AutoClos
         return ovsdbConnectionInstance.getHasDeviceOwnership();
     }
 
+    public void reconcileConnection(InstanceIdentifier<Node> iid, OvsdbNodeAugmentation ovsdbNode){
+        this.retryConnection(iid, ovsdbNode,
+                ConnectionReconciliationTriggers.ON_CONTROLLER_INITIATED_CONNECTION_FAILURE);
+
+    }
+
+    public void stopConnectionReconciliationIfActive(InstanceIdentifier<?> iid, OvsdbNodeAugmentation ovsdbNode) {
+        final ReconciliationTask task = new ConnectionReconciliationTask(
+                reconciliationManager,
+                this,
+                iid,
+                ovsdbNode);
+        reconciliationManager.dequeue(task);
+    }
+
+    public void stopBridgeConfigReconciliationIfActive(InstanceIdentifier<?> iid) {
+        final ReconciliationTask task = new BridgeConfigReconciliationTask(
+                reconciliationManager,
+                this,
+                iid,
+                null);
+        reconciliationManager.dequeue(task);
+    }
+
     private void handleOwnershipChanged(EntityOwnershipChange ownershipChange) {
         OvsdbConnectionInstance ovsdbConnectionInstance = getConnectionInstanceFromEntity(ownershipChange.getEntity());
         LOG.debug("handleOwnershipChanged: {} event received for device {}",
@@ -360,6 +418,7 @@ public class OvsdbConnectionManager implements OvsdbConnectionListener, AutoClos
             //so register for monitor callbacks
             ovsdbConnectionInstance.registerCallbacks();
 
+            reconcileBridgeConfigurations(ovsdbConnectionInstance);
         } else {
             //You were owner of the device, but now you are not. With the current ownership
             //grant mechanism, this scenario should not occur. Because this scenario will occur
@@ -380,26 +439,32 @@ public class OvsdbConnectionManager implements OvsdbConnectionListener, AutoClos
         // are chances that other controller instance went down abruptly and it does
         // not clear manager entry, which OvsdbNodeRemoveCommand look for before cleanup.
 
-        InstanceIdentifier<Node> nodeIid = (InstanceIdentifier<Node>) SouthboundUtil
-                .getInstanceIdentifierCodec().bindingDeserializer(entity.getId());
-
-        final ReadWriteTransaction transaction = db.newReadWriteTransaction();
-        Optional<Node> ovsdbNodeOpt = SouthboundUtil.readNode(transaction,nodeIid);
-        if ( ovsdbNodeOpt.isPresent() ) {
-            Node ovsdbNode = ovsdbNodeOpt.get();
-            OvsdbNodeAugmentation nodeAugmentation = ovsdbNode.getAugmentation(OvsdbNodeAugmentation.class);
-            if (nodeAugmentation != null) {
-                if (nodeAugmentation.getManagedNodeEntry() != null) {
-                    for (ManagedNodeEntry managedNode : nodeAugmentation.getManagedNodeEntry()) {
-                        transaction.delete(
-                                LogicalDatastoreType.OPERATIONAL, managedNode.getBridgeRef().getValue());
+        @SuppressWarnings("unchecked") final InstanceIdentifier<Node> nodeIid =
+                (InstanceIdentifier<Node>) SouthboundUtil
+                        .getInstanceIdentifierCodec().bindingDeserializer(entity.getId());
+
+        txInvoker.invoke(new TransactionCommand() {
+            @Override
+            public void execute(ReadWriteTransaction transaction) {
+                Optional<Node> ovsdbNodeOpt = SouthboundUtil.readNode(transaction, nodeIid);
+                if (ovsdbNodeOpt.isPresent()) {
+                    Node ovsdbNode = ovsdbNodeOpt.get();
+                    OvsdbNodeAugmentation nodeAugmentation = ovsdbNode.getAugmentation(OvsdbNodeAugmentation.class);
+                    if (nodeAugmentation != null) {
+                        if (nodeAugmentation.getManagedNodeEntry() != null) {
+                            for (ManagedNodeEntry managedNode : nodeAugmentation.getManagedNodeEntry()) {
+                                transaction.delete(
+                                        LogicalDatastoreType.OPERATIONAL, managedNode.getBridgeRef().getValue());
+                            }
+                        } else {
+                            LOG.debug("{} had no managed nodes", ovsdbNode.getNodeId().getValue());
+                        }
                     }
-                } else {
-                    LOG.debug("{} had no managed nodes", ovsdbNode.getNodeId().getValue());
+                    transaction.delete(LogicalDatastoreType.OPERATIONAL, nodeIid);
                 }
             }
-            SouthboundUtil.deleteNode(transaction, nodeIid);
-        }
+        });
+
     }
 
     private OpenVSwitch getOpenVswitchTableEntry(OvsdbConnectionInstance connectionInstance) {
@@ -496,6 +561,62 @@ public class OvsdbConnectionManager implements OvsdbConnectionListener, AutoClos
         entityConnectionMap.remove(ovsdbConnectionInstance.getConnectedEntity());
     }
 
+    private void retryConnection(final InstanceIdentifier<Node> iid, final OvsdbNodeAugmentation ovsdbNode,
+                                 ConnectionReconciliationTriggers trigger) {
+        final ReconciliationTask task = new ConnectionReconciliationTask(
+                reconciliationManager,
+                this,
+                iid,
+                ovsdbNode);
+
+        if(reconciliationManager.isEnqueued(task)){
+            return;
+        }
+        switch(trigger){
+            case ON_CONTROLLER_INITIATED_CONNECTION_FAILURE:
+                reconciliationManager.enqueueForRetry(task);
+                break;
+            case ON_DISCONNECT:
+            {
+                ReadOnlyTransaction tx = db.newReadOnlyTransaction();
+                CheckedFuture<Optional<Node>, ReadFailedException> readNodeFuture =
+                        tx.read(LogicalDatastoreType.CONFIGURATION, iid);
+
+                final OvsdbConnectionManager connectionManager = this;
+                Futures.addCallback(readNodeFuture, new FutureCallback<Optional<Node>>() {
+                    @Override
+                    public void onSuccess(@Nullable Optional<Node> node) {
+                        if (node.isPresent()) {
+                            LOG.info("Disconnected/Failed connection {} was controller initiated, attempting " +
+                                    "reconnection", ovsdbNode.getConnectionInfo());
+                            reconciliationManager.enqueue(task);
+
+                        } else {
+                            LOG.debug("Connection {} was switch initiated, no reconciliation is required"
+                                    , iid.firstKeyOf(Node.class).getNodeId());
+                        }
+                    }
+
+                    @Override
+                    public void onFailure(Throwable t) {
+                        LOG.warn("Read Config/DS for Node failed! {}", iid, t);
+                    }
+                });
+                break;
+            }
+            default:
+                break;
+        }
+    }
+
+    private void reconcileBridgeConfigurations(final OvsdbConnectionInstance client) {
+        final InstanceIdentifier<Node> nodeIid = client.getInstanceIdentifier();
+        final ReconciliationTask task = new BridgeConfigReconciliationTask(
+                reconciliationManager, OvsdbConnectionManager.this, nodeIid, client);
+
+        reconciliationManager.enqueue(task);
+    }
+
     private class OvsdbDeviceEntityOwnershipListener implements EntityOwnershipListener {
         private OvsdbConnectionManager cm;
         private EntityOwnershipListenerRegistration listenerRegistration;
@@ -512,4 +633,18 @@ public class OvsdbConnectionManager implements OvsdbConnectionListener, AutoClos
             cm.handleOwnershipChanged(ownershipChange);
         }
     }
+
+    private enum ConnectionReconciliationTriggers {
+        /*
+        Reconciliation trigger for scenario where controller's attempt
+        to connect to switch fails on config data store notification
+        */
+        ON_CONTROLLER_INITIATED_CONNECTION_FAILURE,
+
+        /*
+        Reconciliation trigger for the scenario where controller
+        initiated connection disconnects.
+        */
+        ON_DISCONNECT
+    }
 }