applications/forwardingrules-sync/src/main/java/org/opendaylight/openflowplugin/applications/frsync/impl/SimplifiedOperationalListener.java

   1 /**
   2  * Copyright (c) 2016 Cisco Systems, 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.openflowplugin.applications.frsync.impl;
  10
  11 import java.util.Collection;
  12 import java.util.List;
  13
  14 import org.opendaylight.controller.md.sal.binding.api.DataObjectModification;
  15 import org.opendaylight.controller.md.sal.binding.api.DataObjectModification.ModificationType;
  16 import org.opendaylight.controller.md.sal.binding.api.DataTreeModification;
  17 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  18 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
  19 import org.opendaylight.openflowplugin.applications.frsync.SyncReactor;
  20 import org.opendaylight.openflowplugin.applications.frsync.dao.FlowCapableNodeDao;
  21 import org.opendaylight.openflowplugin.applications.frsync.dao.FlowCapableNodeSnapshotDao;
  22 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
  23 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
  24 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes;
  25 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnector;
  26 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
  27 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
  28 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  29 import org.slf4j.Logger;
  30 import org.slf4j.LoggerFactory;
  31
  32 import com.google.common.base.Optional;
  33 import com.google.common.util.concurrent.ListenableFuture;
  34
  35 /**
  36  * Listens to operational new nodes and delegates add/remove/update/barrier to {@link SyncReactor}.
  37  */
  38 public class SimplifiedOperationalListener extends AbstractFrmSyncListener<Node> {
  39     private static final Logger LOG = LoggerFactory.getLogger(SimplifiedOperationalListener.class);
  40
  41     protected final SyncReactor reactor;
  42
  43     private FlowCapableNodeSnapshotDao operationalSnaphot;
  44
  45     private FlowCapableNodeDao configDao;
  46
  47     public SimplifiedOperationalListener(SyncReactor reactor,
  48             FlowCapableNodeSnapshotDao operationalSnaphot, FlowCapableNodeDao configDao) {
  49         this.reactor = reactor;
  50         this.operationalSnaphot = operationalSnaphot;
  51         this.configDao = configDao;
  52     }
  53
  54     @Override
  55     public void onDataTreeChanged(Collection<DataTreeModification<Node>> modifications) {
  56         LOG.trace("Inventory Operational changes {}", modifications.size());
  57         super.onDataTreeChanged(modifications);
  58     }
  59
  60     /**
  61      * This method behaves like this:
  62      * <ul>
  63      * <li>If node is added to operational store then reconciliation.</li>
  64      * <li>Node is deleted from operational cache is removed.</li>
  65      * <li>Skip this event otherwise.</li>
  66      * </ul>
  67      *
  68      * @throws InterruptedException from syncup
  69      */
  70     protected Optional<ListenableFuture<Boolean>> processNodeModification(
  71             DataTreeModification<Node> modification) throws ReadFailedException, InterruptedException {
  72         updateCache(modification);
  73
  74         if (isAdd(modification) || isAddLogical(modification)) {
  75             return reconciliation(modification);
  76         }
  77         // TODO: else = explicit reconciliation required
  78
  79         return skipModification(modification);
  80     }
  81
  82     /**
  83      * Remove if delete. Update only if FlowCapableNode Augmentation modified.
  84      *
  85      * @param modification
  86      */
  87     protected void updateCache(DataTreeModification<Node> modification) {
  88         try {
  89             boolean isDelete = isDelete(modification) || isDeleteLogical(modification);
  90             if (isDelete) {
  91                 operationalSnaphot.updateCache(nodeId(modification), Optional.<FlowCapableNode>absent());
  92                 return;
  93             }
  94
  95             operationalSnaphot.updateCache(nodeId(modification), Optional.fromNullable(flowCapableNodeAfter(modification)));
  96         } catch(Exception e) {
  97             LOG.error("update cache failed {}", nodeId(modification), e);
  98         }
  99     }
 100
 101     protected Optional<ListenableFuture<Boolean>> skipModification(DataTreeModification<Node> modification) {
 102         LOG.trace("Skipping Inventory Operational modification {}, before {}, after {}", nodeIdValue(modification),
 103                 modification.getRootNode().getDataBefore() == null ? "null" : "nonnull",
 104                 modification.getRootNode().getDataAfter() == null ? "null" : "nonnull");
 105         return Optional.absent();// skip otherwise event
 106     }
 107
 108     /**
 109      * ModificationType.DELETE
 110      */
 111     protected boolean isDelete(DataTreeModification<Node> modification) {
 112         if (ModificationType.DELETE == modification.getRootNode().getModificationType()) {
 113             LOG.trace("Delete {} (physical)", nodeIdValue(modification));
 114             return true;
 115         }
 116
 117         return false;
 118     }
 119
 120     /**
 121      * All connectors disappeared from operational store (logical delete).
 122      */
 123     protected boolean isDeleteLogical(DataTreeModification<Node> modification) {
 124         final DataObjectModification<Node> rootNode = modification.getRootNode();
 125         if (!safeConnectorsEmpty(rootNode.getDataBefore()) && safeConnectorsEmpty(rootNode.getDataAfter())) {
 126             LOG.trace("Delete {} (logical)", nodeIdValue(modification));
 127             return true;
 128         }
 129
 130         return false;
 131     }
 132
 133     protected boolean isAdd(DataTreeModification<Node> modification) {
 134         final DataObjectModification<Node> rootNode = modification.getRootNode();
 135         final Node dataAfter = rootNode.getDataAfter();
 136         final Node dataBefore = rootNode.getDataBefore();
 137
 138         final boolean nodeAppearedInOperational = dataBefore == null && dataAfter != null;
 139         if (nodeAppearedInOperational) {
 140             LOG.trace("Add {} (physical)", nodeIdValue(modification));
 141         }
 142         return nodeAppearedInOperational;
 143     }
 144
 145     /**
 146      * All connectors appeared in operational store (logical add).
 147      */
 148     protected boolean isAddLogical(DataTreeModification<Node> modification) {
 149         final DataObjectModification<Node> rootNode = modification.getRootNode();
 150         if (safeConnectorsEmpty(rootNode.getDataBefore()) && !safeConnectorsEmpty(rootNode.getDataAfter())) {
 151             LOG.trace("Add {} (logical)", nodeIdValue(modification));
 152             return true;
 153         }
 154
 155         return false;
 156     }
 157
 158     protected Optional<ListenableFuture<Boolean>> reconciliation(
 159             DataTreeModification<Node> modification) throws InterruptedException {
 160         final NodeId nodeId = nodeId(modification);
 161
 162         LOG.debug("reconciliation {}", nodeId.getValue());
 163
 164         final Optional<FlowCapableNode> nodeConfiguration = configDao.loadByNodeId(nodeId);
 165         final InstanceIdentifier<FlowCapableNode> nodePath = InstanceIdentifier.create(Nodes.class)
 166                 .child(Node.class, new NodeKey(nodeId(modification))).augmentation(FlowCapableNode.class);
 167         final ListenableFuture<Boolean> rpcResult =
 168                 reactor.syncup(nodePath, nodeConfiguration.orNull(), flowCapableNodeAfter(modification));
 169         return Optional.of(rpcResult);
 170     }
 171
 172     static FlowCapableNode flowCapableNodeAfter(DataTreeModification<Node> modification) {
 173         final Node dataAfter = modification.getRootNode().getDataAfter();
 174         if (dataAfter == null) {
 175             return null;
 176         }
 177         return dataAfter.getAugmentation(FlowCapableNode.class);
 178     }
 179
 180     static boolean safeConnectorsEmpty(Node node) {
 181         if (node == null) {
 182             return true;
 183         }
 184
 185         final List<NodeConnector> nodeConnectors = node.getNodeConnector();
 186         if (nodeConnectors == null || nodeConnectors.isEmpty()) {
 187             return true;
 188         }
 189
 190         return false;
 191     }
 192
 193     static String nodeIdValue(DataTreeModification<Node> modification) {
 194         final NodeId nodeId = nodeId(modification);
 195
 196         if (nodeId == null) {
 197             return null;
 198         }
 199
 200         return nodeId.getValue();
 201     }
 202
 203     static NodeId nodeId(DataTreeModification<Node> modification) {
 204         final DataObjectModification<Node> rootNode = modification.getRootNode();
 205         final Node dataAfter = rootNode.getDataAfter();
 206
 207
 208         if (dataAfter != null) {
 209             return dataAfter.getId();
 210         }
 211
 212         final Node dataBefore = rootNode.getDataBefore();
 213         if (dataBefore != null) {
 214             return dataBefore.getId();
 215         }
 216
 217         return null;
 218     }
 219
 220     @Override
 221     public LogicalDatastoreType dsType() {
 222         return LogicalDatastoreType.OPERATIONAL;
 223     }
 224 }