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