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
  13 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  14 import org.opendaylight.controller.md.sal.binding.api.DataChangeListener;
  15 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope;
  16 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeEvent;
  17 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  18 import org.opendaylight.ovsdb.openstack.netvirt.api.Action;
  19 import org.opendaylight.ovsdb.openstack.netvirt.api.NodeCacheManager;
  20 import org.opendaylight.ovsdb.openstack.netvirt.api.OvsdbInventoryListener;
  21 import org.opendaylight.ovsdb.southbound.SouthboundConstants;
  22 import org.opendaylight.ovsdb.utils.servicehelper.ServiceHelper;
  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 NodeCacheManager nodeCacheManager = null;
  46
  47     public OvsdbDataChangeListener (DataBroker dataBroker) {
  48         LOG.info(">>>>> Registering OvsdbNodeDataChangeListener: dataBroker= {}", dataBroker);
  49         this.dataBroker = dataBroker;
  50         InstanceIdentifier<Node> path = InstanceIdentifier
  51                 .create(NetworkTopology.class)
  52                 .child(Topology.class, new TopologyKey(SouthboundConstants.OVSDB_TOPOLOGY_ID))
  53                 .child(Node.class);
  54         registration = dataBroker.registerDataChangeListener(LogicalDatastoreType.OPERATIONAL, path, this,
  55                 DataChangeScope.SUBTREE);
  56         LOG.info("netvirt OvsdbDataChangeListener: registration= {}", registration);
  57     }
  58
  59     @Override
  60     public void close () throws Exception {
  61         registration.close();
  62     }
  63
  64     /* TODO
  65      * Recognize when netvirt added a bridge to config and then the operational update comes in
  66      * can it be ignored or just viewed as a new switch? ports and interfaces can likely be mapped
  67      * to the old path where there were updates for them for update and insert row.
  68      */
  69     @Override
  70     public void onDataChanged(AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> changes) {
  71         LOG.info(">>>>> onDataChanged: {}", changes);
  72         //TODO: off load this process to execution service, blocking md-sal notification thread
  73         // has performance impact on overall controller performance. With new notification broker
  74         //it might create weird issues.
  75         processOvsdbConnections(changes);
  76         processOvsdbDisconnect(changes);
  77         processOvsdbConnectionAttributeUpdates(changes);
  78         processOpenflowConnections(changes);
  79         processBridgeCreation(changes);
  80         processBridgeDeletion(changes);
  81         processBridgeUpdate(changes);
  82         processPortCreation(changes);
  83         processPortDeletion(changes);
  84         processPortUpdate(changes);
  85
  86
  87     }
  88
  89     private void processOvsdbConnections(AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> changes) {
  90         LOG.trace("processOvsdbConnections - Received changes : {}",changes);
  91         for (Map.Entry<InstanceIdentifier<?>, DataObject> created : changes.getCreatedData().entrySet()) {
  92             if (created.getValue() instanceof OvsdbNodeAugmentation) {
  93                 LOG.info("Processing ovsdb connections : {}", created);
  94                 Node ovsdbNode = getNode(changes.getCreatedData(), created);
  95                 LOG.info("ovsdbNode: {}", ovsdbNode);
  96                 ovsdbUpdate(ovsdbNode, created.getValue(), OvsdbInventoryListener.OvsdbType.NODE, Action.ADD);
  97             }
  98         }
  99     }
 100
 101     private void processOvsdbDisconnect(
 102             AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> changes) {
 103         LOG.info("processOvsdbDisconnect - Received changes : {}", changes);
 104
 105         for(InstanceIdentifier<?> removedOvsdbNode : changes.getRemovedPaths()) {
 106             if(removedOvsdbNode.getTargetType().equals(OvsdbNodeAugmentation.class)){
 107                 //Get top node to get details of all the bridge/termination point augmentation
 108                 // in case we want to do any cleanup task while processing node disconnection
 109                 Node parentNode = getNode(changes.getOriginalData(), removedOvsdbNode);
 110                 if(parentNode == null){
 111                     //Throwing this warning in case behavior of southbound plugin changes.
 112                     LOG.warn("OvsdbNode's {} parent node details are not present in original data,"
 113                             + " it should not happen", parentNode);
 114                     continue;
 115                 }
 116                 //Fetch data of removed connection info from original data
 117                 @SuppressWarnings("unchecked")
 118                 OvsdbNodeAugmentation removedOvsdbNodeAugmentationData = getDataChanges(changes.getOriginalData(),
 119                         (InstanceIdentifier<OvsdbNodeAugmentation>)removedOvsdbNode);
 120
 121                 LOG.debug("Process ovsdb node delete : {} ", removedOvsdbNode);
 122                 //Assuming Openvswitch type represent the ovsdb node connection and not OvsdbType.NODE
 123
 124                 ovsdbUpdate(parentNode, removedOvsdbNodeAugmentationData,
 125                         OvsdbInventoryListener.OvsdbType.OPENVSWITCH, Action.DELETE);
 126             }
 127         }
 128     }
 129
 130     private void processOvsdbConnectionAttributeUpdates(
 131             AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> changes) {
 132         // TODO Auto-generated method stub
 133
 134     }
 135
 136     private void processOpenflowConnections(AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> changes) {
 137         LOG.trace("processOpenflowConnections - processOpenflowConnections created: {}", changes);
 138         for (Map.Entry<InstanceIdentifier<?>, DataObject> change : changes.getCreatedData().entrySet()) {
 139             if (change.getValue() instanceof OvsdbBridgeAugmentation) {
 140                 LOG.info("Processing OpenFlow connections : {}",change);
 141                 OvsdbBridgeAugmentation ovsdbBridgeAugmentation = (OvsdbBridgeAugmentation)change.getValue();
 142                 String datapathId = MdsalUtils.getDatapathId(ovsdbBridgeAugmentation);
 143                 // Having a datapathId means the bridge has connected so it exists
 144                 if (datapathId == null) {
 145                     LOG.info("dataPathId not found");
 146                     continue;
 147                 }
 148                 Node node = getNode(changes.getCreatedData(), change);
 149                 if (node == null) {
 150                     LOG.warn("node not found");
 151                     continue;
 152                 }
 153                 // This value is not being set right now - OvsdbBridgeUpdateCommand
 154                 //if (ovsdbBridgeAugmentation.getBridgeOpenflowNodeRef() != null) {
 155                     nodeCacheManager = (NodeCacheManager) ServiceHelper.getGlobalInstance(NodeCacheManager.class, this);
 156                     nodeCacheManager.nodeAdded(node);
 157                 //}
 158             }
 159         }
 160
 161         for (Map.Entry<InstanceIdentifier<?>, DataObject> change : changes.getUpdatedData().entrySet()) {
 162             if (change.getValue() instanceof OvsdbBridgeAugmentation) {
 163                 LOG.info("Processing OpenFlow connections updates: {}",change);
 164                 OvsdbBridgeAugmentation ovsdbBridgeAugmentation = (OvsdbBridgeAugmentation)change.getValue();
 165                 String datapathId = MdsalUtils.getDatapathId(ovsdbBridgeAugmentation);
 166                 // Having a datapathId means the bridge has connected so it exists
 167                 if (datapathId == null) {
 168                     LOG.info("dataPathId not found");
 169                     continue;
 170                 }
 171                 Node node = getNode(changes.getUpdatedData(), change);
 172                 if (node == null) {
 173                     LOG.warn("node not found");
 174                     continue;
 175                 }
 176                 // This value is not being set right now - OvsdbBridgeUpdateCommand
 177                 // if (ovsdbBridgeAugmentation.getBridgeOpenflowNodeRef() != null) {
 178                     nodeCacheManager = (NodeCacheManager) ServiceHelper.getGlobalInstance(
 179                             NodeCacheManager.class, this);
 180                     nodeCacheManager.nodeAdded(node);
 181                 //}
 182             }
 183         }
 184     }
 185
 186     private void processPortCreation(
 187             AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> changes) {
 188         LOG.info("processPortCreation - Received changes : {}", changes);
 189         for(Map.Entry<InstanceIdentifier<?>, DataObject> newPort : changes.getCreatedData().entrySet()){
 190             if(newPort.getKey() instanceof OvsdbTerminationPointAugmentation){
 191                 LOG.info("Processing creation of new port : {}",newPort);
 192                 //If user created termination point only, Node will get updated
 193                 Node tpParentNode  = getNode(changes.getUpdatedData(), newPort);
 194                 if(tpParentNode == null){
 195                     //If user created port with the bridge itself, Node will be in created data
 196                     tpParentNode = getNode(changes.getCreatedData(),newPort);
 197                 }
 198                 if(tpParentNode == null){
 199                     // Logging this warning, to make sure we didn't change anything
 200                     // in southbound plugin that changes this behavior.
 201                     LOG.warn("Parent Node for port is not found. Port creation must create or "
 202                             + "update the Node. This condition should not occure" );
 203                     continue;
 204                 }
 205
 206                 LOG.debug("Process new port {} creation on Node : {}", newPort.getValue(),tpParentNode);
 207                 ovsdbUpdate(tpParentNode, newPort.getValue(),OvsdbInventoryListener.OvsdbType.PORT, Action.ADD);
 208             }
 209         }
 210     }
 211
 212     private void processPortDeletion(
 213             AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> changes) {
 214         LOG.info("processPortDeletion - Received changes : {}", changes);
 215
 216         for(InstanceIdentifier<?> removedPort : changes.getRemovedPaths()) {
 217             if(removedPort.getTargetType().equals(OvsdbTerminationPointAugmentation.class)){
 218                 Node tpParentNode = getNode(changes.getOriginalData(), removedPort);
 219                 if(tpParentNode == null){
 220                     //Throwing this warning in case behavior of southbound plugin changes.
 221                     LOG.warn("Port's {} parent node details are not present in original data, "
 222                             + "it should not happen", removedPort);
 223                     continue;
 224                 }
 225                 //Fetch data of removed port from original data
 226                 @SuppressWarnings("unchecked")
 227                 OvsdbTerminationPointAugmentation removedTPAugmentationData = getDataChanges(changes.getOriginalData(),
 228                         (InstanceIdentifier<OvsdbTerminationPointAugmentation>)removedPort);
 229
 230                 LOG.debug("Process port {} deletion on Node : {}", removedPort,tpParentNode);
 231                 ovsdbUpdate(tpParentNode, removedTPAugmentationData,
 232                         OvsdbInventoryListener.OvsdbType.PORT, Action.DELETE);
 233
 234             }
 235         }
 236     }
 237
 238     private void processPortUpdate(
 239             AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> changes) {
 240         LOG.info("processPortUpdate - Received changes : {}", changes);
 241
 242         for(Map.Entry<InstanceIdentifier<?>, DataObject> updatedPort : changes.getUpdatedData().entrySet()){
 243             if(updatedPort.getKey() instanceof OvsdbTerminationPointAugmentation){
 244                 LOG.info("Processing port updates : {}",updatedPort);
 245                 // XXX (NOTE): Extract parent node data from originalData(), rather then extracting it from updatedData()
 246                 // because, extracting it from originalData() will give all the
 247                 // (existing data of parent node + old data of the OvsdbTerminationPointAugmentation).
 248                 // If we extract parent node data from updatedData, it will give us
 249                 // ( Node data + new OvsdbTermiantionPointAugmentation data). To determine the update in
 250                 // OvsdbTerminationPointAugmentation, we need to pass it's old and new values to ovsdbUpdate.
 251                 // We anyways pass new data of OvsdbTerminationPointAugmentation to ovsdbUpdate.
 252                 Node tpParentNode  = getNode(changes.getOriginalData(),updatedPort);
 253                 if(tpParentNode == null){
 254                     // Logging this warning, to catch any change in southbound plugin's behavior.
 255                     LOG.warn("Parent Node for port is not found. On Port/Interface update data store"
 256                             + " must provide the parent node update. This condition should not occure "
 257                             + "with the existing models define in southbound plugin." );
 258                     continue;
 259                 }
 260
 261                 LOG.debug("Process port's {} update on Node : {}", updatedPort.getValue(),tpParentNode);
 262                 ovsdbUpdate(tpParentNode, updatedPort.getValue(),OvsdbInventoryListener.OvsdbType.PORT, Action.UPDATE);
 263             }
 264         }
 265     }
 266
 267     private void processBridgeCreation(
 268             AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> changes) {
 269         LOG.info("processBridgeCreation - Received changes : {}", changes);
 270
 271         for(Map.Entry<InstanceIdentifier<?>, DataObject> newBridge : changes.getCreatedData().entrySet()){
 272             if(newBridge.getKey() instanceof OvsdbBridgeAugmentation){
 273                 LOG.info("Processing creation of new bridge : {}",newBridge);
 274                 //Bridge augmentation happens directly on the Node so Node details should also exist in created data.
 275                 Node bridgeParentNode  = getNode(changes.getCreatedData(),newBridge);
 276                 if(bridgeParentNode == null){
 277                     // Logging this warning, to catch any change in southbound plugin behavior
 278                     LOG.warn("Parent Node for bridge is not found. Bridge creation must provide the Node "
 279                             + "details in create Data Changes. This condition should not occure" );
 280                     continue;
 281                 }
 282                 LOG.debug("Process new bridge {} creation on Node : {}", newBridge.getValue(),bridgeParentNode);
 283                 ovsdbUpdate(bridgeParentNode, newBridge.getValue(),
 284                         OvsdbInventoryListener.OvsdbType.BRIDGE, Action.ADD);
 285             }
 286         }
 287     }
 288
 289     private void processBridgeUpdate(
 290             AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> changes) {
 291         // TODO Auto-generated method stub
 292
 293     }
 294
 295     private void processBridgeDeletion(
 296             AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> changes) {
 297         LOG.info("processBridgeDeletion - Received changes : {}", changes);
 298
 299         for(InstanceIdentifier<?> removedBridge : changes.getRemovedPaths()) {
 300             if(removedBridge.getTargetType().equals(OvsdbBridgeAugmentation.class)){
 301                 Node bridgeParentNode = getNode(changes.getOriginalData(), removedBridge);
 302                 if(bridgeParentNode == null){
 303                     //Throwing this warning to catch the behavior change of southbound plugin.
 304                     LOG.warn("Bridge's {} parent node details are not present in original data"
 305                             + ", it should not happen", removedBridge);
 306                     continue;
 307                 }
 308                 //Fetch data of removed bridge from original data
 309                 @SuppressWarnings("unchecked")
 310                 OvsdbBridgeAugmentation removedBridgeAugmentationData = getDataChanges(changes.getOriginalData(),
 311                         (InstanceIdentifier<OvsdbBridgeAugmentation>) removedBridge);
 312
 313                 LOG.debug("Process bridge {} deletion on Node : {}", removedBridge,bridgeParentNode);
 314                 ovsdbUpdate(bridgeParentNode, removedBridgeAugmentationData,
 315                         OvsdbInventoryListener.OvsdbType.BRIDGE, Action.DELETE);
 316             }
 317         }
 318     }
 319
 320     //TODO: Will remove it if not needed
 321     private Node getNodeFromCreatedData(AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> changes,
 322                          Map.Entry<InstanceIdentifier<?>, DataObject> change) {
 323         InstanceIdentifier<Node> nodeInstanceIdentifier = change.getKey().firstIdentifierOf(Node.class);
 324         return (Node)changes.getCreatedData().get(nodeInstanceIdentifier);
 325     }
 326
 327     private Node getNode(Map<InstanceIdentifier<?>, DataObject> changes,
 328                          Map.Entry<InstanceIdentifier<?>, DataObject> change) {
 329         InstanceIdentifier<Node> nodeInstanceIdentifier = change.getKey().firstIdentifierOf(Node.class);
 330         return (Node)changes.get(nodeInstanceIdentifier);
 331     }
 332
 333     private Node getNode(Map<InstanceIdentifier<?>, DataObject> changes,InstanceIdentifier<?> path) {
 334         InstanceIdentifier<Node> nodeInstanceIdentifier = path.firstIdentifierOf(Node.class);
 335         return (Node)changes.get(nodeInstanceIdentifier);
 336     }
 337
 338     private <T extends DataObject> T getDataChanges(Map<InstanceIdentifier<?>, DataObject> changes,InstanceIdentifier<T> path){
 339
 340         for(Map.Entry<InstanceIdentifier<?>,DataObject> change : changes.entrySet()){
 341             if(change.getKey().getTargetType().equals(path.getTargetType())){
 342                 @SuppressWarnings("unchecked")
 343                 T dataObject = (T) change.getValue();
 344                 return dataObject;
 345             }
 346         }
 347         return null;
 348     }
 349
 350     private void ovsdbUpdate(Node node, DataObject resourceAugmentationDataChanges, OvsdbInventoryListener.OvsdbType ovsdbType, Action action) {
 351         Set<OvsdbInventoryListener> mdsalConsumerListeners = OvsdbInventoryServiceImpl.getMdsalConsumerListeners();
 352         for (OvsdbInventoryListener mdsalConsumerListener : mdsalConsumerListeners) {
 353             mdsalConsumerListener.ovsdbUpdate(node, resourceAugmentationDataChanges, ovsdbType, action);
 354         }
 355     }
 356 }