elanmanager/impl/src/main/java/org/opendaylight/netvirt/elan/l2gw/utils/L2GatewayConnectionUtils.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.utils;
  10
  11 import static java.util.Collections.emptyMap;
  12 import static org.opendaylight.netvirt.elan.utils.ElanUtils.isVxlanNetworkOrVxlanSegment;
  13
  14 import com.google.common.collect.Lists;
  15 import com.google.common.util.concurrent.FluentFuture;
  16 import com.google.common.util.concurrent.FutureCallback;
  17 import com.google.common.util.concurrent.Futures;
  18 import com.google.common.util.concurrent.MoreExecutors;
  19 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
  20 import java.util.ArrayList;
  21 import java.util.List;
  22 import java.util.Map;
  23 import java.util.Objects;
  24 import java.util.Optional;
  25 import java.util.Set;
  26 import java.util.concurrent.CopyOnWriteArrayList;
  27 import java.util.concurrent.ExecutionException;
  28 import javax.annotation.PreDestroy;
  29 import javax.inject.Inject;
  30 import javax.inject.Singleton;
  31 import org.eclipse.jdt.annotation.NonNull;
  32 import org.eclipse.jdt.annotation.Nullable;
  33 import org.opendaylight.genius.datastoreutils.SingleTransactionDataBroker;
  34 import org.opendaylight.genius.utils.hwvtep.HwvtepNodeHACache;
  35 import org.opendaylight.genius.utils.hwvtep.HwvtepSouthboundUtils;
  36 import org.opendaylight.infrautils.jobcoordinator.JobCoordinator;
  37 import org.opendaylight.mdsal.binding.api.DataBroker;
  38 import org.opendaylight.mdsal.binding.util.ManagedNewTransactionRunner;
  39 import org.opendaylight.mdsal.binding.util.RetryingManagedNewTransactionRunner;
  40 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  41 import org.opendaylight.netvirt.elan.cache.ElanInstanceCache;
  42 import org.opendaylight.netvirt.elan.l2gw.ha.listeners.HAOpClusteredListener;
  43 import org.opendaylight.netvirt.elan.l2gw.jobs.AssociateHwvtepToElanJob;
  44 import org.opendaylight.netvirt.elan.l2gw.jobs.DisAssociateHwvtepFromElanJob;
  45 import org.opendaylight.netvirt.elan.l2gw.listeners.LocalUcastMacListener;
  46 import org.opendaylight.netvirt.elan.l2gw.recovery.impl.L2GatewayServiceRecoveryHandler;
  47 import org.opendaylight.netvirt.elan.utils.ElanClusterUtils;
  48 import org.opendaylight.netvirt.elanmanager.utils.ElanL2GwCacheUtils;
  49 import org.opendaylight.netvirt.neutronvpn.api.l2gw.L2GatewayCache;
  50 import org.opendaylight.netvirt.neutronvpn.api.l2gw.L2GatewayDevice;
  51 import org.opendaylight.serviceutils.srm.ServiceRecoveryRegistry;
  52 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.yang.types.rev130715.Uuid;
  53 import org.opendaylight.yang.gen.v1.urn.opendaylight.netvirt.elan.rev150602.elan.instances.ElanInstance;
  54 import org.opendaylight.yang.gen.v1.urn.opendaylight.neutron.l2gateways.rev150712.l2gateway.attributes.Devices;
  55 import org.opendaylight.yang.gen.v1.urn.opendaylight.neutron.l2gateways.rev150712.l2gateway.attributes.DevicesKey;
  56 import org.opendaylight.yang.gen.v1.urn.opendaylight.neutron.l2gateways.rev150712.l2gateway.connections.attributes.L2gatewayConnections;
  57 import org.opendaylight.yang.gen.v1.urn.opendaylight.neutron.l2gateways.rev150712.l2gateway.connections.attributes.l2gatewayconnections.L2gatewayConnection;
  58 import org.opendaylight.yang.gen.v1.urn.opendaylight.neutron.l2gateways.rev150712.l2gateways.attributes.L2gateways;
  59 import org.opendaylight.yang.gen.v1.urn.opendaylight.neutron.l2gateways.rev150712.l2gateways.attributes.l2gateways.L2gateway;
  60 import org.opendaylight.yang.gen.v1.urn.opendaylight.neutron.l2gateways.rev150712.l2gateways.attributes.l2gateways.L2gatewayKey;
  61 import org.opendaylight.yang.gen.v1.urn.opendaylight.neutron.rev150712.Neutron;
  62 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.ovsdb.hwvtep.rev150901.HwvtepGlobalAugmentation;
  63 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.ovsdb.hwvtep.rev150901.hwvtep.global.attributes.LocalUcastMacs;
  64 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.ovsdb.hwvtep.rev150901.hwvtep.global.attributes.LocalUcastMacsKey;
  65 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.ovsdb.hwvtep.rev150901.hwvtep.global.attributes.LogicalSwitches;
  66 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NodeId;
  67 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
  68 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  69 import org.slf4j.Logger;
  70 import org.slf4j.LoggerFactory;
  71
  72 @Singleton
  73 public class L2GatewayConnectionUtils implements AutoCloseable {
  74     private static final Logger LOG = LoggerFactory.getLogger(L2GatewayConnectionUtils.class);
  75
  76     private final DataBroker broker;
  77     private final ElanL2GatewayUtils elanL2GatewayUtils;
  78     private final ElanClusterUtils elanClusterUtils;
  79     private final ElanL2GatewayMulticastUtils elanL2GatewayMulticastUtils;
  80     private final JobCoordinator jobCoordinator;
  81     private final L2GatewayCache l2GatewayCache;
  82     private final ElanInstanceCache elanInstanceCache;
  83     private final List<AutoCloseable> closeables = new CopyOnWriteArrayList<>();
  84     private final HwvtepNodeHACache hwvtepNodeHACache;
  85     private final HAOpClusteredListener haOpClusteredListener;
  86     private final ElanRefUtil elanRefUtil;
  87     private final L2GatewayServiceRecoveryHandler l2GatewayServiceRecoveryHandler;
  88     private final ServiceRecoveryRegistry serviceRecoveryRegistry;
  89     private final ManagedNewTransactionRunner txRunner;
  90
  91     @Inject
  92     public L2GatewayConnectionUtils(DataBroker dataBroker, ElanClusterUtils elanClusterUtils,
  93                                     ElanL2GatewayUtils elanL2GatewayUtils, JobCoordinator jobCoordinator,
  94                                     ElanL2GatewayMulticastUtils elanL2GatewayMulticastUtils,
  95                                     L2GatewayCache l2GatewayCache, HAOpClusteredListener haOpClusteredListener,
  96                                     ElanInstanceCache elanInstanceCache, HwvtepNodeHACache hwvtepNodeHACache,
  97                                     ElanRefUtil elanRefUtil,
  98                                     L2GatewayServiceRecoveryHandler l2GatewayServiceRecoveryHandler,
  99                                     ServiceRecoveryRegistry serviceRecoveryRegistry) {
 100         this.broker = dataBroker;
 101         this.elanL2GatewayUtils = elanL2GatewayUtils;
 102         this.elanClusterUtils = elanClusterUtils;
 103         this.elanL2GatewayMulticastUtils = elanL2GatewayMulticastUtils;
 104         this.jobCoordinator = jobCoordinator;
 105         this.l2GatewayCache = l2GatewayCache;
 106         this.haOpClusteredListener = haOpClusteredListener;
 107         this.elanInstanceCache = elanInstanceCache;
 108         this.hwvtepNodeHACache = hwvtepNodeHACache;
 109         this.elanRefUtil = elanRefUtil;
 110         this.l2GatewayServiceRecoveryHandler = l2GatewayServiceRecoveryHandler;
 111         this.serviceRecoveryRegistry = serviceRecoveryRegistry;
 112         this.txRunner = new RetryingManagedNewTransactionRunner(dataBroker);
 113     }
 114
 115     @Override
 116     @PreDestroy
 117     @SuppressWarnings("checkstyle:IllegalCatch")
 118     public void close() {
 119         closeables.forEach(c -> {
 120             try {
 121                 c.close();
 122             } catch (Exception e) {
 123                 LOG.warn("Error closing {}", c, e);
 124             }
 125         });
 126     }
 127
 128     public static boolean isGatewayAssociatedToL2Device(L2GatewayDevice l2GwDevice) {
 129         return !l2GwDevice.getL2GatewayIds().isEmpty();
 130     }
 131
 132     @Nullable
 133     public static L2gateway getNeutronL2gateway(DataBroker broker, Uuid l2GatewayId) {
 134         LOG.debug("getNeutronL2gateway for {}", l2GatewayId.getValue());
 135         InstanceIdentifier<L2gateway> inst = InstanceIdentifier.create(Neutron.class).child(L2gateways.class)
 136                 .child(L2gateway.class, new L2gatewayKey(l2GatewayId));
 137         try {
 138             return SingleTransactionDataBroker.syncReadOptional(broker, LogicalDatastoreType.CONFIGURATION, inst)
 139                     .orElse(null);
 140         } catch (ExecutionException | InterruptedException e) {
 141             LOG.error("getNeutronL2gateway: Exception while reading L2gateway DS for the ID {}", l2GatewayId, e);
 142         }
 143         return null;
 144     }
 145
 146     @NonNull
 147     public static List<L2gateway> getL2gatewayList(DataBroker broker) {
 148         InstanceIdentifier<L2gateways> inst = InstanceIdentifier.create(Neutron.class).child(L2gateways.class);
 149         try {
 150             return new ArrayList<>((SingleTransactionDataBroker.syncReadOptional(broker,
 151                     LogicalDatastoreType.CONFIGURATION, inst).map(L2gateways::getL2gateway)
 152                     .orElse(emptyMap())).values());
 153         } catch (ExecutionException | InterruptedException e) {
 154             LOG.error("getNeutronL2gateway: Exception while reading L2gateway DS", e);
 155         }
 156         return null;
 157     }
 158
 159     @NonNull
 160     public static List<L2gatewayConnection> getAllL2gatewayConnections(DataBroker broker) {
 161         InstanceIdentifier<L2gatewayConnections> inst = InstanceIdentifier.create(Neutron.class)
 162                 .child(L2gatewayConnections.class);
 163         try {
 164             return new ArrayList<>((SingleTransactionDataBroker.syncReadOptional(broker,
 165                     LogicalDatastoreType.CONFIGURATION, inst).map(L2gatewayConnections::getL2gatewayConnection)
 166                     .orElse(emptyMap())).values());
 167         } catch (ExecutionException | InterruptedException e) {
 168             LOG.error("getNeutronL2gateway: Exception while reading L2gateway DS", e);
 169         }
 170         return null;
 171     }
 172
 173     /**
 174      * Gets the associated l2 gw connections.
 175      *
 176      * @param broker
 177      *            the broker
 178      * @param l2GatewayIds
 179      *            the l2 gateway ids
 180      * @return the associated l2 gw connections
 181      */
 182     @NonNull
 183     public static List<L2gatewayConnection> getAssociatedL2GwConnections(DataBroker broker, Set<Uuid> l2GatewayIds) {
 184         List<L2gatewayConnection> allL2GwConns = getAllL2gatewayConnections(broker);
 185         List<L2gatewayConnection> l2GwConnections = new ArrayList<>();
 186         for (Uuid l2GatewayId : l2GatewayIds) {
 187             for (L2gatewayConnection l2GwConn : allL2GwConns) {
 188                 if (Objects.equals(l2GwConn.getL2gatewayId(), l2GatewayId)) {
 189                     l2GwConnections.add(l2GwConn);
 190                 }
 191             }
 192         }
 193         return l2GwConnections;
 194     }
 195
 196     /**
 197      * Gets the associated l2 gw connections.
 198      *
 199      * @param broker
 200      *            the broker
 201      * @param elanName
 202      *            the elan Name
 203      * @return the associated l2 gw connection with elan
 204      */
 205     @NonNull
 206     public static List<L2gatewayConnection> getL2GwConnectionsByElanName(DataBroker broker, String elanName) {
 207         List<L2gatewayConnection> allL2GwConns = getAllL2gatewayConnections(broker);
 208         List<L2gatewayConnection> elanL2GateWayConnections = new ArrayList<>();
 209         for (L2gatewayConnection l2GwConn : allL2GwConns) {
 210             if (l2GwConn.getNetworkId().getValue().equalsIgnoreCase(elanName)) {
 211                 elanL2GateWayConnections.add(l2GwConn);
 212             }
 213         }
 214         return elanL2GateWayConnections;
 215     }
 216
 217     public void addL2GatewayConnection(L2gatewayConnection input) {
 218         addL2GatewayConnection(input, null/*deviceName*/, null);
 219     }
 220
 221     public void addL2GatewayConnection(final L2gatewayConnection input,
 222                                        final String l2GwDeviceName) {
 223         addL2GatewayConnection(input, l2GwDeviceName, null);
 224     }
 225
 226     public void addL2GatewayConnection(final L2gatewayConnection input,
 227                                        @Nullable final String l2GwDeviceName ,
 228                                        @Nullable L2gateway l2Gateway) {
 229         LOG.info("Adding L2gateway Connection with ID: {}", input.key().getUuid());
 230
 231         Uuid networkUuid = input.getNetworkId();
 232
 233         // Taking cluster reboot scenario , if Elan instance is not available when l2GatewayConnection add events
 234         // comes we need to wait for elaninstance to resolve. Hence updating the map with the runnable .
 235         // When elanInstance add comes , it look in to the map and run the associated runnable associated with it.
 236         ElanInstance elanInstance = elanInstanceCache.get(networkUuid.getValue(),
 237             () -> addL2GatewayConnection(input, l2GwDeviceName)).orElse(null);
 238         if (elanInstance == null) {
 239             return;
 240         }
 241
 242         if (!isVxlanNetworkOrVxlanSegment(elanInstance)) {
 243             LOG.error("Neutron network with id {} is not VxlanNetwork", networkUuid.getValue());
 244         } else {
 245             Uuid l2GatewayId = input.getL2gatewayId();
 246             if (l2Gateway == null) {
 247                 l2Gateway = getNeutronL2gateway(broker, l2GatewayId);
 248             }
 249             if (l2Gateway == null) {
 250                 LOG.error("L2Gateway with id {} is not present", l2GatewayId.getValue());
 251             } else {
 252                 associateHwvtepsToElan(elanInstance, l2Gateway, input, l2GwDeviceName);
 253             }
 254         }
 255     }
 256
 257     public void deleteL2GatewayConnection(L2gatewayConnection input) {
 258         LOG.info("Deleting L2gateway Connection with ID: {}", input.key().getUuid());
 259
 260         Uuid networkUuid = input.getNetworkId();
 261         String elanName = networkUuid.getValue();
 262         disAssociateHwvtepsFromElan(elanName, input);
 263     }
 264
 265     private void disAssociateHwvtepsFromElan(String elanName, L2gatewayConnection input) {
 266         Integer defaultVlan = input.getSegmentId();
 267         List<L2GatewayDevice> l2Devices = ElanL2GwCacheUtils.getAllElanDevicesFromCache();
 268         List<Devices> l2gwDevicesToBeDeleted = new ArrayList<>();
 269         for (L2GatewayDevice elanL2gwDevice : l2Devices) {
 270             if (elanL2gwDevice.getL2GatewayIds().contains(input.key().getUuid())) {
 271                 l2gwDevicesToBeDeleted.addAll(elanL2gwDevice.getDevicesForL2gwConnectionId(input.key().getUuid()));
 272             }
 273         }
 274         if (l2gwDevicesToBeDeleted.isEmpty()) {
 275             //delete logical switch
 276             Uuid l2GatewayId = input.getL2gatewayId();
 277             L2gateway l2Gateway = L2GatewayConnectionUtils.getNeutronL2gateway(broker, l2GatewayId);
 278             if (l2Gateway == null) {
 279                 LOG.error("Failed to find the l2gateway for the connection {}", input.getUuid());
 280                 return;
 281             } else if (l2Gateway.getDevices() != null) {
 282                 l2gwDevicesToBeDeleted.addAll(l2Gateway.getDevices().values());
 283             }
 284         }
 285         for (Devices l2Device : l2gwDevicesToBeDeleted) {
 286             String l2DeviceName = l2Device.getDeviceName();
 287             L2GatewayDevice l2GatewayDevice = l2GatewayCache.get(l2DeviceName);
 288             String hwvtepNodeId = l2GatewayDevice.getHwvtepNodeId();
 289             boolean isLastL2GwConnDeleted = false;
 290             L2GatewayDevice elanL2GwDevice = ElanL2GwCacheUtils.getL2GatewayDeviceFromCache(elanName, hwvtepNodeId);
 291             if (elanL2GwDevice != null && isLastL2GwConnBeingDeleted(elanL2GwDevice)) {
 292                 // Delete L2 Gateway device from 'ElanL2GwDevice' cache
 293                 LOG.debug("Elan L2Gw Conn cache removed for id {}", hwvtepNodeId);
 294                 ElanL2GwCacheUtils.removeL2GatewayDeviceFromCache(elanName, hwvtepNodeId);
 295                 isLastL2GwConnDeleted = true;
 296             } else {
 297                 Uuid l2GwConnId = input.key().getUuid();
 298                 LOG.debug("Elan L2Gw Conn cache with id {} is being referred by other L2Gw Conns; so only "
 299                         + "L2 Gw Conn {} reference is removed", hwvtepNodeId, l2GwConnId);
 300                 if (elanL2GwDevice != null) {
 301                     elanL2GwDevice.removeL2GatewayId(l2GwConnId);
 302                 } else {
 303                     isLastL2GwConnDeleted = true;
 304                 }
 305             }
 306
 307             DisAssociateHwvtepFromElanJob disAssociateHwvtepToElanJob =
 308                     new DisAssociateHwvtepFromElanJob(elanL2GatewayUtils, elanL2GatewayMulticastUtils,
 309                             elanL2GwDevice, elanName,
 310                         l2Device, defaultVlan, hwvtepNodeId, isLastL2GwConnDeleted);
 311             elanClusterUtils.runOnlyInOwnerNode(disAssociateHwvtepToElanJob.getJobKey(), "remove l2gw connection job",
 312                     disAssociateHwvtepToElanJob);
 313         }
 314     }
 315
 316     private void associateHwvtepsToElan(ElanInstance elanInstance,
 317             L2gateway l2Gateway, L2gatewayConnection input, @Nullable String l2GwDeviceName) {
 318         String elanName = elanInstance.getElanInstanceName();
 319         Integer defaultVlan = input.getSegmentId();
 320         Uuid l2GwConnId = input.key().getUuid();
 321         Map<DevicesKey, Devices> l2Devices = l2Gateway.getDevices();
 322
 323         LOG.trace("Associating ELAN {} with L2Gw Conn Id {} having below L2Gw devices {}", elanName, l2GwConnId,
 324                 l2Devices);
 325
 326         if (l2Devices == null) {
 327             return;
 328         }
 329
 330         for (Devices l2Device : l2Devices.values()) {
 331             String l2DeviceName = l2Device.getDeviceName();
 332             // L2gateway can have more than one L2 Gw devices. Configure Logical Switch, VLAN mappings,...
 333             // only on the switch which has come up just now and exclude all other devices from
 334             // preprovisioning/re-provisioning
 335             if (l2GwDeviceName != null && !l2GwDeviceName.equals(l2DeviceName)) {
 336                 LOG.debug("Associating Hwvtep to ELAN is not been processed for {}; as only {} got connected now!",
 337                         l2DeviceName, l2GwDeviceName);
 338                 continue;
 339             }
 340             L2GatewayDevice l2GatewayDevice = l2GatewayCache.get(l2DeviceName);
 341             if (isL2GwDeviceConnected(l2GatewayDevice)) {
 342                 NodeId hwvtepNodeId = new NodeId(l2GatewayDevice.getHwvtepNodeId());
 343
 344                 // Delete pending delete logical switch task if scheduled
 345                 elanL2GatewayUtils.cancelDeleteLogicalSwitch(hwvtepNodeId,
 346                         ElanL2GatewayUtils.getLogicalSwitchFromElan(elanName));
 347
 348                 // Add L2 Gateway device to 'ElanL2GwDevice' cache
 349                 boolean createLogicalSwitch;
 350                 addL2DeviceToElanL2GwCache(elanName, l2GatewayDevice, l2GwConnId, l2Device);
 351
 352                 AssociateHwvtepToElanJob associateHwvtepToElanJob = new AssociateHwvtepToElanJob(broker,
 353                         elanL2GatewayUtils, elanL2GatewayMulticastUtils, elanInstanceCache, l2GatewayDevice,
 354                         elanInstance, l2Device, defaultVlan, elanRefUtil);
 355
 356                 elanClusterUtils.runOnlyInOwnerNode(associateHwvtepToElanJob.getJobKey(),
 357                         "create logical switch in hwvtep topo", associateHwvtepToElanJob);
 358
 359             } else {
 360                 LOG.info("L2GwConn create is not handled for device with id {} as it's not connected", l2DeviceName);
 361             }
 362         }
 363     }
 364
 365     public L2GatewayDevice addL2DeviceToElanL2GwCache(String elanName, L2GatewayDevice l2GatewayDevice, Uuid l2GwConnId,
 366             Devices l2Device) {
 367         String l2gwDeviceNodeId = l2GatewayDevice.getHwvtepNodeId();
 368         L2GatewayDevice elanL2GwDevice = ElanL2GwCacheUtils.getL2GatewayDeviceFromCache(elanName, l2gwDeviceNodeId);
 369         if (elanL2GwDevice == null) {
 370             elanL2GwDevice = new L2GatewayDevice(l2GatewayDevice.getDeviceName());
 371             elanL2GwDevice.setHwvtepNodeId(l2gwDeviceNodeId);
 372             elanL2GwDevice.setTunnelIps(l2GatewayDevice.getTunnelIps());
 373             ElanL2GwCacheUtils.addL2GatewayDeviceToCache(elanName, elanL2GwDevice);
 374             LOG.debug("Elan L2GwConn cache created for hwvtep id {}", l2gwDeviceNodeId);
 375         } else {
 376             LOG.debug("Elan L2GwConn cache already exists for hwvtep id {}; updating L2GwConn id {} to it",
 377                     l2gwDeviceNodeId, l2GwConnId);
 378         }
 379         elanL2GwDevice.addL2GatewayId(l2GwConnId);
 380         elanL2GwDevice.addL2gwConnectionIdToDevice(l2GwConnId, l2Device);
 381
 382         //incase of cluster reboot scenario southbound device would have added more macs
 383         //while odl is down, pull them now
 384         readAndCopyLocalUcastMacsToCache(elanName, l2GatewayDevice);
 385
 386         LOG.trace("Elan L2GwConn cache updated with below details: {}", elanL2GwDevice);
 387         return elanL2GwDevice;
 388     }
 389
 390     private static boolean isL2GwDeviceConnected(L2GatewayDevice l2GwDevice) {
 391         return l2GwDevice != null && l2GwDevice.getHwvtepNodeId() != null;
 392     }
 393
 394     protected static boolean isLastL2GwConnBeingDeleted(@NonNull L2GatewayDevice l2GwDevice) {
 395         return l2GwDevice.getL2GatewayIds().size() == 1;
 396     }
 397
 398     private void readAndCopyLocalUcastMacsToCache(final String elanName, final L2GatewayDevice l2GatewayDevice) {
 399         final InstanceIdentifier<Node> nodeIid = HwvtepSouthboundUtils.createInstanceIdentifier(
 400                 new NodeId(l2GatewayDevice.getHwvtepNodeId()));
 401         jobCoordinator.enqueueJob(elanName + ":" + l2GatewayDevice.getDeviceName(), () -> {
 402             FluentFuture<Optional<Node>> fluentFuture = broker.newReadOnlyTransaction().read(
 403                     LogicalDatastoreType.OPERATIONAL, nodeIid);
 404             Futures.addCallback(fluentFuture, new FutureCallback<Optional<Node>>() {
 405                 @Override
 406                 public void onSuccess(Optional<Node> nodeOptional) {
 407                     if (nodeOptional.isPresent()) {
 408                         Node node = nodeOptional.get();
 409                         if (node.augmentation(HwvtepGlobalAugmentation.class) != null) {
 410                             Map<LocalUcastMacsKey, LocalUcastMacs> localUcastMacs =
 411                                     node.augmentation(HwvtepGlobalAugmentation.class).getLocalUcastMacs();
 412                             if (localUcastMacs == null) {
 413                                 return;
 414                             }
 415                             LocalUcastMacListener localUcastMacListener =
 416                                     new LocalUcastMacListener(broker, haOpClusteredListener,
 417                                             elanL2GatewayUtils, jobCoordinator, elanInstanceCache, hwvtepNodeHACache,
 418                                             l2GatewayServiceRecoveryHandler, serviceRecoveryRegistry);
 419                             localUcastMacs.values().stream()
 420                                     .filter((mac) -> macBelongsToLogicalSwitch(mac, elanName))
 421                                     .forEach((mac) -> {
 422                                         InstanceIdentifier<LocalUcastMacs> macIid = getMacIid(nodeIid, mac);
 423                                         localUcastMacListener.added(macIid, mac);
 424                                     });
 425                         }
 426                     }
 427                 }
 428
 429                 @Override
 430                 public void onFailure(Throwable throwable) {
 431                 }
 432             }, MoreExecutors.directExecutor());
 433             return Lists.newArrayList(fluentFuture);
 434         } , 5);
 435     }
 436
 437     /**
 438      * Gets the associated l2 gw connections.
 439      *
 440      * @param l2GatewayId the l2 gateway id
 441      *
 442      * @return the associated l2 gw connections
 443      */
 444     public List<L2gatewayConnection> getL2GwConnectionsByL2GatewayId(Uuid l2GatewayId) {
 445         List<L2gatewayConnection> l2GwConnections = new ArrayList<>();
 446         List<L2gatewayConnection> allL2GwConns = getAllL2gatewayConnections(broker);
 447         for (L2gatewayConnection l2GwConn : allL2GwConns) {
 448             if (Objects.equals(l2GwConn.getL2gatewayId(), l2GatewayId)) {
 449                 l2GwConnections.add(l2GwConn);
 450             }
 451         }
 452         return l2GwConnections;
 453     }
 454
 455     @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD",
 456             justification = "https://github.com/spotbugs/spotbugs/issues/811")
 457     private static boolean macBelongsToLogicalSwitch(LocalUcastMacs mac, String logicalSwitchName) {
 458         InstanceIdentifier<LogicalSwitches> iid = (InstanceIdentifier<LogicalSwitches>)
 459                 mac.getLogicalSwitchRef().getValue();
 460         return iid.firstKeyOf(LogicalSwitches.class).getHwvtepNodeName().getValue().equals(logicalSwitchName);
 461     }
 462
 463     static InstanceIdentifier<LocalUcastMacs> getMacIid(InstanceIdentifier<Node> nodeIid, LocalUcastMacs mac) {
 464         return nodeIid.augmentation(HwvtepGlobalAugmentation.class).child(LocalUcastMacs.class, mac.key());
 465     }
 466 }