elanmanager/impl/src/main/java/org/opendaylight/netvirt/elan/evpn/utils/EvpnUtils.java

   1 /*
   2  * Copyright © 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.netvirt.elan.evpn.utils;
   9
  10 import com.google.common.base.Optional;
  11 import com.google.common.util.concurrent.Futures;
  12 import com.google.common.util.concurrent.MoreExecutors;
  13 import com.google.common.util.concurrent.SettableFuture;
  14 import java.math.BigInteger;
  15 import java.util.ArrayList;
  16 import java.util.Collections;
  17 import java.util.List;
  18 import java.util.concurrent.ExecutionException;
  19 import java.util.concurrent.Future;
  20 import java.util.function.BiConsumer;
  21 import java.util.function.BiPredicate;
  22 import java.util.function.Function;
  23 import java.util.function.Predicate;
  24 import javax.inject.Inject;
  25 import javax.inject.Singleton;
  26 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  27 import org.opendaylight.controller.md.sal.binding.api.ReadWriteTransaction;
  28 import org.opendaylight.controller.md.sal.binding.api.WriteTransaction;
  29 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  30 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
  31 import org.opendaylight.genius.interfacemanager.globals.InterfaceInfo;
  32 import org.opendaylight.genius.interfacemanager.interfaces.IInterfaceManager;
  33 import org.opendaylight.genius.itm.globals.ITMConstants;
  34 import org.opendaylight.genius.mdsalutil.FlowEntity;
  35 import org.opendaylight.genius.mdsalutil.InstructionInfo;
  36 import org.opendaylight.genius.mdsalutil.MDSALUtil;
  37 import org.opendaylight.genius.mdsalutil.MatchInfo;
  38 import org.opendaylight.genius.mdsalutil.NWUtil;
  39 import org.opendaylight.genius.mdsalutil.NwConstants;
  40 import org.opendaylight.genius.mdsalutil.instructions.InstructionGotoTable;
  41 import org.opendaylight.genius.mdsalutil.instructions.InstructionWriteMetadata;
  42 import org.opendaylight.genius.mdsalutil.matches.MatchTunnelId;
  43 import org.opendaylight.genius.utils.ServiceIndex;
  44 import org.opendaylight.infrautils.jobcoordinator.JobCoordinator;
  45 import org.opendaylight.netvirt.bgpmanager.api.IBgpManager;
  46 import org.opendaylight.netvirt.elan.l2gw.utils.SettableFutureCallback;
  47 import org.opendaylight.netvirt.elan.utils.ElanConstants;
  48 import org.opendaylight.netvirt.elan.utils.ElanUtils;
  49 import org.opendaylight.netvirt.elanmanager.api.ElanHelper;
  50 import org.opendaylight.netvirt.vpnmanager.api.IVpnManager;
  51 import org.opendaylight.yang.gen.v1.urn.huawei.params.xml.ns.yang.l3vpn.rev140815.vpn.instances.VpnInstance;
  52 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.IpAddress;
  53 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.instruction.list.Instruction;
  54 import org.opendaylight.yang.gen.v1.urn.opendaylight.genius.interfacemanager.servicebinding.rev160406.service.bindings.services.info.BoundServices;
  55 import org.opendaylight.yang.gen.v1.urn.opendaylight.genius.itm.op.rev160406.ExternalTunnelList;
  56 import org.opendaylight.yang.gen.v1.urn.opendaylight.genius.itm.op.rev160406.external.tunnel.list.ExternalTunnel;
  57 import org.opendaylight.yang.gen.v1.urn.opendaylight.genius.itm.rev160406.DcGatewayIpList;
  58 import org.opendaylight.yang.gen.v1.urn.opendaylight.genius.itm.rev160406.dc.gateway.ip.list.DcGatewayIp;
  59 import org.opendaylight.yang.gen.v1.urn.opendaylight.genius.itm.rpcs.rev160406.GetDpnEndpointIpsInputBuilder;
  60 import org.opendaylight.yang.gen.v1.urn.opendaylight.genius.itm.rpcs.rev160406.GetDpnEndpointIpsOutput;
  61 import org.opendaylight.yang.gen.v1.urn.opendaylight.genius.itm.rpcs.rev160406.ItmRpcService;
  62 import org.opendaylight.yang.gen.v1.urn.opendaylight.netvirt.elan.rev150602.EvpnAugmentation;
  63 import org.opendaylight.yang.gen.v1.urn.opendaylight.netvirt.elan.rev150602.elan.instances.ElanInstance;
  64 import org.opendaylight.yang.gen.v1.urn.opendaylight.netvirt.elan.rev150602.forwarding.entries.MacEntry;
  65 import org.opendaylight.yang.gen.v1.urn.opendaylight.netvirt.fibmanager.rev150330.VrfEntryBase;
  66 import org.opendaylight.yang.gen.v1.urn.opendaylight.netvirt.neutronvpn.rev150602.neutron.vpn.portip.port.data.VpnPortipToPort;
  67 import org.opendaylight.yangtools.yang.binding.DataObject;
  68 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  69 import org.opendaylight.yangtools.yang.common.RpcResult;
  70 import org.slf4j.Logger;
  71 import org.slf4j.LoggerFactory;
  72
  73 @Singleton
  74 public class EvpnUtils {
  75
  76     private static final Logger LOG = LoggerFactory.getLogger(EvpnUtils.class);
  77
  78     private final BiPredicate<String, String> isNetAttach = (var1, var2) -> (var1 == null && var2 != null);
  79     private final BiPredicate<String, String> isNetDetach = (var1, var2) -> (var1 != null && var2 == null);
  80     private final Predicate<MacEntry> isIpv4PrefixAvailable = (macEntry) -> (macEntry != null
  81         && macEntry.getIpPrefix() != null && macEntry.getIpPrefix().getIpv4Address() != null);
  82     private final DataBroker broker;
  83     private final IInterfaceManager interfaceManager;
  84     private final ElanUtils elanUtils;
  85     private final ItmRpcService itmRpcService;
  86     private final JobCoordinator jobCoordinator;
  87     private final IBgpManager bgpManager;
  88     private final IVpnManager vpnManager;
  89
  90     @Inject
  91     public EvpnUtils(DataBroker broker, IInterfaceManager interfaceManager, ElanUtils elanUtils,
  92             ItmRpcService itmRpcService, IVpnManager vpnManager, IBgpManager bgpManager,
  93             JobCoordinator jobCoordinator) {
  94         this.broker = broker;
  95         this.interfaceManager = interfaceManager;
  96         this.elanUtils = elanUtils;
  97         this.itmRpcService = itmRpcService;
  98         this.vpnManager = vpnManager;
  99         this.bgpManager = bgpManager;
 100         this.jobCoordinator = jobCoordinator;
 101     }
 102
 103     public boolean isWithdrawEvpnRT2Routes(ElanInstance original, ElanInstance update) {
 104         return isNetDetach.test(getEvpnNameFromElan(original), getEvpnNameFromElan(update));
 105     }
 106
 107     public boolean isAdvertiseEvpnRT2Routes(ElanInstance original, ElanInstance update) {
 108         return isNetAttach.test(getEvpnNameFromElan(original), getEvpnNameFromElan(update));
 109     }
 110
 111     @SuppressWarnings("checkstyle:IllegalCatch")
 112     public void advertiseEvpnRT2Routes(EvpnAugmentation evpnAugmentation, String elanName)  {
 113         if (evpnAugmentation == null || evpnAugmentation.getEvpnName() == null) {
 114             return;
 115         }
 116         String evpnName = evpnAugmentation.getEvpnName();
 117         List<MacEntry> macEntries = elanUtils.getElanMacEntries(elanName);
 118         if (macEntries == null || macEntries.isEmpty()) {
 119             LOG.trace("advertiseEvpnRT2Routes no elan mac entries found for {}", elanName);
 120             return;
 121         }
 122         String rd = vpnManager.getVpnRd(broker, evpnName);
 123         ElanInstance elanInfo = ElanUtils.getElanInstanceByName(broker, elanName);
 124         macEntries.stream().filter(isIpv4PrefixAvailable).forEach(macEntry -> {
 125             InterfaceInfo interfaceInfo = interfaceManager.getInterfaceInfo(macEntry.getInterface());
 126             if (interfaceInfo == null) {
 127                 LOG.debug("advertiseEvpnRT2Routes, interfaceInfo is null for interface {}", macEntry.getInterface());
 128                 return;
 129             }
 130             advertisePrefix(elanInfo, rd, macEntry.getMacAddress().getValue(),
 131                     macEntry.getIpPrefix().getIpv4Address().getValue(),
 132                     interfaceInfo.getInterfaceName(), interfaceInfo.getDpId());
 133         });
 134     }
 135
 136     public String getEndpointIpAddressForDPN(BigInteger dpnId) {
 137
 138         Future<RpcResult<GetDpnEndpointIpsOutput>> result = itmRpcService.getDpnEndpointIps(
 139                 new GetDpnEndpointIpsInputBuilder()
 140                         .setSourceDpid(dpnId)
 141                         .build());
 142         RpcResult<GetDpnEndpointIpsOutput> rpcResult = null;
 143         try {
 144             rpcResult = result.get();
 145         } catch (InterruptedException e) {
 146             LOG.error("getnextHopIpFromRpcOutput : InterruptedException for dpnid {}", e, dpnId);
 147             return null;
 148         } catch (ExecutionException e) {
 149             LOG.error("getnextHopIpFromRpcOutput : ExecutionException for dpnid {}", e, dpnId);
 150             return null;
 151         }
 152         if (!rpcResult.isSuccessful()) {
 153             LOG.warn("RPC Call to getDpnEndpointIps returned with Errors {}", rpcResult.getErrors());
 154             return null;
 155         }
 156
 157         List<IpAddress> nexthopIpList = rpcResult.getResult().getNexthopipList();
 158         return nexthopIpList.get(0).getIpv4Address().getValue();
 159     }
 160
 161     public Optional<String> getGatewayMacAddressForInterface(String vpnName,
 162                                                                                     String ifName, String ipAddress) {
 163         VpnPortipToPort gwPort = vpnManager.getNeutronPortFromVpnPortFixedIp(broker, vpnName, ipAddress);
 164         return Optional.of(gwPort != null && gwPort.isSubnetIp()
 165                 ? gwPort.getMacAddress()
 166                 : interfaceManager.getInterfaceInfoFromOperationalDataStore(ifName).getMacAddress());
 167     }
 168
 169     public String getL3vpnNameFromElan(ElanInstance elanInfo) {
 170         if (elanInfo == null) {
 171             LOG.debug("getL3vpnNameFromElan :elanInfo is NULL");
 172             return null;
 173         }
 174         EvpnAugmentation evpnAugmentation = elanInfo.getAugmentation(EvpnAugmentation.class);
 175         return evpnAugmentation != null ? evpnAugmentation.getL3vpnName() : null;
 176     }
 177
 178     public static String getEvpnNameFromElan(ElanInstance elanInfo) {
 179         if (elanInfo == null) {
 180             LOG.debug("getEvpnNameFromElan :elanInfo is NULL");
 181             return null;
 182         }
 183         EvpnAugmentation evpnAugmentation = elanInfo.getAugmentation(EvpnAugmentation.class);
 184         return evpnAugmentation != null ? evpnAugmentation.getEvpnName() : null;
 185     }
 186
 187     public String getEvpnRd(ElanInstance elanInfo) {
 188         String evpnName = getEvpnNameFromElan(elanInfo);
 189         if (evpnName == null) {
 190             LOG.debug("getEvpnRd : evpnName is NULL for elanInfo {}", elanInfo);
 191             return null;
 192         }
 193         return vpnManager.getVpnRd(broker, evpnName);
 194     }
 195
 196     public void advertisePrefix(ElanInstance elanInfo, String macAddress, String prefix,
 197                                  String interfaceName, BigInteger dpnId) {
 198         String rd = getEvpnRd(elanInfo);
 199         advertisePrefix(elanInfo, rd, macAddress, prefix, interfaceName, dpnId);
 200     }
 201
 202     @SuppressWarnings("checkstyle:IllegalCatch")
 203     public void advertisePrefix(ElanInstance elanInfo, String rd,
 204                                  String macAddress, String prefix, String interfaceName, BigInteger dpnId) {
 205         if (rd == null) {
 206             LOG.debug("advertisePrefix : rd is NULL for elanInfo {}, macAddress {}", elanInfo, macAddress);
 207             return;
 208         }
 209         String nextHop = getEndpointIpAddressForDPN(dpnId);
 210         if (nextHop == null) {
 211             LOG.debug("Failed to get the dpn tep ip for dpn {}", dpnId);
 212             return;
 213         }
 214         int vpnLabel = 0;
 215         long l2vni = elanInfo.getSegmentationId();
 216         long l3vni = 0;
 217         String gatewayMacAddr = null;
 218         String l3VpName = getL3vpnNameFromElan(elanInfo);
 219         if (l3VpName != null) {
 220             VpnInstance l3VpnInstance = vpnManager.getVpnInstance(broker, l3VpName);
 221             l3vni = l3VpnInstance.getL3vni();
 222             com.google.common.base.Optional<String> gatewayMac = getGatewayMacAddressForInterface(l3VpName,
 223                     interfaceName, prefix);
 224             gatewayMacAddr = gatewayMac.isPresent() ? gatewayMac.get() : null;
 225
 226         }
 227         LOG.info("Advertising routes with rd {},  macAddress {}, prefix {}, nextHop {},"
 228                         + " vpnLabel {}, l3vni {}, l2vni {}, gatewayMac {}", rd, macAddress, prefix, nextHop,
 229                 vpnLabel, l3vni, l2vni, gatewayMacAddr);
 230         try {
 231             bgpManager.advertisePrefix(rd, macAddress, prefix, nextHop,
 232                     VrfEntryBase.EncapType.Vxlan, vpnLabel, l3vni, l2vni, gatewayMacAddr);
 233         } catch (Exception e) {
 234             LOG.error("Failed to advertisePrefix", e);
 235         }
 236     }
 237
 238     public void advertisePrefix(ElanInstance elanInfo, MacEntry macEntry) {
 239         InterfaceInfo interfaceInfo = interfaceManager.getInterfaceInfo(macEntry.getInterface());
 240         if (interfaceInfo == null) {
 241             LOG.debug("advertisePrefix, interfaceInfo is null for interface {}", macEntry.getInterface());
 242             return;
 243         }
 244
 245         if (!isIpv4PrefixAvailable.test(macEntry)) {
 246             LOG.debug("advertisePrefix macEntry does not have IPv4 prefix {}", macEntry);
 247             return;
 248         }
 249         advertisePrefix(elanInfo, macEntry.getMacAddress().getValue(),
 250                 macEntry.getIpPrefix().getIpv4Address().getValue(),
 251                 interfaceInfo.getInterfaceName(), interfaceInfo.getDpId());
 252     }
 253
 254     public void withdrawEvpnRT2Routes(EvpnAugmentation evpnAugmentation, String elanName) {
 255         if (evpnAugmentation == null || evpnAugmentation.getEvpnName() == null) {
 256             LOG.trace("withdrawEvpnRT2Routes, evpnAugmentation is null");
 257             return;
 258         }
 259
 260         String evpnName = evpnAugmentation.getEvpnName();
 261         String rd = vpnManager.getVpnRd(broker, evpnName);
 262         if (rd == null) {
 263             LOG.debug("withdrawEvpnRT2Routes : rd is null ", elanName);
 264             return;
 265         }
 266         List<MacEntry> macEntries = elanUtils.getElanMacEntries(elanName);
 267         if (macEntries == null || macEntries.isEmpty()) {
 268             LOG.debug("withdrawEvpnRT2Routes : macEntries  is empty for elan {} ", elanName);
 269             return;
 270         }
 271         for (MacEntry macEntry : macEntries) {
 272             if (!isIpv4PrefixAvailable.test(macEntry)) {
 273                 LOG.debug("withdrawEvpnRT2Routes macEntry does not have IPv4 prefix {}", macEntry);
 274                 continue;
 275             }
 276             String prefix = macEntry.getIpPrefix().getIpv4Address().getValue();
 277             LOG.info("Withdrawing routes with rd {}, prefix {}", rd, prefix);
 278             bgpManager.withdrawPrefix(rd, prefix);
 279         }
 280     }
 281
 282     public void withdrawPrefix(ElanInstance elanInfo, String prefix) {
 283         String rd = getEvpnRd(elanInfo);
 284         if (rd == null) {
 285             return;
 286         }
 287         bgpManager.withdrawPrefix(rd, prefix);
 288     }
 289
 290     public void withdrawPrefix(ElanInstance elanInfo, MacEntry macEntry) {
 291         if (!isIpv4PrefixAvailable.test(macEntry)) {
 292             LOG.debug("withdrawPrefix macEntry does not have IPv4 prefix {}", macEntry);
 293             return;
 294         }
 295         withdrawPrefix(elanInfo, macEntry.getIpPrefix().getIpv4Address().getValue());
 296     }
 297
 298     public static InstanceIdentifier<ExternalTunnelList> getExternaTunnelListIdentifier() {
 299         return InstanceIdentifier
 300                 .builder(ExternalTunnelList.class).build();
 301     }
 302
 303     public Optional<ExternalTunnelList> getExternalTunnelList() {
 304         InstanceIdentifier<ExternalTunnelList> externalTunnelListId = getExternaTunnelListIdentifier();
 305         ExternalTunnelList externalTunnelList = null;
 306         try {
 307             externalTunnelList = elanUtils.read2(LogicalDatastoreType.CONFIGURATION,
 308                     externalTunnelListId).orNull();
 309         } catch (ReadFailedException e) {
 310             LOG.error("getExternalTunnelList: unable to read ExternalTunnelList, exception ", e);
 311         }
 312         return Optional.fromNullable(externalTunnelList);
 313     }
 314
 315     public static InstanceIdentifier<DcGatewayIpList> getDcGatewayIpListIdentifier() {
 316         return InstanceIdentifier
 317                 .builder(DcGatewayIpList.class).build();
 318     }
 319
 320     public Optional<DcGatewayIpList> getDcGatewayIpList() {
 321         InstanceIdentifier<DcGatewayIpList> dcGatewayIpListInstanceIdentifier = getDcGatewayIpListIdentifier();
 322         DcGatewayIpList dcGatewayIpListConfig = null;
 323         try {
 324             dcGatewayIpListConfig = elanUtils.read2(LogicalDatastoreType.CONFIGURATION,
 325                     dcGatewayIpListInstanceIdentifier).orNull();
 326         } catch (ReadFailedException e) {
 327             LOG.error("getDcGatewayTunnelInterfaceNameList: unable to read DcGatewayTunnelList, exception ", e);
 328         }
 329         return Optional.fromNullable(dcGatewayIpListConfig);
 330     }
 331
 332     public List<String> getDcGatewayTunnelInterfaceNameList() {
 333         final List<String> tunnelInterfaceNameList = new ArrayList<>();
 334         Optional<DcGatewayIpList> dcGatewayIpListOptional = getDcGatewayIpList();
 335         if (!dcGatewayIpListOptional.isPresent()) {
 336             LOG.info("No DC gateways configured while programming the l2vni table.");
 337             return tunnelInterfaceNameList;
 338         }
 339         List<DcGatewayIp> dcGatewayIps = dcGatewayIpListOptional.get().getDcGatewayIp();
 340
 341         Optional<ExternalTunnelList> externalTunnelListOptional = getExternalTunnelList();
 342         if (!externalTunnelListOptional.isPresent()) {
 343             LOG.info("No External Tunnel Configured while programming the l2vni table.");
 344             return tunnelInterfaceNameList;
 345         }
 346         List<ExternalTunnel> externalTunnels = externalTunnelListOptional.get().getExternalTunnel();
 347
 348         dcGatewayIps
 349                 .forEach(dcIp -> externalTunnels
 350                 .stream()
 351                 .filter(externalTunnel -> externalTunnel.getDestinationDevice()
 352                         .contains(dcIp.getIpAddress().getIpv4Address().toString()))
 353                 .forEach(externalTunnel -> tunnelInterfaceNameList.add(externalTunnel.getTunnelInterfaceName())));
 354
 355         return tunnelInterfaceNameList;
 356     }
 357
 358     public void bindElanServiceToExternalTunnel(String elanName, String interfaceName) {
 359         WriteTransaction tx = broker.newWriteOnlyTransaction();
 360         int instructionKey = 0;
 361         LOG.trace("Binding external interface {} elan {}", interfaceName, elanName);
 362         List<Instruction> instructions = new ArrayList<>();
 363         instructions.add(MDSALUtil.buildAndGetGotoTableInstruction(
 364                 NwConstants.L2VNI_EXTERNAL_TUNNEL_DEMUX_TABLE, ++instructionKey));
 365         short elanServiceIndex = ServiceIndex.getIndex(NwConstants.ELAN_SERVICE_NAME, NwConstants.ELAN_SERVICE_INDEX);
 366         BoundServices serviceInfo = ElanUtils.getBoundServices(
 367                 ElanUtils.getElanServiceName(elanName, interfaceName), elanServiceIndex,
 368                 NwConstants.ELAN_SERVICE_INDEX, NwConstants.COOKIE_ELAN_INGRESS_TABLE, instructions);
 369         InstanceIdentifier<BoundServices> bindServiceId = ElanUtils.buildServiceId(interfaceName, elanServiceIndex);
 370         Optional<BoundServices> existingElanService = elanUtils.read(broker, LogicalDatastoreType.CONFIGURATION,
 371                 bindServiceId);
 372         if (!existingElanService.isPresent()) {
 373             tx.put(LogicalDatastoreType.CONFIGURATION, bindServiceId, serviceInfo, true);
 374             tx.submit();
 375         }
 376     }
 377
 378     public void unbindElanServiceFromExternalTunnel(String elanName, String interfaceName) {
 379         WriteTransaction tx = broker.newWriteOnlyTransaction();
 380         LOG.trace("UnBinding external interface {} elan {}", interfaceManager, elanName);
 381         short elanServiceIndex = ServiceIndex.getIndex(NwConstants.ELAN_SERVICE_NAME, NwConstants.ELAN_SERVICE_INDEX);
 382         InstanceIdentifier<BoundServices> bindServiceId = ElanUtils.buildServiceId(interfaceName, elanServiceIndex);
 383         Optional<BoundServices> existingElanService = elanUtils.read(broker, LogicalDatastoreType.CONFIGURATION,
 384                 bindServiceId);
 385         if (!existingElanService.isPresent()) {
 386             tx.delete(LogicalDatastoreType.CONFIGURATION, bindServiceId);
 387             tx.submit();
 388         }
 389     }
 390
 391     private List<InstructionInfo> getInstructionsForExtTunnelTable(Long elanTag) {
 392         List<InstructionInfo> mkInstructions = new ArrayList<>();
 393         mkInstructions.add(new InstructionWriteMetadata(ElanUtils.getElanMetadataLabel(elanTag, false),
 394                 ElanHelper.getElanMetadataMask()));
 395         mkInstructions.add(new InstructionGotoTable(NwConstants.ELAN_DMAC_TABLE));
 396         return mkInstructions;
 397     }
 398
 399     private String getFlowRef(long tableId, long elanTag, BigInteger dpnId) {
 400         return new StringBuilder().append(tableId).append(elanTag).append(dpnId).toString();
 401     }
 402
 403     private void programEvpnL2vniFlow(ElanInstance elanInfo, BiConsumer<BigInteger, FlowEntity> flowHandler) {
 404         long elanTag = elanInfo.getElanTag();
 405         List<MatchInfo> mkMatches = new ArrayList<>();
 406         mkMatches.add(new MatchTunnelId(BigInteger.valueOf(elanInfo.getSegmentationId())));
 407         NWUtil.getOperativeDPNs(broker).forEach(dpnId -> {
 408             LOG.debug("Updating tunnel flow to dpnid {}", dpnId);
 409             List<InstructionInfo> instructions = getInstructionsForExtTunnelTable(elanTag);
 410             String flowRef = getFlowRef(NwConstants.L2VNI_EXTERNAL_TUNNEL_DEMUX_TABLE, elanTag, dpnId);
 411             FlowEntity flowEntity = MDSALUtil.buildFlowEntity(
 412                     dpnId,
 413                     NwConstants.L2VNI_EXTERNAL_TUNNEL_DEMUX_TABLE,
 414                     flowRef,
 415                     5, // prio
 416                     elanInfo.getElanInstanceName(), // flowName
 417                     0, // idleTimeout
 418                     0, // hardTimeout
 419                     ITMConstants.COOKIE_ITM_EXTERNAL.add(BigInteger.valueOf(elanTag)),
 420                     mkMatches,
 421                     instructions);
 422             flowHandler.accept(dpnId, flowEntity);
 423         });
 424     }
 425
 426     public void programEvpnL2vniDemuxTable(String elanName, final BiConsumer<String, String> serviceHandler,
 427                                            BiConsumer<BigInteger, FlowEntity> flowHandler) {
 428         ElanInstance elanInfo = ElanUtils.getElanInstanceByName(broker, elanName);
 429         List<String> tunnelInterfaceNameList = getDcGatewayTunnelInterfaceNameList();
 430         if (tunnelInterfaceNameList.isEmpty()) {
 431             LOG.info("No DC gateways tunnels while programming l2vni table for elan {}.", elanName);
 432             return;
 433         }
 434
 435         tunnelInterfaceNameList.forEach(tunnelInterfaceName -> {
 436             serviceHandler.accept(elanName, tunnelInterfaceName);
 437         });
 438         programEvpnL2vniFlow(elanInfo, flowHandler);
 439     }
 440
 441     @SuppressWarnings({ "unchecked", "rawtypes" })
 442     public <T extends DataObject> void asyncReadAndExecute(final LogicalDatastoreType datastoreType,
 443             final InstanceIdentifier<T> iid, final String jobKey, final Function<Optional<T>, Void> function) {
 444         jobCoordinator.enqueueJob(jobKey, () -> {
 445             SettableFuture<Optional<T>> settableFuture = SettableFuture.create();
 446             List futures = Collections.singletonList(settableFuture);
 447
 448             ReadWriteTransaction tx = broker.newReadWriteTransaction();
 449
 450             Futures.addCallback(tx.read(datastoreType, iid), new SettableFutureCallback<Optional<T>>(settableFuture) {
 451                 @Override
 452                 public void onSuccess(Optional<T> data) {
 453                     function.apply(data);
 454                     super.onSuccess(data);
 455                 }
 456             }, MoreExecutors.directExecutor());
 457
 458             return futures;
 459         }, ElanConstants.JOB_MAX_RETRIES);
 460     }
 461 }