southbound/southbound-impl/src/main/java/org/opendaylight/ovsdb/southbound/OvsdbConnectionManager.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 package org.opendaylight.ovsdb.southbound;
   9
  10 import static org.opendaylight.ovsdb.lib.operations.Operations.op;
  11
  12 import com.google.common.base.Optional;
  13 import com.google.common.base.Preconditions;
  14 import com.google.common.util.concurrent.CheckedFuture;
  15 import com.google.common.util.concurrent.FutureCallback;
  16 import com.google.common.util.concurrent.Futures;
  17 import java.net.InetAddress;
  18 import java.net.UnknownHostException;
  19 import java.util.ArrayList;
  20 import java.util.List;
  21 import java.util.Map;
  22 import java.util.concurrent.ConcurrentHashMap;
  23 import java.util.concurrent.ExecutionException;
  24 import java.util.concurrent.TimeUnit;
  25 import java.util.concurrent.TimeoutException;
  26 import javax.annotation.Nonnull;
  27 import javax.annotation.Nullable;
  28 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  29 import org.opendaylight.controller.md.sal.binding.api.ReadOnlyTransaction;
  30 import org.opendaylight.controller.md.sal.binding.api.ReadWriteTransaction;
  31 import org.opendaylight.controller.md.sal.common.api.clustering.CandidateAlreadyRegisteredException;
  32 import org.opendaylight.controller.md.sal.common.api.clustering.Entity;
  33 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipCandidateRegistration;
  34 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipChange;
  35 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipListener;
  36 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipListenerRegistration;
  37 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipService;
  38 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipState;
  39 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  40 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
  41 import org.opendaylight.ovsdb.lib.OvsdbClient;
  42 import org.opendaylight.ovsdb.lib.OvsdbConnection;
  43 import org.opendaylight.ovsdb.lib.OvsdbConnectionListener;
  44 import org.opendaylight.ovsdb.lib.operations.Operation;
  45 import org.opendaylight.ovsdb.lib.operations.OperationResult;
  46 import org.opendaylight.ovsdb.lib.operations.Select;
  47 import org.opendaylight.ovsdb.lib.schema.DatabaseSchema;
  48 import org.opendaylight.ovsdb.lib.schema.GenericTableSchema;
  49 import org.opendaylight.ovsdb.lib.schema.typed.TyperUtils;
  50 import org.opendaylight.ovsdb.schema.openvswitch.OpenVSwitch;
  51 import org.opendaylight.ovsdb.southbound.reconciliation.ReconciliationManager;
  52 import org.opendaylight.ovsdb.southbound.reconciliation.ReconciliationTask;
  53 import org.opendaylight.ovsdb.southbound.reconciliation.configuration.BridgeConfigReconciliationTask;
  54 import org.opendaylight.ovsdb.southbound.reconciliation.connection.ConnectionReconciliationTask;
  55 import org.opendaylight.ovsdb.southbound.transactions.md.OvsdbNodeRemoveCommand;
  56 import org.opendaylight.ovsdb.southbound.transactions.md.TransactionCommand;
  57 import org.opendaylight.ovsdb.southbound.transactions.md.TransactionInvoker;
  58 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.ovsdb.rev150105.OvsdbBridgeAttributes;
  59 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.ovsdb.rev150105.OvsdbBridgeAugmentation;
  60 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.ovsdb.rev150105.OvsdbNodeAugmentation;
  61 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.ovsdb.rev150105.ovsdb.node.attributes.ConnectionInfo;
  62 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.ovsdb.rev150105.ovsdb.node.attributes.ManagedNodeEntry;
  63 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
  64 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  65 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  66 import org.slf4j.Logger;
  67 import org.slf4j.LoggerFactory;
  68
  69 public class OvsdbConnectionManager implements OvsdbConnectionListener, AutoCloseable {
  70     private Map<ConnectionInfo, OvsdbConnectionInstance> clients =
  71             new ConcurrentHashMap<>();
  72     private static final Logger LOG = LoggerFactory.getLogger(OvsdbConnectionManager.class);
  73     private static final String ENTITY_TYPE = "ovsdb";
  74
  75     private DataBroker db;
  76     private TransactionInvoker txInvoker;
  77     private Map<ConnectionInfo,InstanceIdentifier<Node>> instanceIdentifiers =
  78             new ConcurrentHashMap<>();
  79     private Map<Entity, OvsdbConnectionInstance> entityConnectionMap =
  80             new ConcurrentHashMap<>();
  81     private EntityOwnershipService entityOwnershipService;
  82     private OvsdbDeviceEntityOwnershipListener ovsdbDeviceEntityOwnershipListener;
  83     private OvsdbConnection ovsdbConnection;
  84     private final ReconciliationManager reconciliationManager;
  85
  86     public OvsdbConnectionManager(DataBroker db,TransactionInvoker txInvoker,
  87                                   EntityOwnershipService entityOwnershipService,
  88                                   OvsdbConnection ovsdbConnection) {
  89         this.db = db;
  90         this.txInvoker = txInvoker;
  91         this.entityOwnershipService = entityOwnershipService;
  92         this.ovsdbDeviceEntityOwnershipListener = new OvsdbDeviceEntityOwnershipListener(this, entityOwnershipService);
  93         this.ovsdbConnection = ovsdbConnection;
  94         this.reconciliationManager = new ReconciliationManager(db);
  95     }
  96
  97     @Override
  98     public void connected(@Nonnull final OvsdbClient externalClient) {
  99         LOG.info("Library connected {} from {}:{} to {}:{}",
 100                 externalClient.getConnectionInfo().getType(),
 101                 externalClient.getConnectionInfo().getRemoteAddress(),
 102                 externalClient.getConnectionInfo().getRemotePort(),
 103                 externalClient.getConnectionInfo().getLocalAddress(),
 104                 externalClient.getConnectionInfo().getLocalPort());
 105         List<String> databases = new ArrayList<>();
 106         try {
 107             databases = externalClient.getDatabases().get(1000, TimeUnit.MILLISECONDS);
 108             if (databases.contains(SouthboundConstants.OPEN_V_SWITCH)) {
 109                 OvsdbConnectionInstance client = connectedButCallBacksNotRegistered(externalClient);
 110                 // Register Cluster Ownership for ConnectionInfo
 111                 registerEntityForOwnership(client);
 112             }
 113         } catch (InterruptedException | ExecutionException | TimeoutException e) {
 114             LOG.warn("Unable to fetch Database list from device {}. Disconnecting from the device.",
 115                     externalClient.getConnectionInfo().getRemoteAddress(), e);
 116             externalClient.disconnect();
 117         }
 118
 119     }
 120
 121     public OvsdbConnectionInstance connectedButCallBacksNotRegistered(final OvsdbClient externalClient) {
 122         LOG.info("OVSDB Connection from {}:{}",externalClient.getConnectionInfo().getRemoteAddress(),
 123                 externalClient.getConnectionInfo().getRemotePort());
 124         ConnectionInfo key = SouthboundMapper.createConnectionInfo(externalClient);
 125         OvsdbConnectionInstance ovsdbConnectionInstance = getConnectionInstance(key);
 126
 127         // Check if existing ovsdbConnectionInstance for the OvsdbClient present.
 128         // In such cases, we will see if the ovsdbConnectionInstance has same externalClient.
 129         if (ovsdbConnectionInstance != null) {
 130             if (ovsdbConnectionInstance.hasOvsdbClient(externalClient)) {
 131                 LOG.warn("OVSDB Connection Instance {} already exists for client {}", key, externalClient);
 132                 return ovsdbConnectionInstance;
 133             }
 134             LOG.warn("OVSDB Connection Instance {} being replaced with client {}", key, externalClient);
 135
 136             // Unregister Cluster Ownership for ConnectionInfo
 137             // Because the ovsdbConnectionInstance is about to be completely replaced!
 138             unregisterEntityForOwnership(ovsdbConnectionInstance);
 139
 140             ovsdbConnectionInstance.disconnect();
 141
 142             removeConnectionInstance(key);
 143
 144             stopBridgeConfigReconciliationIfActive(ovsdbConnectionInstance.getInstanceIdentifier());
 145         }
 146
 147         ovsdbConnectionInstance = new OvsdbConnectionInstance(key, externalClient, txInvoker,
 148                 getInstanceIdentifier(key));
 149         ovsdbConnectionInstance.createTransactInvokers();
 150         return ovsdbConnectionInstance;
 151     }
 152
 153     @Override
 154     public void disconnected(OvsdbClient client) {
 155         LOG.info("Library disconnected {} from {}:{} to {}:{}. Cleaning up the operational data store",
 156                 client.getConnectionInfo().getType(),
 157                 client.getConnectionInfo().getRemoteAddress(),
 158                 client.getConnectionInfo().getRemotePort(),
 159                 client.getConnectionInfo().getLocalAddress(),
 160                 client.getConnectionInfo().getLocalPort());
 161         ConnectionInfo key = SouthboundMapper.createConnectionInfo(client);
 162         OvsdbConnectionInstance ovsdbConnectionInstance = getConnectionInstance(key);
 163         if (ovsdbConnectionInstance != null) {
 164             // Unregister Entity ownership as soon as possible ,so this instance should
 165             // not be used as a candidate in Entity election (given that this instance is
 166             // about to disconnect as well), if current owner get disconnected from
 167             // OVSDB device.
 168             unregisterEntityForOwnership(ovsdbConnectionInstance);
 169
 170             txInvoker.invoke(new OvsdbNodeRemoveCommand(ovsdbConnectionInstance, null, null));
 171
 172             removeConnectionInstance(key);
 173
 174             //Controller initiated connection can be terminated from switch side.
 175             //So cleanup the instance identifier cache.
 176             removeInstanceIdentifier(key);
 177             stopBridgeConfigReconciliationIfActive(ovsdbConnectionInstance.getInstanceIdentifier());
 178             retryConnection(ovsdbConnectionInstance.getInstanceIdentifier(),
 179                     ovsdbConnectionInstance.getOvsdbNodeAugmentation(),
 180                     ConnectionReconciliationTriggers.ON_DISCONNECT);
 181         } else {
 182             LOG.warn("disconnected : Connection instance not found for OVSDB Node {} ", key);
 183         }
 184         LOG.trace("OvsdbConnectionManager: exit disconnected client: {}", client);
 185     }
 186
 187     public OvsdbClient connect(InstanceIdentifier<Node> iid,
 188             OvsdbNodeAugmentation ovsdbNode) throws UnknownHostException {
 189         LOG.info("Connecting to {}", SouthboundUtil.connectionInfoToString(ovsdbNode.getConnectionInfo()));
 190
 191         // TODO handle case where we already have a connection
 192         // TODO use transaction chains to handle ordering issues between disconnected
 193         // TODO and connected when writing to the operational store
 194         InetAddress ip = SouthboundMapper.createInetAddress(ovsdbNode.getConnectionInfo().getRemoteIp());
 195         OvsdbClient client = ovsdbConnection.connect(ip,
 196                 ovsdbNode.getConnectionInfo().getRemotePort().getValue());
 197         // For connections from the controller to the ovs instance, the library doesn't call
 198         // this method for us
 199         if (client != null) {
 200             putInstanceIdentifier(ovsdbNode.getConnectionInfo(), iid.firstIdentifierOf(Node.class));
 201             OvsdbConnectionInstance ovsdbConnectionInstance = connectedButCallBacksNotRegistered(client);
 202             ovsdbConnectionInstance.setOvsdbNodeAugmentation(ovsdbNode);
 203
 204             // Register Cluster Ownership for ConnectionInfo
 205             registerEntityForOwnership(ovsdbConnectionInstance);
 206         } else {
 207             LOG.warn("Failed to connect to OVSDB Node {}", ovsdbNode.getConnectionInfo());
 208         }
 209         return client;
 210     }
 211
 212     public void disconnect(OvsdbNodeAugmentation ovsdbNode) throws UnknownHostException {
 213         LOG.info("Disconnecting from {}", SouthboundUtil.connectionInfoToString(ovsdbNode.getConnectionInfo()));
 214         OvsdbConnectionInstance client = getConnectionInstance(ovsdbNode.getConnectionInfo());
 215         if (client != null) {
 216             // Unregister Cluster Onwership for ConnectionInfo
 217             unregisterEntityForOwnership(client);
 218
 219             client.disconnect();
 220
 221             removeInstanceIdentifier(ovsdbNode.getConnectionInfo());
 222
 223             stopBridgeConfigReconciliationIfActive(client.getInstanceIdentifier());
 224         } else {
 225             LOG.debug("disconnect : connection instance not found for {}",ovsdbNode.getConnectionInfo());
 226         }
 227     }
 228
 229 /*    public void init(ConnectionInfo key) {
 230         OvsdbConnectionInstance client = getConnectionInstance(key);
 231
 232         // TODO (FF): make sure that this cluster instance is the 'entity owner' fo the given OvsdbConnectionInstance ?
 233
 234         if (client != null) {
 235
 236              *  Note: registerCallbacks() is idemPotent... so if you call it repeatedly all is safe,
 237              *  it only registersCallbacks on the *first* call.
 238
 239             client.registerCallbacks();
 240         }
 241     }
 242 */
 243     @Override
 244     public void close() {
 245         if (ovsdbDeviceEntityOwnershipListener != null) {
 246             ovsdbDeviceEntityOwnershipListener.close();
 247         }
 248
 249         for (OvsdbClient client: clients.values()) {
 250             client.disconnect();
 251         }
 252     }
 253
 254     private void putConnectionInstance(ConnectionInfo key,OvsdbConnectionInstance instance) {
 255         ConnectionInfo connectionInfo = SouthboundMapper.suppressLocalIpPort(key);
 256         clients.put(connectionInfo, instance);
 257     }
 258
 259     private void removeConnectionInstance(ConnectionInfo key) {
 260         ConnectionInfo connectionInfo = SouthboundMapper.suppressLocalIpPort(key);
 261         clients.remove(connectionInfo);
 262     }
 263
 264     private void putInstanceIdentifier(ConnectionInfo key,InstanceIdentifier<Node> iid) {
 265         ConnectionInfo connectionInfo = SouthboundMapper.suppressLocalIpPort(key);
 266         instanceIdentifiers.put(connectionInfo, iid);
 267     }
 268
 269     private void removeInstanceIdentifier(ConnectionInfo key) {
 270         ConnectionInfo connectionInfo = SouthboundMapper.suppressLocalIpPort(key);
 271         instanceIdentifiers.remove(connectionInfo);
 272     }
 273
 274     public InstanceIdentifier<Node> getInstanceIdentifier(ConnectionInfo key) {
 275         ConnectionInfo connectionInfo = SouthboundMapper.suppressLocalIpPort(key);
 276         return instanceIdentifiers.get(connectionInfo);
 277     }
 278
 279     public OvsdbConnectionInstance getConnectionInstance(ConnectionInfo key) {
 280         ConnectionInfo connectionInfo = SouthboundMapper.suppressLocalIpPort(key);
 281         return clients.get(connectionInfo);
 282     }
 283
 284     public OvsdbConnectionInstance getConnectionInstance(OvsdbBridgeAttributes mn) {
 285         Optional<OvsdbNodeAugmentation> optional = SouthboundUtil.getManagingNode(db, mn);
 286         if (optional.isPresent()) {
 287             return getConnectionInstance(optional.get().getConnectionInfo());
 288         } else {
 289             return null;
 290         }
 291     }
 292
 293     public OvsdbConnectionInstance getConnectionInstance(Node node) {
 294         Preconditions.checkNotNull(node);
 295         OvsdbNodeAugmentation ovsdbNode = node.getAugmentation(OvsdbNodeAugmentation.class);
 296         OvsdbBridgeAugmentation ovsdbManagedNode = node.getAugmentation(OvsdbBridgeAugmentation.class);
 297         if (ovsdbNode != null) {
 298             return getConnectionInstance(ovsdbNode.getConnectionInfo());
 299         } else if (ovsdbManagedNode != null) {
 300             return getConnectionInstance(ovsdbManagedNode);
 301         } else {
 302             LOG.warn("This is not a node that gives any hint how to find its OVSDB Manager: {}",node);
 303             return null;
 304         }
 305     }
 306
 307     public OvsdbConnectionInstance getConnectionInstance(InstanceIdentifier<Node> nodePath) {
 308         try {
 309             ReadOnlyTransaction transaction = db.newReadOnlyTransaction();
 310             CheckedFuture<Optional<Node>, ReadFailedException> nodeFuture = transaction.read(
 311                     LogicalDatastoreType.OPERATIONAL, nodePath);
 312             transaction.close();
 313             Optional<Node> optional = nodeFuture.get();
 314             if (optional != null && optional.isPresent() && optional.get() != null) {
 315                 return this.getConnectionInstance(optional.get());
 316             } else {
 317                 LOG.warn("Found non-topological node {} on path {}",optional);
 318                 return null;
 319             }
 320         } catch (Exception e) {
 321             LOG.warn("Failed to get Ovsdb Node {}",nodePath, e);
 322             return null;
 323         }
 324     }
 325
 326     public OvsdbClient getClient(ConnectionInfo connectionInfo) {
 327         return getConnectionInstance(connectionInfo);
 328     }
 329
 330     public OvsdbClient getClient(OvsdbBridgeAttributes mn) {
 331         return getConnectionInstance(mn);
 332     }
 333
 334     public OvsdbClient getClient(Node node) {
 335         return getConnectionInstance(node);
 336     }
 337
 338     public Boolean getHasDeviceOwnership(ConnectionInfo connectionInfo) {
 339         OvsdbConnectionInstance ovsdbConnectionInstance = getConnectionInstance(connectionInfo);
 340         if (ovsdbConnectionInstance == null) {
 341             return Boolean.FALSE;
 342         }
 343         return ovsdbConnectionInstance.getHasDeviceOwnership();
 344     }
 345
 346     public void reconcileConnection(InstanceIdentifier<Node> iid, OvsdbNodeAugmentation ovsdbNode) {
 347         this.retryConnection(iid, ovsdbNode,
 348                 ConnectionReconciliationTriggers.ON_CONTROLLER_INITIATED_CONNECTION_FAILURE);
 349
 350     }
 351
 352     public void stopConnectionReconciliationIfActive(InstanceIdentifier<?> iid, OvsdbNodeAugmentation ovsdbNode) {
 353         final ReconciliationTask task = new ConnectionReconciliationTask(
 354                 reconciliationManager,
 355                 this,
 356                 iid,
 357                 ovsdbNode);
 358         reconciliationManager.dequeue(task);
 359     }
 360
 361     public void stopBridgeConfigReconciliationIfActive(InstanceIdentifier<?> iid) {
 362         final ReconciliationTask task = new BridgeConfigReconciliationTask(
 363                 reconciliationManager,
 364                 this,
 365                 iid,
 366                 null);
 367         reconciliationManager.dequeue(task);
 368     }
 369
 370     private void handleOwnershipChanged(EntityOwnershipChange ownershipChange) {
 371         OvsdbConnectionInstance ovsdbConnectionInstance = getConnectionInstanceFromEntity(ownershipChange.getEntity());
 372         LOG.debug("handleOwnershipChanged: {} event received for device {}",
 373                 ownershipChange, ovsdbConnectionInstance != null ? ovsdbConnectionInstance.getConnectionInfo()
 374                         : "that's currently NOT registered by *this* southbound plugin instance");
 375
 376         if (ovsdbConnectionInstance == null) {
 377             if (ownershipChange.isOwner()) {
 378                 LOG.warn("handleOwnershipChanged: *this* instance is elected as an owner of the device {} but it "
 379                         + "is NOT registered for ownership", ownershipChange.getEntity());
 380             } else {
 381                 // EntityOwnershipService sends notification to all the nodes, irrespective of whether
 382                 // that instance registered for the device ownership or not. It is to make sure that
 383                 // If all the controller instance that was connected to the device are down, so the
 384                 // running instance can clear up the operational data store even though it was not
 385                 // connected to the device.
 386                 LOG.debug("handleOwnershipChanged: No connection instance found for {}", ownershipChange.getEntity());
 387             }
 388
 389             // If entity has no owner, clean up the operational data store (it's possible because owner controller
 390             // might went down abruptly and didn't get a chance to clean up the operational data store.
 391             if (!ownershipChange.hasOwner()) {
 392                 LOG.info("{} has no owner, cleaning up the operational data store", ownershipChange.getEntity());
 393                 cleanEntityOperationalData(ownershipChange.getEntity());
 394             }
 395             return;
 396         }
 397         //Connection detail need to be cached, irrespective of ownership result.
 398         putConnectionInstance(ovsdbConnectionInstance.getMDConnectionInfo(),ovsdbConnectionInstance);
 399
 400         if (ownershipChange.isOwner() == ovsdbConnectionInstance.getHasDeviceOwnership()) {
 401             LOG.info("handleOwnershipChanged: no change in ownership for {}. Ownership status is : {}",
 402                     ovsdbConnectionInstance.getConnectionInfo(), ovsdbConnectionInstance.getHasDeviceOwnership()
 403                             ? SouthboundConstants.OwnershipStates.OWNER.getState()
 404                             : SouthboundConstants.OwnershipStates.NONOWNER.getState());
 405             return;
 406         }
 407
 408         ovsdbConnectionInstance.setHasDeviceOwnership(ownershipChange.isOwner());
 409         // You were not an owner, but now you are
 410         if (ownershipChange.isOwner()) {
 411             LOG.info("handleOwnershipChanged: *this* southbound plugin instance is an OWNER of the device {}",
 412                     ovsdbConnectionInstance.getConnectionInfo());
 413
 414             //*this* instance of southbound plugin is owner of the device,
 415             //so register for monitor callbacks
 416             ovsdbConnectionInstance.registerCallbacks();
 417
 418             reconcileBridgeConfigurations(ovsdbConnectionInstance);
 419         } else {
 420             //You were owner of the device, but now you are not. With the current ownership
 421             //grant mechanism, this scenario should not occur. Because this scenario will occur
 422             //when this controller went down or switch flap the connection, but in both the case
 423             //it will go through the re-registration process. We need to implement this condition
 424             //when clustering service implement a ownership grant strategy which can revoke the
 425             //device ownership for load balancing the devices across the instances.
 426             //Once this condition occur, we should unregister the callback.
 427             LOG.error("handleOwnershipChanged: *this* southbound plugin instance is no longer the owner of device {}."
 428                     + "This should NOT happen.",
 429                     ovsdbConnectionInstance.getNodeId().getValue());
 430         }
 431     }
 432
 433     private void cleanEntityOperationalData(Entity entity) {
 434
 435         //Do explicit cleanup rather than using OvsdbNodeRemoveCommand, because there
 436         // are chances that other controller instance went down abruptly and it does
 437         // not clear manager entry, which OvsdbNodeRemoveCommand look for before cleanup.
 438
 439         @SuppressWarnings("unchecked") final InstanceIdentifier<Node> nodeIid =
 440                 (InstanceIdentifier<Node>) SouthboundUtil
 441                         .getInstanceIdentifierCodec().bindingDeserializer(entity.getId());
 442
 443         txInvoker.invoke(new TransactionCommand() {
 444             @Override
 445             public void execute(ReadWriteTransaction transaction) {
 446                 Optional<Node> ovsdbNodeOpt = SouthboundUtil.readNode(transaction, nodeIid);
 447                 if (ovsdbNodeOpt.isPresent()) {
 448                     Node ovsdbNode = ovsdbNodeOpt.get();
 449                     OvsdbNodeAugmentation nodeAugmentation = ovsdbNode.getAugmentation(OvsdbNodeAugmentation.class);
 450                     if (nodeAugmentation != null) {
 451                         if (nodeAugmentation.getManagedNodeEntry() != null) {
 452                             for (ManagedNodeEntry managedNode : nodeAugmentation.getManagedNodeEntry()) {
 453                                 transaction.delete(
 454                                         LogicalDatastoreType.OPERATIONAL, managedNode.getBridgeRef().getValue());
 455                             }
 456                         } else {
 457                             LOG.debug("{} had no managed nodes", ovsdbNode.getNodeId().getValue());
 458                         }
 459                     }
 460                     transaction.delete(LogicalDatastoreType.OPERATIONAL, nodeIid);
 461                 }
 462             }
 463         });
 464
 465     }
 466
 467     private OpenVSwitch getOpenVswitchTableEntry(OvsdbConnectionInstance connectionInstance) {
 468         DatabaseSchema dbSchema = null;
 469         OpenVSwitch openVSwitchRow = null;
 470         try {
 471             dbSchema = connectionInstance.getSchema(OvsdbSchemaContants.databaseName).get();
 472         } catch (InterruptedException | ExecutionException e) {
 473             LOG.warn("Not able to fetch schema for database {} from device {}",
 474                     OvsdbSchemaContants.databaseName,connectionInstance.getConnectionInfo(),e);
 475         }
 476         if (dbSchema != null) {
 477             GenericTableSchema openVSwitchSchema = TyperUtils.getTableSchema(dbSchema, OpenVSwitch.class);
 478
 479             List<String> openVSwitchTableColumn = new ArrayList<>();
 480             openVSwitchTableColumn.addAll(openVSwitchSchema.getColumns());
 481             Select<GenericTableSchema> selectOperation = op.select(openVSwitchSchema);
 482             selectOperation.setColumns(openVSwitchTableColumn);
 483
 484             List<Operation> operations = new ArrayList<>();
 485             operations.add(selectOperation);
 486             operations.add(op.comment("Fetching Open_VSwitch table rows"));
 487             try {
 488                 List<OperationResult> results = connectionInstance.transact(dbSchema, operations).get();
 489                 if (results != null ) {
 490                     OperationResult selectResult = results.get(0);
 491                     openVSwitchRow = TyperUtils.getTypedRowWrapper(
 492                             dbSchema,OpenVSwitch.class,selectResult.getRows().get(0));
 493
 494                 }
 495             } catch (InterruptedException | ExecutionException e) {
 496                 LOG.warn("Not able to fetch OpenVswitch table row from device {}",
 497                         connectionInstance.getConnectionInfo(),e);
 498             }
 499         }
 500         return openVSwitchRow;
 501     }
 502
 503     private Entity getEntityFromConnectionInstance(@Nonnull OvsdbConnectionInstance ovsdbConnectionInstance) {
 504         InstanceIdentifier<Node> iid = ovsdbConnectionInstance.getInstanceIdentifier();
 505         if ( iid == null ) {
 506             /* Switch initiated connection won't have iid, till it gets OpenVSwitch
 507              * table update but update callback is always registered after ownership
 508              * is granted. So we are explicitly fetch the row here to get the iid.
 509              */
 510             OpenVSwitch openvswitchRow = getOpenVswitchTableEntry(ovsdbConnectionInstance);
 511             iid = SouthboundMapper.getInstanceIdentifier(openvswitchRow);
 512             LOG.info("InstanceIdentifier {} generated for device "
 513                     + "connection {}",iid,ovsdbConnectionInstance.getConnectionInfo());
 514             ovsdbConnectionInstance.setInstanceIdentifier(iid);
 515         }
 516         YangInstanceIdentifier entityId = SouthboundUtil.getInstanceIdentifierCodec().getYangInstanceIdentifier(iid);
 517         Entity deviceEntity = new Entity(ENTITY_TYPE, entityId);
 518         LOG.debug("Entity {} created for device connection {}",
 519                 deviceEntity, ovsdbConnectionInstance.getConnectionInfo());
 520         return deviceEntity;
 521     }
 522
 523     private OvsdbConnectionInstance getConnectionInstanceFromEntity(Entity entity) {
 524         return entityConnectionMap.get(entity);
 525     }
 526
 527     private void registerEntityForOwnership(OvsdbConnectionInstance ovsdbConnectionInstance) {
 528
 529         Entity candidateEntity = getEntityFromConnectionInstance(ovsdbConnectionInstance);
 530         entityConnectionMap.put(candidateEntity, ovsdbConnectionInstance);
 531         ovsdbConnectionInstance.setConnectedEntity(candidateEntity);
 532         try {
 533             EntityOwnershipCandidateRegistration registration =
 534                     entityOwnershipService.registerCandidate(candidateEntity);
 535             ovsdbConnectionInstance.setDeviceOwnershipCandidateRegistration(registration);
 536             LOG.info("OVSDB entity {} is registered for ownership.", candidateEntity);
 537
 538             //If entity already has owner, it won't get notification from EntityOwnershipService
 539             //so cache the connection instances.
 540             Optional<EntityOwnershipState> ownershipStateOpt =
 541                     entityOwnershipService.getOwnershipState(candidateEntity);
 542             if (ownershipStateOpt.isPresent()) {
 543                 EntityOwnershipState ownershipState = ownershipStateOpt.get();
 544                 if (ownershipState.hasOwner() && !ownershipState.isOwner()) {
 545                     LOG.info("OVSDB entity {} is already owned by other southbound plugin "
 546                                     + "instance, so *this* instance is NOT an OWNER of the device",
 547                             ovsdbConnectionInstance.getConnectionInfo());
 548                     putConnectionInstance(ovsdbConnectionInstance.getMDConnectionInfo(),ovsdbConnectionInstance);
 549                 }
 550             }
 551         } catch (CandidateAlreadyRegisteredException e) {
 552             LOG.warn("OVSDB entity {} was already registered for ownership", candidateEntity, e);
 553         }
 554
 555     }
 556
 557     private void unregisterEntityForOwnership(OvsdbConnectionInstance ovsdbConnectionInstance) {
 558         ovsdbConnectionInstance.closeDeviceOwnershipCandidateRegistration();
 559         entityConnectionMap.remove(ovsdbConnectionInstance.getConnectedEntity());
 560     }
 561
 562     private void retryConnection(final InstanceIdentifier<Node> iid, final OvsdbNodeAugmentation ovsdbNode,
 563                                  ConnectionReconciliationTriggers trigger) {
 564         final ReconciliationTask task = new ConnectionReconciliationTask(
 565                 reconciliationManager,
 566                 this,
 567                 iid,
 568                 ovsdbNode);
 569
 570         if (reconciliationManager.isEnqueued(task)) {
 571             return;
 572         }
 573         switch (trigger) {
 574             case ON_CONTROLLER_INITIATED_CONNECTION_FAILURE:
 575                 reconciliationManager.enqueueForRetry(task);
 576                 break;
 577             case ON_DISCONNECT:
 578             {
 579                 ReadOnlyTransaction tx = db.newReadOnlyTransaction();
 580                 CheckedFuture<Optional<Node>, ReadFailedException> readNodeFuture =
 581                         tx.read(LogicalDatastoreType.CONFIGURATION, iid);
 582
 583                 final OvsdbConnectionManager connectionManager = this;
 584                 Futures.addCallback(readNodeFuture, new FutureCallback<Optional<Node>>() {
 585                     @Override
 586                     public void onSuccess(@Nullable Optional<Node> node) {
 587                         if (node.isPresent()) {
 588                             LOG.info("Disconnected/Failed connection {} was controller initiated, attempting "
 589                                     + "reconnection", ovsdbNode.getConnectionInfo());
 590                             reconciliationManager.enqueue(task);
 591
 592                         } else {
 593                             LOG.debug("Connection {} was switch initiated, no reconciliation is required",
 594                                     iid.firstKeyOf(Node.class).getNodeId());
 595                         }
 596                     }
 597
 598                     @Override
 599                     public void onFailure(Throwable throwable) {
 600                         LOG.warn("Read Config/DS for Node failed! {}", iid, throwable);
 601                     }
 602                 });
 603                 break;
 604             }
 605             default:
 606                 break;
 607         }
 608     }
 609
 610     private void reconcileBridgeConfigurations(final OvsdbConnectionInstance client) {
 611         final InstanceIdentifier<Node> nodeIid = client.getInstanceIdentifier();
 612         final ReconciliationTask task = new BridgeConfigReconciliationTask(
 613                 reconciliationManager, OvsdbConnectionManager.this, nodeIid, client);
 614
 615         reconciliationManager.enqueue(task);
 616     }
 617
 618     private class OvsdbDeviceEntityOwnershipListener implements EntityOwnershipListener {
 619         private OvsdbConnectionManager cm;
 620         private EntityOwnershipListenerRegistration listenerRegistration;
 621
 622         OvsdbDeviceEntityOwnershipListener(OvsdbConnectionManager cm, EntityOwnershipService entityOwnershipService) {
 623             this.cm = cm;
 624             listenerRegistration = entityOwnershipService.registerListener(ENTITY_TYPE, this);
 625         }
 626
 627         public void close() {
 628             listenerRegistration.close();
 629         }
 630
 631         @Override
 632         public void ownershipChanged(EntityOwnershipChange ownershipChange) {
 633             cm.handleOwnershipChanged(ownershipChange);
 634         }
 635     }
 636
 637     private enum ConnectionReconciliationTriggers {
 638         /*
 639         Reconciliation trigger for scenario where controller's attempt
 640         to connect to switch fails on config data store notification
 641         */
 642         ON_CONTROLLER_INITIATED_CONNECTION_FAILURE,
 643
 644         /*
 645         Reconciliation trigger for the scenario where controller
 646         initiated connection disconnects.
 647         */
 648         ON_DISCONNECT
 649     }
 650 }