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.InstanceIdentifier;
  40 import org.slf4j.Logger;
  41 import org.slf4j.LoggerFactory;
  42
  43 /**
  44  * Data-tree change listener for OVSDB.
  45  */
  46 public class OvsdbDataTreeChangeListener implements ClusteredDataTreeChangeListener<Node>, AutoCloseable {
  47
  48     /** Our registration. */
  49     private final ListenerRegistration<DataTreeChangeListener<Node>> registration;
  50
  51     /** The connection manager. */
  52     private final OvsdbConnectionManager cm;
  53
  54     /** The data broker. */
  55     private final DataBroker db;
  56
  57     /** Logger. */
  58     private static final Logger LOG = LoggerFactory.getLogger(OvsdbDataTreeChangeListener.class);
  59
  60     /**
  61      * Create an instance and register the listener.
  62      *
  63      * @param db The data broker.
  64      * @param cm The connection manager.
  65      */
  66     OvsdbDataTreeChangeListener(DataBroker db, OvsdbConnectionManager cm) {
  67         LOG.info("Registering OvsdbNodeDataChangeListener");
  68         this.cm = cm;
  69         this.db = db;
  70         InstanceIdentifier<Node> path = InstanceIdentifier
  71                 .create(NetworkTopology.class)
  72                 .child(Topology.class, new TopologyKey(SouthboundConstants.OVSDB_TOPOLOGY_ID))
  73                 .child(Node.class);
  74         DataTreeIdentifier<Node> dataTreeIdentifier =
  75                 new DataTreeIdentifier<>(LogicalDatastoreType.CONFIGURATION, path);
  76         registration = db.registerDataTreeChangeListener(dataTreeIdentifier, this);
  77     }
  78
  79     @Override
  80     public void close() {
  81         registration.close();
  82     }
  83
  84     @Override
  85     public void onDataTreeChanged(@Nonnull Collection<DataTreeModification<Node>> changes) {
  86         LOG.trace("onDataTreeChanged: {}", changes);
  87
  88         // Connect first if necessary
  89         connect(changes);
  90
  91         // Update connections if necessary
  92         updateConnections(changes);
  93
  94         // Update the actual data
  95         updateData(changes);
  96
  97         // Disconnect if necessary
  98         disconnect(changes);
  99
 100         LOG.trace("onDataTreeChanged: exit");
 101     }
 102
 103     private void connect(@Nonnull Collection<DataTreeModification<Node>> changes) {
 104         for (DataTreeModification<Node> change : changes) {
 105             if (change.getRootNode().getModificationType() == DataObjectModification.ModificationType.WRITE || change
 106                     .getRootNode().getModificationType() == DataObjectModification.ModificationType.SUBTREE_MODIFIED) {
 107                 DataObjectModification<OvsdbNodeAugmentation> ovsdbNodeModification =
 108                         change.getRootNode().getModifiedAugmentation(OvsdbNodeAugmentation.class);
 109                 if (ovsdbNodeModification != null && ovsdbNodeModification.getDataBefore() == null
 110                         && ovsdbNodeModification.getDataAfter() != null) {
 111                     OvsdbNodeAugmentation ovsdbNode = ovsdbNodeModification.getDataAfter();
 112                     ConnectionInfo key = ovsdbNode.getConnectionInfo();
 113                     InstanceIdentifier<Node> iid = cm.getInstanceIdentifier(key);
 114                     if ( iid != null) {
 115                         LOG.warn("Connection to device {} already exists. Plugin does not allow multiple connections "
 116                                 + "to same device, hence dropping the request {}", key, ovsdbNode);
 117                     } else {
 118                         try {
 119                             InstanceIdentifier<Node> instanceIdentifier = change.getRootPath().getRootIdentifier();
 120                             LOG.info("Connecting on key {} to {}", instanceIdentifier, ovsdbNode);
 121                             cm.connect(instanceIdentifier, ovsdbNode);
 122                         } catch (UnknownHostException e) {
 123                             LOG.warn("Failed to connect to ovsdbNode", e);
 124                         }
 125                     }
 126                 }
 127             }
 128         }
 129     }
 130
 131     private void disconnect(@Nonnull Collection<DataTreeModification<Node>> changes) {
 132         for (DataTreeModification<Node> change : changes) {
 133             if (change.getRootNode().getModificationType() == DataObjectModification.ModificationType.DELETE) {
 134                 DataObjectModification<OvsdbNodeAugmentation> ovsdbNodeModification =
 135                         change.getRootNode().getModifiedAugmentation(OvsdbNodeAugmentation.class);
 136                 if (ovsdbNodeModification != null && ovsdbNodeModification.getDataBefore() != null) {
 137                     OvsdbNodeAugmentation ovsdbNode = ovsdbNodeModification.getDataBefore();
 138                     ConnectionInfo key = ovsdbNode.getConnectionInfo();
 139                     InstanceIdentifier<Node> iid = cm.getInstanceIdentifier(key);
 140                     try {
 141                         LOG.info("Disconnecting from {}", ovsdbNode);
 142                         cm.disconnect(ovsdbNode);
 143                         cm.stopConnectionReconciliationIfActive(iid.firstIdentifierOf(Node.class), ovsdbNode);
 144                     } catch (UnknownHostException e) {
 145                         LOG.warn("Failed to disconnect ovsdbNode", e);
 146                     }
 147                 }
 148             }
 149         }
 150     }
 151
 152     private void updateConnections(@Nonnull Collection<DataTreeModification<Node>> changes) {
 153         for (DataTreeModification<Node> change : changes) {
 154             if (change.getRootNode().getModificationType() == DataObjectModification.ModificationType.WRITE || change
 155                     .getRootNode().getModificationType() == DataObjectModification.ModificationType.SUBTREE_MODIFIED) {
 156                 DataObjectModification<OvsdbNodeAugmentation> ovsdbNodeModification =
 157                         change.getRootNode().getModifiedAugmentation(OvsdbNodeAugmentation.class);
 158                 if (ovsdbNodeModification != null && ovsdbNodeModification.getDataBefore() != null
 159                         && ovsdbNodeModification.getDataAfter() != 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             DataObjectModification<OvsdbBridgeAugmentation> bridgeModification =
 192                     change.getRootNode().getModifiedAugmentation(OvsdbBridgeAugmentation.class);
 193             OvsdbConnectionInstance client = null;
 194             Node node = change.getRootNode().getDataAfter();
 195             if (bridgeModification != null && bridgeModification.getDataAfter() != null) {
 196                 client = cm.getConnectionInstance(bridgeModification.getDataAfter());
 197             } else {
 198                 DataObjectModification<OvsdbNodeAugmentation> nodeModification =
 199                         change.getRootNode().getModifiedAugmentation(OvsdbNodeAugmentation.class);
 200                 if (nodeModification != null && nodeModification.getDataAfter() != null && nodeModification
 201                         .getDataAfter().getConnectionInfo() != null) {
 202                     client = cm.getConnectionInstance(nodeModification.getDataAfter().getConnectionInfo());
 203                 } else {
 204                     if (node != null) {
 205                         List<TerminationPoint> terminationPoints = node.getTerminationPoint();
 206                         if (terminationPoints != null && !terminationPoints.isEmpty()) {
 207                             InstanceIdentifier<Node> nodeIid = SouthboundMapper.createInstanceIdentifier(
 208                                     node.getNodeId());
 209                             client = cm.getConnectionInstance(nodeIid);
 210                         }
 211                     }
 212                 }
 213             }
 214             if (client != null) {
 215                 LOG.debug("Found client for {}", node);
 216                     /*
 217                      * As of now data change sets are processed by single thread, so we can assume that device will
 218                      * be connected and ownership will be decided before sending any instructions down to the device.
 219                      * Note:Processing order in onDataChange() method should not change. If processing is changed to
 220                      * use multiple thread, we might need to take care of corner cases, where ownership is not decided
 221                      * but transaction are ready to go to switch. In that scenario, either we need to queue those task
 222                      * till ownership is decided for that specific device.
 223                      * Given that each DataChangeNotification is notified through separate thread, so we are already
 224                      * multi threaded and i don't see any need to further parallelism per DataChange
 225                      * notifications processing.
 226                      */
 227                 if ( cm.getHasDeviceOwnership(client.getMDConnectionInfo())) {
 228                     LOG.debug("*This* instance of southbound plugin is an owner of the device {}", node);
 229                     result.put(change.getRootPath().getRootIdentifier(), client);
 230                 } else {
 231                     LOG.debug("*This* instance of southbound plugin is *not* an owner of the device {}", node);
 232                 }
 233             } else {
 234                 LOG.debug("Did not find client for {}", node);
 235             }
 236         }
 237         return result;
 238     }
 239 }