elanmanager/impl/src/main/java/org/opendaylight/netvirt/elan/l2gw/listeners/HwvtepPhysicalSwitchListener.java

   1 /*
   2  * Copyright (c) 2016 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
   9 package org.opendaylight.netvirt.elan.l2gw.listeners;
  10
  11 import static org.opendaylight.genius.infra.Datastore.CONFIGURATION;
  12
  13 import com.google.common.base.Optional;
  14 import java.util.Collections;
  15 import java.util.HashSet;
  16 import java.util.List;
  17 import java.util.Objects;
  18 import java.util.Set;
  19 import java.util.function.BiPredicate;
  20 import java.util.function.Predicate;
  21 import javax.annotation.PostConstruct;
  22 import javax.inject.Inject;
  23 import javax.inject.Singleton;
  24 import org.eclipse.jdt.annotation.Nullable;
  25 import org.opendaylight.controller.md.sal.binding.api.ClusteredDataTreeChangeListener;
  26 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  27 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  28 import org.opendaylight.genius.datastoreutils.hwvtep.HwvtepAbstractDataTreeChangeListener;
  29 import org.opendaylight.genius.infra.ManagedNewTransactionRunner;
  30 import org.opendaylight.genius.infra.ManagedNewTransactionRunnerImpl;
  31 import org.opendaylight.genius.mdsalutil.MDSALUtil;
  32 import org.opendaylight.genius.utils.hwvtep.HwvtepNodeHACache;
  33 import org.opendaylight.genius.utils.hwvtep.HwvtepSouthboundConstants;
  34 import org.opendaylight.genius.utils.hwvtep.HwvtepSouthboundUtils;
  35 import org.opendaylight.infrautils.utils.concurrent.ListenableFutures;
  36 import org.opendaylight.netvirt.elan.l2gw.ha.HwvtepHAUtil;
  37 import org.opendaylight.netvirt.elan.l2gw.ha.listeners.HAOpClusteredListener;
  38 import org.opendaylight.netvirt.elan.l2gw.utils.ElanL2GatewayUtils;
  39 import org.opendaylight.netvirt.elan.l2gw.utils.L2GatewayConnectionUtils;
  40 import org.opendaylight.netvirt.elan.l2gw.utils.L2GatewayUtils;
  41 import org.opendaylight.netvirt.elan.l2gw.utils.L2gwServiceProvider;
  42 import org.opendaylight.netvirt.elan.l2gw.utils.StaleVlanBindingsCleaner;
  43 import org.opendaylight.netvirt.elan.utils.ElanClusterUtils;
  44 import org.opendaylight.netvirt.elanmanager.utils.ElanL2GwCacheUtils;
  45 import org.opendaylight.netvirt.neutronvpn.api.l2gw.L2GatewayCache;
  46 import org.opendaylight.netvirt.neutronvpn.api.l2gw.L2GatewayDevice;
  47 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.IpAddress;
  48 import org.opendaylight.yang.gen.v1.urn.opendaylight.genius.itm.rpcs.rev160406.ItmRpcService;
  49 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.ovsdb.hwvtep.rev150901.PhysicalSwitchAugmentation;
  50 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.ovsdb.hwvtep.rev150901.PhysicalSwitchAugmentationBuilder;
  51 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.ovsdb.hwvtep.rev150901.hwvtep.physical._switch.attributes.TunnelIps;
  52 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NetworkTopology;
  53 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NodeId;
  54 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.Topology;
  55 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyKey;
  56 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
  57 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.NodeKey;
  58 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  59 import org.slf4j.Logger;
  60 import org.slf4j.LoggerFactory;
  61
  62 /**
  63  * Listener to handle physical switch updates.
  64  */
  65 @Singleton
  66 public class HwvtepPhysicalSwitchListener
  67         extends HwvtepAbstractDataTreeChangeListener<PhysicalSwitchAugmentation, HwvtepPhysicalSwitchListener>
  68         implements ClusteredDataTreeChangeListener<PhysicalSwitchAugmentation> {
  69
  70     /** The Constant LOG. */
  71     private static final Logger LOG = LoggerFactory.getLogger(HwvtepPhysicalSwitchListener.class);
  72
  73     private static final BiPredicate<L2GatewayDevice, InstanceIdentifier<Node>> DEVICE_NOT_CACHED_OR_PARENT_CONNECTED =
  74         (l2GatewayDevice, globalIid) -> l2GatewayDevice == null || l2GatewayDevice.getHwvtepNodeId() == null
  75                 || !Objects.equals(l2GatewayDevice.getHwvtepNodeId(),
  76                         globalIid.firstKeyOf(Node.class).getNodeId().getValue());
  77
  78     private static final Predicate<PhysicalSwitchAugmentation> TUNNEL_IP_AVAILABLE =
  79         phySwitch -> !HwvtepHAUtil.isEmpty(phySwitch.getTunnelIps());
  80
  81     private static final Predicate<PhysicalSwitchAugmentation> TUNNEL_IP_NOT_AVAILABLE = TUNNEL_IP_AVAILABLE.negate();
  82
  83     private static final BiPredicate<PhysicalSwitchAugmentation, L2GatewayDevice> TUNNEL_IP_CHANGED =
  84         (phySwitchAfter, existingDevice) -> TUNNEL_IP_AVAILABLE.test(phySwitchAfter)
  85                 && !Objects.equals(
  86                         existingDevice.getTunnelIp(),  phySwitchAfter.getTunnelIps().get(0).getTunnelIpsKey());
  87
  88     /** The data broker. */
  89     private final DataBroker dataBroker;
  90     private final ManagedNewTransactionRunner txRunner;
  91
  92     /** The itm rpc service. */
  93     private final ItmRpcService itmRpcService;
  94
  95     private final ElanClusterUtils elanClusterUtils;
  96
  97     private final HwvtepNodeHACache hwvtepNodeHACache;
  98
  99     private final L2gwServiceProvider l2gwServiceProvider;
 100
 101     private final BiPredicate<L2GatewayDevice, InstanceIdentifier<Node>> childConnectedAfterParent;
 102
 103     private final Predicate<L2GatewayDevice> alreadyHasL2Gwids =
 104         (l2GwDevice) -> l2GwDevice != null && HwvtepHAUtil.isEmpty(l2GwDevice.getL2GatewayIds());
 105
 106     private final BiPredicate<L2GatewayDevice, InstanceIdentifier<Node>> parentConnectedAfterChild;
 107     private final HAOpClusteredListener haOpClusteredListener;
 108
 109     private final L2GatewayCache l2GatewayCache;
 110
 111     private final StaleVlanBindingsCleaner staleVlanBindingsCleaner;
 112
 113     /**
 114      * Instantiates a new hwvtep physical switch listener.
 115      */
 116     @Inject
 117     public HwvtepPhysicalSwitchListener(final DataBroker dataBroker, ItmRpcService itmRpcService,
 118             ElanClusterUtils elanClusterUtils, L2gwServiceProvider l2gwServiceProvider,
 119             HAOpClusteredListener haListener, L2GatewayCache l2GatewayCache,
 120             StaleVlanBindingsCleaner staleVlanBindingsCleaner,
 121             HwvtepNodeHACache hwvtepNodeHACache) {
 122         super(PhysicalSwitchAugmentation.class, HwvtepPhysicalSwitchListener.class, hwvtepNodeHACache);
 123         this.dataBroker = dataBroker;
 124         this.txRunner = new ManagedNewTransactionRunnerImpl(dataBroker);
 125         this.itmRpcService = itmRpcService;
 126         this.elanClusterUtils = elanClusterUtils;
 127         this.l2gwServiceProvider = l2gwServiceProvider;
 128         this.staleVlanBindingsCleaner = staleVlanBindingsCleaner;
 129         this.haOpClusteredListener = haListener;
 130         this.l2GatewayCache = l2GatewayCache;
 131         this.hwvtepNodeHACache = hwvtepNodeHACache;
 132
 133         childConnectedAfterParent = (l2GwDevice, globalIid) -> {
 134             return !hwvtepNodeHACache.isHAParentNode(globalIid)
 135                     && l2GwDevice != null && l2GwDevice.getHwvtepNodeId() != null
 136                     && !Objects.equals(l2GwDevice.getHwvtepNodeId(), globalIid.firstKeyOf(Node.class)
 137                     .getNodeId().getValue());
 138         };
 139
 140         parentConnectedAfterChild = (l2GwDevice, globalIid) -> {
 141             InstanceIdentifier<Node> existingIid = globalIid;
 142             if (l2GwDevice != null && l2GwDevice.getHwvtepNodeId() != null) {
 143                 existingIid = HwvtepHAUtil.convertToInstanceIdentifier(l2GwDevice.getHwvtepNodeId());
 144             }
 145             return hwvtepNodeHACache.isHAParentNode(globalIid)
 146                     && l2GwDevice != null && l2GwDevice.getHwvtepNodeId() != null
 147                     && !Objects.equals(l2GwDevice.getHwvtepNodeId(), globalIid.firstKeyOf(Node.class)
 148                     .getNodeId().getValue())
 149                     && Objects.equals(globalIid, hwvtepNodeHACache.getParent(existingIid));
 150         };
 151     }
 152
 153     @Override
 154     @PostConstruct
 155     public void init() {
 156         registerListener(LogicalDatastoreType.OPERATIONAL, dataBroker);
 157     }
 158
 159     @Override
 160     protected InstanceIdentifier<PhysicalSwitchAugmentation> getWildCardPath() {
 161         return InstanceIdentifier.create(NetworkTopology.class)
 162                 .child(Topology.class, new TopologyKey(HwvtepSouthboundConstants.HWVTEP_TOPOLOGY_ID)).child(Node.class)
 163                 .augmentation(PhysicalSwitchAugmentation.class);
 164     }
 165
 166     @Override
 167     protected HwvtepPhysicalSwitchListener getDataTreeChangeListener() {
 168         return HwvtepPhysicalSwitchListener.this;
 169     }
 170
 171     @Override
 172     protected void removed(InstanceIdentifier<PhysicalSwitchAugmentation> identifier,
 173             PhysicalSwitchAugmentation phySwitchDeleted) {
 174         NodeId nodeId = getNodeId(identifier);
 175         String psName = phySwitchDeleted.getHwvtepNodeName().getValue();
 176         LOG.info("Received physical switch {} removed event for node {}", psName, nodeId.getValue());
 177
 178         L2GatewayDevice l2GwDevice = l2GatewayCache.get(psName);
 179         if (l2GwDevice != null) {
 180             if (!L2GatewayConnectionUtils.isGatewayAssociatedToL2Device(l2GwDevice)) {
 181                 l2GatewayCache.remove(psName);
 182                 LOG.debug("{} details removed from L2Gateway Cache", psName);
 183                 MDSALUtil.syncDelete(this.dataBroker, LogicalDatastoreType.CONFIGURATION,
 184                         HwvtepSouthboundUtils.createInstanceIdentifier(nodeId));
 185             } else {
 186                 LOG.debug("{} details are not removed from L2Gateway Cache as it has L2Gateway reference", psName);
 187             }
 188
 189             l2GwDevice.setConnected(false);
 190             //ElanL2GwCacheUtils.removeL2GatewayDeviceFromAllElanCache(psName);
 191         } else {
 192             LOG.error("Unable to find L2 Gateway details for {}", psName);
 193         }
 194     }
 195
 196     /**
 197      * Upon update checks if the tunnels Ip was null earlier and it got newly added.
 198      * In that case simply call add.
 199      * If not then check if Tunnel Ip has been updated from an old value to new value.
 200      * If yes. delete old ITM tunnels of odl Tunnel Ipand add new ITM tunnels with new Tunnel
 201      * IP then call added ().
 202      *
 203      * @param identifier iid
 204      * @param phySwitchBefore ps Node before update
 205      * @param phySwitchAfter ps Node after update
 206      */
 207     @Override
 208     protected void updated(InstanceIdentifier<PhysicalSwitchAugmentation> identifier,
 209             PhysicalSwitchAugmentation phySwitchBefore, PhysicalSwitchAugmentation phySwitchAfter) {
 210         NodeId nodeId = getNodeId(identifier);
 211         LOG.trace("Received PhysicalSwitch Update Event for node {}: PhysicalSwitch Before: {}, "
 212                 + "PhysicalSwitch After: {}", nodeId.getValue(), phySwitchBefore, phySwitchAfter);
 213         String psName = getPsName(identifier);
 214         if (psName == null) {
 215             LOG.error("Could not find the physical switch name for node {}", nodeId.getValue());
 216             return;
 217         }
 218         L2GatewayDevice existingDevice = l2GatewayCache.get(psName);
 219         LOG.info("Received physical switch {} update event for node {}", psName, nodeId.getValue());
 220         InstanceIdentifier<Node> globalNodeIid = getManagedByNodeIid(identifier);
 221
 222         if (DEVICE_NOT_CACHED_OR_PARENT_CONNECTED.test(existingDevice, globalNodeIid)) {
 223             if (TUNNEL_IP_AVAILABLE.test(phySwitchAfter)) {
 224                 added(identifier, phySwitchAfter);
 225             }
 226         } else {
 227             if (!Objects.equals(phySwitchAfter.getTunnelIps(), phySwitchBefore.getTunnelIps())
 228                     && TUNNEL_IP_CHANGED.test(phySwitchAfter, existingDevice)) {
 229
 230                 final String hwvtepId = existingDevice.getHwvtepNodeId();
 231                 elanClusterUtils.runOnlyInOwnerNode(existingDevice.getDeviceName(),
 232                     "handling Physical Switch add create itm tunnels ",
 233                     () -> {
 234                         LOG.info("Deleting itm tunnels for device {}", existingDevice.getDeviceName());
 235                         L2GatewayUtils.deleteItmTunnels(itmRpcService, hwvtepId,
 236                                 existingDevice.getDeviceName(), existingDevice.getTunnelIp());
 237                         Thread.sleep(10000L);//TODO remove these sleeps
 238                         LOG.info("Creating itm tunnels for device {}", existingDevice.getDeviceName());
 239                         ElanL2GatewayUtils.createItmTunnels(dataBroker, itmRpcService, hwvtepId, psName,
 240                                 phySwitchAfter.getTunnelIps().get(0).getTunnelIpsKey());
 241                         return Collections.emptyList();
 242                     }
 243                 );
 244                 try {
 245                     Thread.sleep(20000L);//TODO remove the sleep by using better itm api to detect finish of prev op
 246                 } catch (InterruptedException e) {
 247                     LOG.error("Interrupted ");
 248                 }
 249                 existingDevice.setTunnelIps(new HashSet<>());
 250                 added(identifier, phySwitchAfter);
 251             }
 252         }
 253     }
 254
 255     @Override
 256     protected void added(InstanceIdentifier<PhysicalSwitchAugmentation> identifier,
 257                          final PhysicalSwitchAugmentation phySwitchAdded) {
 258         String globalNodeId = getManagedByNodeId(identifier);
 259         final InstanceIdentifier<Node> globalNodeIid = getManagedByNodeIid(identifier);
 260         NodeId nodeId = getNodeId(identifier);
 261         if (TUNNEL_IP_NOT_AVAILABLE.test(phySwitchAdded)) {
 262             LOG.error("Could not find the /tunnel ips for node {}", nodeId.getValue());
 263             return;
 264         }
 265         final String psName = getPsName(identifier);
 266         LOG.trace("Received physical switch {} added event received for node {}", psName, nodeId.getValue());
 267
 268         haOpClusteredListener.runAfterNodeIsConnected(globalNodeIid, (node) -> {
 269             LOG.trace("Running job for node {} ", globalNodeIid);
 270             if (!node.isPresent()) {
 271                 LOG.error("Global node is absent {}", globalNodeId);
 272                 return;
 273             }
 274             HwvtepHAUtil.addToCacheIfHAChildNode(globalNodeIid, node.get(), hwvtepNodeHACache);
 275             if (hwvtepNodeHACache.isHAEnabledDevice(globalNodeIid)) {
 276                 LOG.trace("Ha enabled device {}", globalNodeIid);
 277                 return;
 278             }
 279             LOG.trace("Updating cache for node {}", globalNodeIid);
 280             L2GatewayDevice l2GwDevice = l2GatewayCache.get(psName);
 281             if (childConnectedAfterParent.test(l2GwDevice, globalNodeIid)) {
 282                 LOG.trace("Device {} {} is already Connected by {}",
 283                         psName, globalNodeId, l2GwDevice.getHwvtepNodeId());
 284                 return;
 285             }
 286             InstanceIdentifier<Node> existingIid = globalNodeIid;
 287             if (l2GwDevice != null && l2GwDevice.getHwvtepNodeId() != null) {
 288                 existingIid = HwvtepHAUtil.convertToInstanceIdentifier(l2GwDevice.getHwvtepNodeId());
 289             }
 290             if (parentConnectedAfterChild.test(l2GwDevice, globalNodeIid)
 291                     && alreadyHasL2Gwids.test(l2GwDevice)) {
 292                 LOG.error("Child node {} having l2gw configured became ha node "
 293                                 + " removing the l2device {} from all elan cache and provision parent node {}",
 294                           existingIid, psName, globalNodeIid);
 295                 ElanL2GwCacheUtils.removeL2GatewayDeviceFromAllElanCache(l2GwDevice.getHwvtepNodeId());
 296             }
 297
 298             l2GwDevice = l2GatewayCache.addOrGet(psName);
 299             l2GwDevice.setConnected(true);
 300             l2GwDevice.setHwvtepNodeId(globalNodeId);
 301
 302             List<TunnelIps> tunnelIps = phySwitchAdded.getTunnelIps();
 303             if (tunnelIps != null) {
 304                 for (TunnelIps tunnelIp : tunnelIps) {
 305                     IpAddress tunnelIpAddr = tunnelIp.getTunnelIpsKey();
 306                     l2GwDevice.addTunnelIp(tunnelIpAddr);
 307                 }
 308             }
 309
 310             handleAdd(l2GwDevice);
 311             elanClusterUtils.runOnlyInOwnerNode("Update config tunnels IP ",
 312                 () -> updateConfigTunnelIp(identifier, phySwitchAdded));
 313         });
 314     }
 315
 316     /**
 317      * Handle add.
 318      *
 319      * @param l2GwDevice
 320      *            the l2 gw device
 321      */
 322     private void handleAdd(L2GatewayDevice l2GwDevice) {
 323         final String psName = l2GwDevice.getDeviceName();
 324         final String hwvtepNodeId = l2GwDevice.getHwvtepNodeId();
 325         Set<IpAddress> tunnelIps = l2GwDevice.getTunnelIps();
 326         for (final IpAddress tunnelIpAddr : tunnelIps) {
 327             if (L2GatewayConnectionUtils.isGatewayAssociatedToL2Device(l2GwDevice)) {
 328                 LOG.debug("L2Gateway {} associated for {} physical switch; creating ITM tunnels for {}",
 329                         l2GwDevice.getL2GatewayIds(), psName, tunnelIpAddr);
 330                 l2gwServiceProvider.provisionItmAndL2gwConnection(l2GwDevice, psName, hwvtepNodeId, tunnelIpAddr);
 331             } else {
 332                 LOG.info("l2gw.provision.skip {}:{}", hwvtepNodeId, psName);
 333             }
 334         }
 335         elanClusterUtils.runOnlyInOwnerNode("Stale entry cleanup", () -> {
 336             InstanceIdentifier<Node> globalNodeIid = HwvtepSouthboundUtils.createInstanceIdentifier(
 337                     new NodeId(hwvtepNodeId));
 338             InstanceIdentifier<Node> psIid = HwvtepSouthboundUtils.createInstanceIdentifier(
 339                     HwvtepSouthboundUtils.createManagedNodeId(new NodeId(hwvtepNodeId), psName));
 340             staleVlanBindingsCleaner.scheduleStaleCleanup(psName, globalNodeIid, psIid);
 341         });
 342     }
 343
 344
 345     /**
 346      * Gets the node id.
 347      *
 348      * @param identifier
 349      *            the identifier
 350      * @return the node id
 351      */
 352     private NodeId getNodeId(InstanceIdentifier<PhysicalSwitchAugmentation> identifier) {
 353         return identifier.firstKeyOf(Node.class).getNodeId();
 354     }
 355
 356     private String getManagedByNodeId(InstanceIdentifier<PhysicalSwitchAugmentation> identifier) {
 357         String psNodeId = identifier.firstKeyOf(Node.class).getNodeId().getValue();
 358         if (psNodeId.contains(HwvtepHAUtil.PHYSICALSWITCH)) {
 359             return psNodeId.substring(0, psNodeId.indexOf(HwvtepHAUtil.PHYSICALSWITCH));
 360         }
 361         return psNodeId;
 362     }
 363
 364     @Nullable
 365     private InstanceIdentifier<Node> getManagedByNodeIid(InstanceIdentifier<PhysicalSwitchAugmentation> identifier) {
 366         String psNodeId = identifier.firstKeyOf(Node.class).getNodeId().getValue();
 367         if (psNodeId.contains(HwvtepHAUtil.PHYSICALSWITCH)) {
 368             psNodeId = psNodeId.substring(0, psNodeId.indexOf(HwvtepHAUtil.PHYSICALSWITCH));
 369             return identifier.firstIdentifierOf(Topology.class).child(Node.class, new NodeKey(new NodeId(psNodeId)));
 370         }
 371         return null;
 372     }
 373
 374     @Nullable
 375     private String getPsName(InstanceIdentifier<PhysicalSwitchAugmentation> identifier) {
 376         String psNodeId = identifier.firstKeyOf(Node.class).getNodeId().getValue();
 377         if (psNodeId.contains(HwvtepHAUtil.PHYSICALSWITCH)) {
 378             return psNodeId.substring(psNodeId.indexOf(HwvtepHAUtil.PHYSICALSWITCH) + HwvtepHAUtil.PHYSICALSWITCH
 379                     .length());
 380         }
 381         return null;
 382     }
 383
 384     private void updateConfigTunnelIp(InstanceIdentifier<PhysicalSwitchAugmentation> identifier,
 385                                       PhysicalSwitchAugmentation phySwitchAdded) {
 386         if (phySwitchAdded.getTunnelIps() != null) {
 387             ListenableFutures.addErrorLogging(
 388                 txRunner.callWithNewReadWriteTransactionAndSubmit(CONFIGURATION, tx -> {
 389                     Optional<PhysicalSwitchAugmentation> existingSwitch = tx.read(identifier).get();
 390                     PhysicalSwitchAugmentationBuilder psBuilder = new PhysicalSwitchAugmentationBuilder();
 391                     if (existingSwitch.isPresent()) {
 392                         psBuilder = new PhysicalSwitchAugmentationBuilder(existingSwitch.get());
 393                     }
 394                     psBuilder.setTunnelIps(phySwitchAdded.getTunnelIps());
 395                     tx.put(identifier, psBuilder.build(), true);
 396                     LOG.trace("Updating config tunnel ips {}", identifier);
 397                 }), LOG, "Failed to update the config tunnel ips {}", identifier);
 398         }
 399     }
 400 }