southbound/southbound-impl/src/main/java/org/opendaylight/ovsdb/southbound/OvsdbDataTreeChangeListener.java

   1 /*
   2  * Copyright © 2016 Red Hat, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 package org.opendaylight.ovsdb.southbound;
  10
  11 import java.net.UnknownHostException;
  12 import java.util.Collection;
  13 import java.util.HashMap;
  14 import java.util.List;
  15 import java.util.Map;
  16 import java.util.Map.Entry;
  17
  18 import javax.annotation.Nonnull;
  19
  20 import org.opendaylight.controller.md.sal.binding.api.ClusteredDataTreeChangeListener;
  21 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  22 import org.opendaylight.controller.md.sal.binding.api.DataObjectModification;
  23 import org.opendaylight.controller.md.sal.binding.api.DataTreeChangeListener;
  24 import org.opendaylight.controller.md.sal.binding.api.DataTreeIdentifier;
  25 import org.opendaylight.controller.md.sal.binding.api.DataTreeModification;
  26 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  27 import org.opendaylight.ovsdb.lib.OvsdbClient;
  28 import org.opendaylight.ovsdb.southbound.ovsdb.transact.BridgeOperationalState;
  29 import org.opendaylight.ovsdb.southbound.ovsdb.transact.TransactCommandAggregator;
  30 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.ovsdb.rev150105.OvsdbBridgeAugmentation;
  31 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.ovsdb.rev150105.OvsdbNodeAugmentation;
  32 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.ovsdb.rev150105.ovsdb.node.attributes.ConnectionInfo;
  33 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NetworkTopology;
  34 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.Topology;
  35 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyKey;
  36 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
  37 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.node.TerminationPoint;
  38 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  39 import org.opendaylight.yangtools.yang.binding.Augmentation;
  40 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  41 import org.slf4j.Logger;
  42 import org.slf4j.LoggerFactory;
  43
  44 /**
  45  * Data-tree change listener for OVSDB.
  46  */
  47 public class OvsdbDataTreeChangeListener implements ClusteredDataTreeChangeListener<Node>, AutoCloseable {
  48
  49     /** Our registration. */
  50     private final ListenerRegistration<DataTreeChangeListener<Node>> registration;
  51
  52     /** The connection manager. */
  53     private final OvsdbConnectionManager cm;
  54
  55     /** The data broker. */
  56     private final DataBroker db;
  57
  58     /** Logger. */
  59     private static final Logger LOG = LoggerFactory.getLogger(OvsdbDataTreeChangeListener.class);
  60
  61     /**
  62      * Create an instance and register the listener.
  63      *
  64      * @param db The data broker.
  65      * @param cm The connection manager.
  66      */
  67     OvsdbDataTreeChangeListener(DataBroker db, OvsdbConnectionManager cm) {
  68         LOG.info("Registering OvsdbNodeDataChangeListener");
  69         this.cm = cm;
  70         this.db = db;
  71         InstanceIdentifier<Node> path = InstanceIdentifier
  72                 .create(NetworkTopology.class)
  73                 .child(Topology.class, new TopologyKey(SouthboundConstants.OVSDB_TOPOLOGY_ID))
  74                 .child(Node.class);
  75         DataTreeIdentifier<Node> dataTreeIdentifier =
  76                 new DataTreeIdentifier<>(LogicalDatastoreType.CONFIGURATION, path);
  77         registration = db.registerDataTreeChangeListener(dataTreeIdentifier, this);
  78     }
  79
  80     @Override
  81     public void close() {
  82         registration.close();
  83     }
  84
  85     @Override
  86     public void onDataTreeChanged(@Nonnull Collection<DataTreeModification<Node>> changes) {
  87         LOG.trace("onDataTreeChanged: {}", changes);
  88
  89         // Connect first if necessary
  90         connect(changes);
  91
  92         // Update connections if necessary
  93         updateConnections(changes);
  94
  95         // Update the actual data
  96         updateData(changes);
  97
  98         // Disconnect if necessary
  99         disconnect(changes);
 100
 101         LOG.trace("onDataTreeChanged: exit");
 102     }
 103
 104     private void connect(@Nonnull Collection<DataTreeModification<Node>> changes) {
 105         for (DataTreeModification<Node> change : changes) {
 106             if (change.getRootNode().getModificationType() == DataObjectModification.ModificationType.WRITE || change
 107                     .getRootNode().getModificationType() == DataObjectModification.ModificationType.SUBTREE_MODIFIED) {
 108                 DataObjectModification<OvsdbNodeAugmentation> ovsdbNodeModification =
 109                         change.getRootNode().getModifiedAugmentation(OvsdbNodeAugmentation.class);
 110                 if (ovsdbNodeModification != null && ovsdbNodeModification.getDataBefore() == null
 111                         && ovsdbNodeModification.getDataAfter() != null
 112                         && ovsdbNodeModification.getDataAfter().getConnectionInfo() != null) {
 113                     OvsdbNodeAugmentation ovsdbNode = ovsdbNodeModification.getDataAfter();
 114                     ConnectionInfo key = ovsdbNode.getConnectionInfo();
 115                     InstanceIdentifier<Node> iid = cm.getInstanceIdentifier(key);
 116                     if ( iid != null) {
 117                         LOG.warn("Connection to device {} already exists. Plugin does not allow multiple connections "
 118                                 + "to same device, hence dropping the request {}", key, ovsdbNode);
 119                     } else {
 120                         try {
 121                             InstanceIdentifier<Node> instanceIdentifier = change.getRootPath().getRootIdentifier();
 122                             LOG.info("Connecting on key {} to {}", instanceIdentifier, ovsdbNode);
 123                             cm.connect(instanceIdentifier, ovsdbNode);
 124                         } catch (UnknownHostException e) {
 125                             LOG.warn("Failed to connect to ovsdbNode", e);
 126                         }
 127                     }
 128                 }
 129             }
 130         }
 131     }
 132
 133     private void disconnect(@Nonnull Collection<DataTreeModification<Node>> changes) {
 134         for (DataTreeModification<Node> change : changes) {
 135             if (change.getRootNode().getModificationType() == DataObjectModification.ModificationType.DELETE) {
 136                 DataObjectModification<OvsdbNodeAugmentation> ovsdbNodeModification =
 137                         change.getRootNode().getModifiedAugmentation(OvsdbNodeAugmentation.class);
 138                 if (ovsdbNodeModification != null && ovsdbNodeModification.getDataBefore() != null) {
 139                     OvsdbNodeAugmentation ovsdbNode = ovsdbNodeModification.getDataBefore();
 140                     ConnectionInfo key = ovsdbNode.getConnectionInfo();
 141                     InstanceIdentifier<Node> iid = cm.getInstanceIdentifier(key);
 142                     try {
 143                         LOG.info("Disconnecting from {}", ovsdbNode);
 144                         cm.disconnect(ovsdbNode);
 145                         cm.stopConnectionReconciliationIfActive(iid.firstIdentifierOf(Node.class), ovsdbNode);
 146                     } catch (UnknownHostException e) {
 147                         LOG.warn("Failed to disconnect ovsdbNode", e);
 148                     }
 149                 }
 150             }
 151         }
 152     }
 153
 154     private void updateConnections(@Nonnull Collection<DataTreeModification<Node>> changes) {
 155         for (DataTreeModification<Node> change : changes) {
 156             if (change.getRootNode().getModificationType() == DataObjectModification.ModificationType.WRITE || change
 157                     .getRootNode().getModificationType() == DataObjectModification.ModificationType.SUBTREE_MODIFIED) {
 158                 DataObjectModification<OvsdbNodeAugmentation> ovsdbNodeModification =
 159                         change.getRootNode().getModifiedAugmentation(OvsdbNodeAugmentation.class);
 160                 if (ovsdbNodeModification != null && ovsdbNodeModification.getDataBefore() != null
 161                         && ovsdbNodeModification.getDataAfter() != null
 162                         && ovsdbNodeModification.getDataAfter().getConnectionInfo() != null) {
 163                     OvsdbClient client = cm.getClient(ovsdbNodeModification.getDataAfter().getConnectionInfo());
 164                     if (client == null) {
 165                         if (ovsdbNodeModification.getDataBefore() != null) {
 166                             try {
 167                                 cm.disconnect(ovsdbNodeModification.getDataBefore());
 168                                 cm.connect(change.getRootPath().getRootIdentifier(), ovsdbNodeModification
 169                                         .getDataAfter());
 170                             } catch (UnknownHostException e) {
 171                                 LOG.warn("Error disconnecting from or connecting to ovsdbNode", e);
 172                             }
 173                         }
 174                     }
 175                 }
 176             }
 177         }
 178     }
 179
 180     private void updateData(@Nonnull Collection<DataTreeModification<Node>> changes) {
 181         for (Entry<InstanceIdentifier<Node>, OvsdbConnectionInstance> connectionInstanceEntry :
 182                 connectionInstancesFromChanges(changes).entrySet()) {
 183             OvsdbConnectionInstance connectionInstance = connectionInstanceEntry.getValue();
 184             connectionInstance.transact(new TransactCommandAggregator(),
 185                     new BridgeOperationalState(db, changes), changes);
 186         }
 187     }
 188
 189     private Map<InstanceIdentifier<Node>, OvsdbConnectionInstance> connectionInstancesFromChanges(
 190             @Nonnull Collection<DataTreeModification<Node>> changes) {
 191         Map<InstanceIdentifier<Node>,OvsdbConnectionInstance> result =
 192                 new HashMap<>();
 193         for (DataTreeModification<Node> change : changes) {
 194             OvsdbConnectionInstance client = null;
 195             Node node = change.getRootNode().getDataAfter() != null?
 196                     change.getRootNode().getDataAfter() : change.getRootNode().getDataBefore();
 197             if (node != null) {
 198                 InstanceIdentifier<Node> nodeIid;
 199                 Augmentation nodeAug = node.getAugmentation(OvsdbNodeAugmentation.class) !=null?
 200                         node.getAugmentation(OvsdbNodeAugmentation.class):node.getAugmentation(OvsdbBridgeAugmentation.class);
 201
 202                 if(nodeAug instanceof OvsdbNodeAugmentation) {
 203                     OvsdbNodeAugmentation ovsdbNode = (OvsdbNodeAugmentation) nodeAug;
 204                     if(ovsdbNode.getConnectionInfo() != null) {
 205                         client = cm.getConnectionInstance(ovsdbNode.getConnectionInfo());
 206                     }else {
 207                         client = cm.getConnectionInstance(SouthboundMapper.createInstanceIdentifier(node.getNodeId()));
 208                     }
 209                 }
 210                 if(nodeAug instanceof OvsdbBridgeAugmentation) {
 211                     OvsdbBridgeAugmentation bridgeAugmentation = (OvsdbBridgeAugmentation)nodeAug;
 212                     if(bridgeAugmentation.getManagedBy() != null) {
 213                         nodeIid = (InstanceIdentifier<Node>) bridgeAugmentation.getManagedBy().getValue();
 214                         client = cm.getConnectionInstance(nodeIid);
 215                     }
 216                 }
 217             } else {
 218                 LOG.warn("Following change don't have after/before data {}", change);
 219             }
 220             if (client != null) {
 221                 LOG.debug("Found client for {}", node);
 222                     /*
 223                      * As of now data change sets are processed by single thread, so we can assume that device will
 224                      * be connected and ownership will be decided before sending any instructions down to the device.
 225                      * Note:Processing order in onDataChange() method should not change. If processing is changed to
 226                      * use multiple thread, we might need to take care of corner cases, where ownership is not decided
 227                      * but transaction are ready to go to switch. In that scenario, either we need to queue those task
 228                      * till ownership is decided for that specific device.
 229                      * Given that each DataChangeNotification is notified through separate thread, so we are already
 230                      * multi threaded and i don't see any need to further parallelism per DataChange
 231                      * notifications processing.
 232                      */
 233                 if ( cm.getHasDeviceOwnership(client.getMDConnectionInfo())) {
 234                     LOG.debug("*This* instance of southbound plugin is an owner of the device {}", node);
 235                     result.put(change.getRootPath().getRootIdentifier(), client);
 236                 } else {
 237                     LOG.debug("*This* instance of southbound plugin is *not* an owner of the device {}", node);
 238                 }
 239             } else {
 240                 LOG.debug("Did not find client for {}", node);
 241             }
 242         }
 243         return result;
 244     }
 245 }