openstack/net-virt/src/main/java/org/opendaylight/ovsdb/openstack/netvirt/impl/OvsdbDataChangeListener.java

   1 /*
   2  * Copyright (c) 2015 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 package org.opendaylight.ovsdb.openstack.netvirt.impl;
   9
  10 import java.util.Map;
  11 import java.util.Set;
  12 import java.util.concurrent.ExecutorService;
  13 import java.util.concurrent.Executors;
  14
  15 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  16 import org.opendaylight.controller.md.sal.binding.api.DataChangeListener;
  17 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope;
  18 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeEvent;
  19 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  20 import org.opendaylight.ovsdb.openstack.netvirt.MdsalHelper;
  21 import org.opendaylight.ovsdb.openstack.netvirt.api.Action;
  22 import org.opendaylight.ovsdb.openstack.netvirt.api.OvsdbInventoryListener;
  23 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.ovsdb.rev150105.OvsdbBridgeAugmentation;
  24 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.ovsdb.rev150105.OvsdbNodeAugmentation;
  25 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.ovsdb.rev150105.OvsdbTerminationPointAugmentation;
  26 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NetworkTopology;
  27 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.Topology;
  28 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyKey;
  29 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
  30 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  31 import org.opendaylight.yangtools.yang.binding.DataObject;
  32 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  33 import org.slf4j.Logger;
  34 import org.slf4j.LoggerFactory;
  35
  36 /**
  37  * MDSAL dataChangeListener for the OVSDB Southbound
  38  *
  39  * @author Sam Hague (shague@redhat.com)
  40  */
  41 public class OvsdbDataChangeListener implements DataChangeListener, AutoCloseable {
  42     private static final Logger LOG = LoggerFactory.getLogger(OvsdbDataChangeListener.class);
  43     private DataBroker dataBroker = null;
  44     private ListenerRegistration<DataChangeListener> registration;
  45     private final ExecutorService executorService = Executors.newFixedThreadPool(1);
  46
  47
  48     public OvsdbDataChangeListener (DataBroker dataBroker) {
  49         this.dataBroker = dataBroker;
  50         InstanceIdentifier<Node> path = InstanceIdentifier
  51                 .create(NetworkTopology.class)
  52                 .child(Topology.class, new TopologyKey(MdsalHelper.OVSDB_TOPOLOGY_ID))
  53                 .child(Node.class);
  54         registration = dataBroker.registerDataChangeListener(LogicalDatastoreType.OPERATIONAL, path, this,
  55                 DataChangeScope.SUBTREE);
  56         LOG.info("netvirt OvsdbDataChangeListener: dataBroker= {}, registration= {}",
  57                 dataBroker, registration);
  58     }
  59
  60     @Override
  61     public void close () throws Exception {
  62         registration.close();
  63         executorService.shutdown();
  64     }
  65
  66     @Override
  67     public void onDataChanged(final AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> changes) {
  68
  69         executorService.submit(new Runnable(){
  70
  71             @Override
  72             public void run() {
  73                 LOG.trace(">>>>> onDataChanged: {}", changes);
  74                 processOvsdbConnections(changes);
  75                 processOvsdbConnectionAttributeUpdates(changes);
  76                 processBridgeCreation(changes);
  77                 processBridgeUpdate(changes);
  78                 processPortCreation(changes);
  79                 processPortUpdate(changes);
  80                 processPortDeletion(changes);
  81                 processBridgeDeletion(changes);
  82                 processOvsdbDisconnect(changes);
  83             }
  84         });
  85     }
  86
  87     private void processOvsdbConnections(AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> changes) {
  88         for (Map.Entry<InstanceIdentifier<?>, DataObject> created : changes.getCreatedData().entrySet()) {
  89             if (created.getValue() instanceof OvsdbNodeAugmentation) {
  90                 Node ovsdbNode = getNode(changes.getCreatedData(), created);
  91                 LOG.trace("processOvsdbConnections: <{}>, ovsdbNode: <{}>", created, ovsdbNode);
  92                 ovsdbUpdate(ovsdbNode, created.getValue(), OvsdbInventoryListener.OvsdbType.NODE, Action.ADD);
  93             }
  94         }
  95     }
  96
  97     private void processOvsdbDisconnect(
  98             AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> changes) {
  99
 100         for(InstanceIdentifier<?> removedOvsdbNode : changes.getRemovedPaths()) {
 101             if(removedOvsdbNode.getTargetType().equals(OvsdbNodeAugmentation.class)){
 102                 //Get top node to get details of all the bridge/termination point augmentation
 103                 // in case we want to do any cleanup task while processing node disconnection
 104                 Node parentNode = getNode(changes.getOriginalData(), removedOvsdbNode);
 105                 if(parentNode == null){
 106                     //Throwing this warning in case behavior of southbound plugin changes.
 107                     LOG.warn("OvsdbNode's {} parent node details are not present in original data,"
 108                             + " it should not happen", parentNode);
 109                     continue;
 110                 }
 111                 //Fetch data of removed connection info from original data
 112                 @SuppressWarnings("unchecked")
 113                 OvsdbNodeAugmentation removedOvsdbNodeAugmentationData = getDataChanges(changes.getOriginalData(),
 114                         (InstanceIdentifier<OvsdbNodeAugmentation>) removedOvsdbNode);
 115
 116                 LOG.trace("processOvsdbDisconnect: {} ", removedOvsdbNode);
 117                 ////Assuming Openvswitch type represent the ovsdb node connection and not OvsdbType.NODE
 118
 119                 ovsdbUpdate(parentNode, removedOvsdbNodeAugmentationData,
 120                         OvsdbInventoryListener.OvsdbType.NODE, Action.DELETE);
 121             }
 122         }
 123     }
 124
 125     private void processOvsdbConnectionAttributeUpdates(
 126             AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> changes) {
 127
 128         for(Map.Entry<InstanceIdentifier<?>, DataObject> updatedOvsdbNode : changes.getUpdatedData().entrySet()){
 129             if(updatedOvsdbNode.getKey().getTargetType().equals(OvsdbNodeAugmentation.class)){
 130                 LOG.trace("processOvsdbConnectionAttributeUpdates: {}", updatedOvsdbNode);
 131                 /* XXX (NOTE): Till now we don't really need the old ovsdb connection attributes data before update.
 132                  * I am passing the updated data of both Node and resource augmentation data (connection attributes).
 133                  * If in future we need old OvsdbNodeAugmentation attributes data, we will extract it from
 134                  * original data and pass it as a resourceAugmentationData.
 135                  */
 136                 Node parentNode  = getNode(changes.getUpdatedData(), updatedOvsdbNode);
 137                 if (parentNode == null) {
 138                     // Logging this warning, to catch any change in southbound plugin's behavior.
 139                     LOG.warn("Parent Node for OvsdbNodeAugmentation is not found. On OvsdbNodeAugmentation update "
 140                             + "data store must provide the parent node update. This condition should not occur "
 141                             + "with the existing models defined in southbound plugin." );
 142                     continue;
 143                 }
 144                 LOG.trace("processOvsdbConnectionAttributeUpdates <{}> related update on Node: <{}>",
 145                         updatedOvsdbNode.getValue(), parentNode);
 146
 147                 ovsdbUpdate(parentNode, updatedOvsdbNode.getValue(),
 148                         OvsdbInventoryListener.OvsdbType.NODE, Action.UPDATE);
 149             }
 150         }
 151     }
 152
 153     private void processPortCreation(
 154             AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> changes) {
 155         for(Map.Entry<InstanceIdentifier<?>, DataObject> newPort : changes.getCreatedData().entrySet()){
 156             if(newPort.getKey().getTargetType().equals(OvsdbTerminationPointAugmentation.class)){
 157                 //LOG.trace("processPortCreation: {}", newPort);
 158                 //If user created termination point only, Node will get updated
 159                 Node tpParentNode  = getNode(changes.getUpdatedData(), newPort);
 160                 if(tpParentNode == null){
 161                     //If user created port with the bridge itself, Node will be in created data
 162                     tpParentNode = getNode(changes.getCreatedData(),newPort);
 163                 }
 164                 if(tpParentNode == null){
 165                     // Logging this warning, to make sure we didn't change anything
 166                     // in southbound plugin that changes this behavior.
 167                     LOG.warn("Parent Node for port is not found. Port creation must create or "
 168                             + "update the Node. This condition should not occur." );
 169                     continue;
 170                 }
 171
 172                 LOG.trace("processPortCreation <{}> creation on Node <{}>", newPort.getValue(), tpParentNode);
 173                 ovsdbUpdate(tpParentNode, newPort.getValue(),OvsdbInventoryListener.OvsdbType.PORT, Action.ADD);
 174             }
 175         }
 176     }
 177
 178     private void processPortDeletion(
 179             AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> changes) {
 180
 181         for(InstanceIdentifier<?> removedPort : changes.getRemovedPaths()) {
 182             if(removedPort.getTargetType().equals(OvsdbTerminationPointAugmentation.class)){
 183                 Node tpParentNode = getNode(changes.getOriginalData(), removedPort);
 184                 if(tpParentNode == null){
 185                     //Throwing this warning in case behavior of southbound plugin changes.
 186                     LOG.warn("Port's {} parent node details are not present in original data, "
 187                             + "it should not happen", removedPort);
 188                     continue;
 189                 }
 190                 //Fetch data of removed port from original data
 191                 @SuppressWarnings("unchecked")
 192                 OvsdbTerminationPointAugmentation removedTPAugmentationData = getDataChanges(changes.getOriginalData(),
 193                         (InstanceIdentifier<OvsdbTerminationPointAugmentation>)removedPort);
 194
 195                 LOG.trace("processPortDeletion <{}> deletion on Node <{}>", removedPort, tpParentNode);
 196                 ovsdbUpdate(tpParentNode, removedTPAugmentationData,
 197                         OvsdbInventoryListener.OvsdbType.PORT, Action.DELETE);
 198             }
 199         }
 200     }
 201
 202     private void processPortUpdate(
 203             AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> changes) {
 204
 205         for (Map.Entry<InstanceIdentifier<?>, DataObject> updatedPort : changes.getUpdatedData().entrySet()){
 206             if (updatedPort.getKey().getTargetType().equals(OvsdbTerminationPointAugmentation.class)){
 207                 //LOG.trace("processPortUpdate: <{}>", updatedPort);
 208                 /* XXX (NOTE): Till now we don't really need the old termination point data before update.
 209                  * I am passing the updated data of both Node and resource augmentation data (termination-point).
 210                  * If in future we need old TerminationPointAugmentation data, we will extract it from
 211                  * original data and pass it as a resourceAugmentationData.
 212                  */
 213                 Node tpParentNode  = getNode(changes.getUpdatedData(),updatedPort);
 214                 if (tpParentNode == null){
 215                     // Logging this warning, to catch any change in southbound plugin's behavior.
 216                     LOG.warn("Parent Node for port is not found. On Port/Interface update data store"
 217                             + " must provide the parent node update. This condition should not occure "
 218                             + "with the existing models define in southbound plugin." );
 219                     continue;
 220                 }
 221
 222                 LOG.trace("processPortUpdate <{}> update on Node <{}>", updatedPort.getValue(), tpParentNode);
 223                 ovsdbUpdate(tpParentNode, updatedPort.getValue(),
 224                         OvsdbInventoryListener.OvsdbType.PORT, Action.UPDATE);
 225             }
 226         }
 227     }
 228
 229     private void processBridgeCreation(
 230             AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> changes) {
 231
 232         for(Map.Entry<InstanceIdentifier<?>, DataObject> newBridge : changes.getCreatedData().entrySet()){
 233             if(newBridge.getKey().getTargetType().equals(OvsdbBridgeAugmentation.class)){
 234                 //LOG.trace("processBridgeCreation <{}>", newBridge);
 235                 //Bridge augmentation happens directly on the Node so Node details should also exist in created data.
 236                 Node bridgeParentNode  = getNode(changes.getCreatedData(),newBridge);
 237                 if(bridgeParentNode == null){
 238                     // Logging this warning, to catch any change in southbound plugin behavior
 239                     LOG.warn("Parent Node for bridge is not found. Bridge creation must provide the Node "
 240                             + "details in create Data Changes. This condition should not occur." );
 241                     continue;
 242                 }
 243                 LOG.trace("processBridgeCreation <{}> creation on Node <{}>", newBridge.getValue(), bridgeParentNode);
 244                 ovsdbUpdate(bridgeParentNode, newBridge.getValue(),
 245                         OvsdbInventoryListener.OvsdbType.BRIDGE, Action.ADD);
 246             }
 247         }
 248     }
 249
 250     private void processBridgeUpdate(
 251             AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> changes) {
 252
 253         for (Map.Entry<InstanceIdentifier<?>, DataObject> updatedBridge : changes.getUpdatedData().entrySet()) {
 254             if(updatedBridge.getKey().getTargetType().equals(OvsdbBridgeAugmentation.class)){
 255                 //LOG.trace("processBridgeUpdate <{}>", updatedBridge);
 256                 /* XXX (NOTE): Till now we don't really need the old bridge data before update.
 257                  * I am passing the updated data of both Node and resource augmentation data.
 258                  * If in future we need old bridgeAugmentationData, we will extract it from
 259                  * original data and pass it as a resourceAugmentationData.
 260                  */
 261
 262                 Node bridgeParentNode = getNode(changes.getUpdatedData(), updatedBridge);
 263                 if(bridgeParentNode == null){
 264                     // Logging this warning, to catch any change in southbound plugin behavior
 265                     LOG.warn("Parent Node for bridge is not found. Bridge update must provide the Node "
 266                             + "details in updated Data Changes. This condition should not occure" );
 267                     continue;
 268                 }
 269                 LOG.trace("processBridgeUpdate <{}> update on Node <{}>", updatedBridge.getValue(), bridgeParentNode);
 270                 ovsdbUpdate(bridgeParentNode, updatedBridge.getValue(),
 271                         OvsdbInventoryListener.OvsdbType.BRIDGE, Action.UPDATE);
 272             }
 273         }
 274     }
 275
 276     private void processBridgeDeletion(
 277             AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> changes) {
 278
 279         for(InstanceIdentifier<?> removedBridge : changes.getRemovedPaths()) {
 280             if(removedBridge.getTargetType().equals(OvsdbBridgeAugmentation.class)){
 281                 Node bridgeParentNode = getNode(changes.getOriginalData(), removedBridge);
 282                 if(bridgeParentNode == null){
 283                     //Throwing this warning to catch the behavior change of southbound plugin.
 284                     LOG.warn("Bridge's {} parent node details are not present in original data"
 285                             + ", it should not happen", removedBridge);
 286                     continue;
 287                 }
 288                 //Fetch data of removed bridge from original data
 289                 @SuppressWarnings("unchecked")
 290                 OvsdbBridgeAugmentation removedBridgeAugmentationData = getDataChanges(changes.getOriginalData(),
 291                         (InstanceIdentifier<OvsdbBridgeAugmentation>) removedBridge);
 292
 293                 LOG.debug("processBridgeDeletion <{}> deletion on Node <{}>", removedBridge,bridgeParentNode);
 294                 ovsdbUpdate(bridgeParentNode, removedBridgeAugmentationData,
 295                         OvsdbInventoryListener.OvsdbType.BRIDGE, Action.DELETE);
 296             }
 297         }
 298     }
 299
 300     private Node getNode(Map<InstanceIdentifier<?>, DataObject> changes,
 301                          Map.Entry<InstanceIdentifier<?>, DataObject> change) {
 302         InstanceIdentifier<Node> nodeInstanceIdentifier = change.getKey().firstIdentifierOf(Node.class);
 303         return (Node)changes.get(nodeInstanceIdentifier);
 304     }
 305
 306     private Node getNode(Map<InstanceIdentifier<?>, DataObject> changes,InstanceIdentifier<?> path) {
 307         InstanceIdentifier<Node> nodeInstanceIdentifier = path.firstIdentifierOf(Node.class);
 308         return (Node)changes.get(nodeInstanceIdentifier);
 309     }
 310
 311     private <T extends DataObject> T getDataChanges(
 312             Map<InstanceIdentifier<?>, DataObject> changes,InstanceIdentifier<T> path){
 313
 314         for(Map.Entry<InstanceIdentifier<?>,DataObject> change : changes.entrySet()){
 315             if(change.getKey().getTargetType().equals(path.getTargetType())){
 316                 @SuppressWarnings("unchecked")
 317                 T dataObject = (T) change.getValue();
 318                 return dataObject;
 319             }
 320         }
 321         return null;
 322     }
 323
 324     private void ovsdbUpdate(Node node, DataObject resourceAugmentationDataChanges,
 325             OvsdbInventoryListener.OvsdbType ovsdbType, Action action) {
 326
 327         Set<OvsdbInventoryListener> mdsalConsumerListeners = OvsdbInventoryServiceImpl.getMdsalConsumerListeners();
 328         for (OvsdbInventoryListener mdsalConsumerListener : mdsalConsumerListeners) {
 329             mdsalConsumerListener.ovsdbUpdate(node, resourceAugmentationDataChanges, ovsdbType, action);
 330         }
 331     }
 332 }