hwvtepsouthbound/hwvtepsouthbound-impl/src/main/java/org/opendaylight/ovsdb/hwvtepsouthbound/HwvtepOperGlobalListener.java

   1 /*
   2  * Copyright (c) 2017 Ericsson India Global Services Pvt Ltd. 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.hwvtepsouthbound;
   9
  10 import java.util.ArrayList;
  11 import java.util.Collection;
  12 import java.util.List;
  13 import java.util.Map;
  14 import java.util.Objects;
  15 import java.util.concurrent.Callable;
  16 import java.util.concurrent.ConcurrentHashMap;
  17 import java.util.concurrent.ScheduledFuture;
  18 import java.util.concurrent.TimeUnit;
  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.DataObjectModification.ModificationType;
  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.utils.mdsal.utils.Scheduler;
  28 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.ovsdb.hwvtep.rev150901.HwvtepGlobalAugmentation;
  29 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.ovsdb.hwvtep.rev150901.hwvtep.global.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.InstanceIdentifier;
  36 import org.slf4j.Logger;
  37 import org.slf4j.LoggerFactory;
  38
  39 public class HwvtepOperGlobalListener implements ClusteredDataTreeChangeListener<Node>, AutoCloseable {
  40
  41     private static final Logger LOG = LoggerFactory.getLogger(HwvtepOperGlobalListener.class);
  42     private static final Map<InstanceIdentifier<Node>, ConnectionInfo> NODE_CONNECTION_INFO = new ConcurrentHashMap<>();
  43     private static final Map<InstanceIdentifier<Node>, ScheduledFuture> TIMEOUT_FTS = new ConcurrentHashMap<>();
  44
  45     private ListenerRegistration<HwvtepOperGlobalListener> registration;
  46     private final HwvtepConnectionManager hcm;
  47     private final DataBroker db;
  48     private static final Map<InstanceIdentifier<Node>, List<Callable<Void>>> NODE_DELET_WAITING_JOBS
  49             = new ConcurrentHashMap<>();
  50     private static final Map<InstanceIdentifier<Node>, Node> CONNECTED_NODES = new ConcurrentHashMap<>();
  51
  52
  53     HwvtepOperGlobalListener(final DataBroker db, HwvtepConnectionManager hcm) {
  54         LOG.info("Registering HwvtepOperGlobalListener");
  55         this.db = db;
  56         this.hcm = hcm;
  57         registerListener();
  58     }
  59
  60     private void registerListener() {
  61         final DataTreeIdentifier<Node> treeId =
  62                         new DataTreeIdentifier<Node>(LogicalDatastoreType.OPERATIONAL, getWildcardPath());
  63
  64         registration = db.registerDataTreeChangeListener(treeId, HwvtepOperGlobalListener.this);
  65     }
  66
  67     @Override
  68     public void close() {
  69         if (registration != null) {
  70             registration.close();
  71         }
  72     }
  73
  74     @Override
  75     @SuppressWarnings("checkstyle:IllegalCatch")
  76     public void onDataTreeChanged(final Collection<DataTreeModification<Node>> changes) {
  77         LOG.trace("onDataTreeChanged: ");
  78         try {
  79             connect(changes);
  80             updated(changes);
  81             disconnect(changes);
  82         } catch (Exception e) {
  83             LOG.error("Failed to handle dcn event ", e);
  84         }
  85     }
  86
  87     public static void runAfterTimeoutIfNodeNotCreated(InstanceIdentifier<Node> iid, Runnable job) {
  88         ScheduledFuture<?> ft = TIMEOUT_FTS.get(iid);
  89         if (ft != null) {
  90             ft.cancel(false);
  91         }
  92         ft = Scheduler.getScheduledExecutorService().schedule(() -> {
  93             TIMEOUT_FTS.remove(iid);
  94             if (!NODE_CONNECTION_INFO.containsKey(iid)) {
  95                 job.run();
  96             }
  97         }, HwvtepSouthboundConstants.EOS_TIMEOUT, TimeUnit.SECONDS);
  98         TIMEOUT_FTS.put(iid, ft);
  99     }
 100
 101     public void runAfterNodeDeleted(InstanceIdentifier<Node> iid, Callable<Void> job) throws Exception {
 102         synchronized (HwvtepOperGlobalListener.class) {
 103             if (NODE_DELET_WAITING_JOBS.containsKey(iid)) {
 104                 LOG.error("Node present in the cache {} adding to delete queue", iid);
 105                 NODE_DELET_WAITING_JOBS.get(iid).add(job);
 106                 //Also delete the node so that reconciliation kicks in
 107                 deleteTheNodeOfOldConnection(iid, getNodeConnectionInfo(iid));
 108                 HwvtepSouthboundUtil.getScheduledExecutorService().schedule(() -> {
 109                     runPendingJobs(iid);
 110                 }, HwvtepSouthboundConstants.HWVTEP_REGISTER_CALLBACKS_WAIT_TIMEOUT, TimeUnit.SECONDS);
 111             } else {
 112                 LOG.info("Node not present in the cache {} running the job now", iid);
 113                 job.call();
 114             }
 115         }
 116     }
 117
 118     @SuppressWarnings("checkstyle:IllegalCatch")
 119     private synchronized void runPendingJobs(InstanceIdentifier<Node> iid) {
 120         List<Callable<Void>> jobs = NODE_DELET_WAITING_JOBS.remove(iid);
 121         if (jobs != null && !jobs.isEmpty()) {
 122             jobs.forEach((job) -> {
 123                 try {
 124                     LOG.error("Node disconnected job found {} running it now ", iid);
 125                     job.call();
 126                 } catch (Exception e) {
 127                     LOG.error("Failed to run callable ", e);
 128                 }
 129             });
 130             jobs.clear();
 131         }
 132     }
 133
 134     private void connect(Collection<DataTreeModification<Node>> changes) {
 135         changes.forEach((change) -> {
 136             InstanceIdentifier<Node> key = change.getRootPath().getRootIdentifier();
 137             DataObjectModification<Node> mod = change.getRootNode();
 138             Node node = getCreated(mod);
 139             if (node == null) {
 140                 return;
 141             }
 142             CONNECTED_NODES.put(key, node);
 143             ScheduledFuture ft = TIMEOUT_FTS.remove(key);
 144             if (ft != null) {
 145                 ft.cancel(false);
 146             }
 147             HwvtepGlobalAugmentation globalAugmentation = node.augmentation(HwvtepGlobalAugmentation.class);
 148             if (globalAugmentation != null) {
 149                 ConnectionInfo connectionInfo = globalAugmentation.getConnectionInfo();
 150                 if (connectionInfo != null) {
 151                     NODE_CONNECTION_INFO.put(key, connectionInfo);
 152                 }
 153             }
 154             if (node != null) {
 155                 synchronized (HwvtepOperGlobalListener.class) {
 156                     NODE_DELET_WAITING_JOBS.putIfAbsent(key, new ArrayList<>());
 157                 }
 158             }
 159         });
 160     }
 161
 162     private void updated(Collection<DataTreeModification<Node>> changes) {
 163         changes.forEach((change) -> {
 164             InstanceIdentifier<Node> key = change.getRootPath().getRootIdentifier();
 165             DataObjectModification<Node> mod = change.getRootNode();
 166             Node node = getUpdated(mod);
 167             if (node != null) {
 168                 CONNECTED_NODES.put(key, node);
 169             }
 170         });
 171     }
 172
 173     public static Node getNode(final InstanceIdentifier<Node> key) {
 174         return CONNECTED_NODES.get(key);
 175     }
 176
 177     private void disconnect(Collection<DataTreeModification<Node>> changes) {
 178         changes.forEach((change) -> {
 179             InstanceIdentifier<Node> key = change.getRootPath().getRootIdentifier();
 180             DataObjectModification<Node> mod = change.getRootNode();
 181             Node node = getRemoved(mod);
 182             if (node != null) {
 183                 CONNECTED_NODES.remove(key);
 184                 NODE_CONNECTION_INFO.remove(key);
 185                 synchronized (HwvtepOperGlobalListener.class) {
 186                     runPendingJobs(key);
 187                 }
 188             }
 189         });
 190     }
 191
 192     private static String getNodeId(InstanceIdentifier<Node> iid) {
 193         return iid.firstKeyOf(Node.class).getNodeId().getValue();
 194     }
 195
 196     public void scheduleOldConnectionNodeDelete(InstanceIdentifier<Node> iid) {
 197         ConnectionInfo oldConnectionInfo = getNodeConnectionInfo(iid);
 198         HwvtepSouthboundUtil.getScheduledExecutorService().schedule(() -> {
 199             deleteTheNodeOfOldConnection(iid, oldConnectionInfo);
 200         }, HwvtepSouthboundConstants.STALE_HWVTEP_CLEANUP_DELAY_SECS, TimeUnit.SECONDS);
 201     }
 202
 203     private void deleteTheNodeOfOldConnection(InstanceIdentifier<Node> iid,
 204                                                     ConnectionInfo oldConnectionInfo) {
 205         if (oldConnectionInfo == null) {
 206             return;
 207         }
 208         ConnectionInfo latestConnectionInfo = getNodeConnectionInfo(iid);
 209         if (Objects.equals(latestConnectionInfo, oldConnectionInfo)) {
 210             //Still old connection node is not deleted
 211             LOG.debug("Delete Node {} from oper ", getNodeId(iid));
 212             hcm.cleanupOperationalNode(iid);
 213         }
 214     }
 215
 216     private static ConnectionInfo getNodeConnectionInfo(InstanceIdentifier<Node> iid) {
 217         return NODE_CONNECTION_INFO.get(iid);
 218     }
 219
 220     private static Node getCreated(final DataObjectModification<Node> mod) {
 221         if (mod.getModificationType() == ModificationType.WRITE && mod.getDataBefore() == null) {
 222             return mod.getDataAfter();
 223         }
 224         return null;
 225     }
 226
 227     private static Node getRemoved(final DataObjectModification<Node> mod) {
 228         if (mod.getModificationType() == ModificationType.DELETE) {
 229             return mod.getDataBefore();
 230         }
 231         return null;
 232     }
 233
 234     private static InstanceIdentifier<Node> getWildcardPath() {
 235         return InstanceIdentifier.create(NetworkTopology.class)
 236                 .child(Topology.class, new TopologyKey(HwvtepSouthboundConstants.HWVTEP_TOPOLOGY_ID))
 237                 .child(Node.class);
 238     }
 239
 240     private Node getUpdated(DataObjectModification<Node> mod) {
 241         Node node = null;
 242         switch (mod.getModificationType()) {
 243             case SUBTREE_MODIFIED:
 244                 node = mod.getDataAfter();
 245                 break;
 246             case WRITE:
 247                 if (mod.getDataBefore() !=  null) {
 248                     node = mod.getDataAfter();
 249                 }
 250                 break;
 251             default:
 252                 break;
 253         }
 254         return node;
 255     }
 256
 257
 258 }