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