NETVIRT-1630 migrate to md-sal APIs
[netvirt.git] / neutronvpn / impl / src / main / java / org / opendaylight / netvirt / neutronvpn / NeutronPortChangeListener.java
index c152bbb0e046e2ac436eecf0e8ae0b6338476d34..5f4f768338c53dcc3c6d6df101fc187db3938d2b 100644 (file)
@@ -9,8 +9,8 @@ package org.opendaylight.netvirt.neutronvpn;
 
 import static org.opendaylight.genius.infra.Datastore.CONFIGURATION;
 
-import com.google.common.base.Optional;
 import com.google.common.base.Strings;
+import com.google.common.util.concurrent.ListenableFuture;
 import com.google.gson.Gson;
 import com.google.gson.JsonArray;
 import com.google.gson.JsonElement;
@@ -23,16 +23,15 @@ import java.util.HashSet;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.ExecutionException;
 import java.util.stream.Collectors;
-import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import javax.inject.Inject;
 import javax.inject.Singleton;
 import org.apache.commons.lang3.ObjectUtils;
 import org.eclipse.jdt.annotation.Nullable;
-import org.opendaylight.controller.md.sal.binding.api.DataBroker;
-import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
-import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
-import org.opendaylight.genius.datastoreutils.AsyncDataTreeChangeListenerBase;
 import org.opendaylight.genius.datastoreutils.SingleTransactionDataBroker;
 import org.opendaylight.genius.datastoreutils.listeners.DataTreeEventCallbackRegistrar;
 import org.opendaylight.genius.infra.Datastore;
@@ -41,11 +40,16 @@ import org.opendaylight.genius.infra.ManagedNewTransactionRunnerImpl;
 import org.opendaylight.genius.infra.TypedWriteTransaction;
 import org.opendaylight.genius.mdsalutil.MDSALUtil;
 import org.opendaylight.infrautils.jobcoordinator.JobCoordinator;
+import org.opendaylight.infrautils.utils.concurrent.Executors;
 import org.opendaylight.infrautils.utils.concurrent.ListenableFutures;
+import org.opendaylight.mdsal.binding.api.DataBroker;
+import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
+import org.opendaylight.mdsal.common.api.ReadFailedException;
 import org.opendaylight.netvirt.elanmanager.api.IElanService;
 import org.opendaylight.netvirt.neutronvpn.api.enums.IpVersionChoice;
 import org.opendaylight.netvirt.neutronvpn.api.utils.NeutronConstants;
 import org.opendaylight.netvirt.neutronvpn.api.utils.NeutronUtils;
+import org.opendaylight.serviceutils.tools.listener.AbstractAsyncDataTreeChangeListener;
 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.iana._if.type.rev170119.L2vlan;
 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.interfaces.rev140508.interfaces.Interface;
 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.interfaces.rev140508.interfaces.InterfaceBuilder;
@@ -85,7 +89,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 @Singleton
-public class NeutronPortChangeListener extends AsyncDataTreeChangeListenerBase<Port, NeutronPortChangeListener> {
+public class NeutronPortChangeListener extends AbstractAsyncDataTreeChangeListener<Port> {
     private static final Logger LOG = LoggerFactory.getLogger(NeutronPortChangeListener.class);
     private final DataBroker dataBroker;
     private final ManagedNewTransactionRunner txRunner;
@@ -99,6 +103,7 @@ public class NeutronPortChangeListener extends AsyncDataTreeChangeListenerBase<P
     private final DataTreeEventCallbackRegistrar eventCallbacks;
     private final NeutronvpnConfig neutronvpnConfig;
 
+    @Inject
     public NeutronPortChangeListener(final DataBroker dataBroker,
                                      final NeutronvpnManager neutronvpnManager,
                                      final NeutronvpnNatManager neutronvpnNatManager,
@@ -109,7 +114,9 @@ public class NeutronPortChangeListener extends AsyncDataTreeChangeListenerBase<P
                                      final HostConfigCache hostConfigCache,
                                      final DataTreeEventCallbackRegistrar dataTreeEventCallbackRegistrar,
                                      final NeutronvpnConfig neutronvpnConfig) {
-        super(Port.class, NeutronPortChangeListener.class);
+        super(dataBroker, LogicalDatastoreType.CONFIGURATION, InstanceIdentifier.create(Neutron.class)
+                .child(Ports.class).child(Port.class),
+                Executors.newSingleThreadExecutor("NeutronPortChangeListener", LOG));
         this.dataBroker = dataBroker;
         this.txRunner = new ManagedNewTransactionRunnerImpl(dataBroker);
         nvpnManager = neutronvpnManager;
@@ -124,26 +131,19 @@ public class NeutronPortChangeListener extends AsyncDataTreeChangeListenerBase<P
 
     }
 
-    @Override
-    @PostConstruct
     public void init() {
         LOG.info("{} init", getClass().getSimpleName());
-        registerListener(LogicalDatastoreType.CONFIGURATION, dataBroker);
     }
 
     @Override
-    protected InstanceIdentifier<Port> getWildCardPath() {
-        return InstanceIdentifier.create(Neutron.class).child(Ports.class).child(Port.class);
+    @PreDestroy
+    public void close() {
+        super.close();
+        Executors.shutdownAndAwaitTermination(getExecutorService());
     }
 
     @Override
-    protected NeutronPortChangeListener getDataTreeChangeListener() {
-        return NeutronPortChangeListener.this;
-    }
-
-
-    @Override
-    protected void add(InstanceIdentifier<Port> identifier, Port input) {
+    public void add(InstanceIdentifier<Port> identifier, Port input) {
         LOG.trace("Received port add event: port={}", input);
         String portName = input.getUuid().getValue();
         LOG.trace("Adding Port : key: {}, value={}", identifier, input);
@@ -182,7 +182,7 @@ public class NeutronPortChangeListener extends AsyncDataTreeChangeListenerBase<P
     }
 
     @Override
-    protected void remove(InstanceIdentifier<Port> identifier, Port input) {
+    public void remove(InstanceIdentifier<Port> identifier, Port input) {
         LOG.trace("Removing Port : key: {}, value={}", identifier, input);
         Network network = neutronvpnUtils.getNeutronNetwork(input.getNetworkId());
         // need to proceed with deletion in case network is null for a case where v2 sync happens and a read for
@@ -214,7 +214,7 @@ public class NeutronPortChangeListener extends AsyncDataTreeChangeListenerBase<P
     }
 
     @Override
-    protected void update(InstanceIdentifier<Port> identifier, Port original, Port update) {
+    public void update(InstanceIdentifier<Port> identifier, Port original, Port update) {
         LOG.trace("Received port update event: original={}, update={}", original, update);
         // Switchdev ports need to be bounded to a host before creation
         // in order to validate the supported vnic types from the hostconfig
@@ -225,7 +225,6 @@ public class NeutronPortChangeListener extends AsyncDataTreeChangeListenerBase<P
         }
         final String portName = update.getUuid().getValue();
         Network network = neutronvpnUtils.getNeutronNetwork(update.getNetworkId());
-        LOG.info("Update port {} from network {}", portName, update.getNetworkId().toString());
         if (network == null || !NeutronvpnUtils.isNetworkTypeSupported(network)) {
             LOG.warn("neutron vpn received a port update() for a network without a provider extension augmentation "
                     + "or with an unsupported network type for the port {} which is part of network {}",
@@ -261,11 +260,10 @@ public class NeutronPortChangeListener extends AsyncDataTreeChangeListenerBase<P
                                && NeutronvpnUtils.isDhcpServerPort(update);
         if (origSecurityEnabled || updatedSecurityEnabled || isDhcpServerPort) {
             InstanceIdentifier<Interface>  interfaceIdentifier = NeutronvpnUtils.buildVlanInterfaceIdentifier(portName);
-            jobCoordinator.enqueueJob("PORT- " + portName,
-                () -> Collections.singletonList(txRunner.callWithNewReadWriteTransactionAndSubmit(CONFIGURATION,
+            jobCoordinator.enqueueJob("PORT- " + portName, () -> {
+                ListenableFuture<Void> future = txRunner.callWithNewReadWriteTransactionAndSubmit(CONFIGURATION,
                     confTx -> {
-                        Optional<Interface> optionalInf =
-                                confTx.read(interfaceIdentifier).get();
+                        Optional<Interface> optionalInf = confTx.read(interfaceIdentifier).get();
                         if (optionalInf.isPresent()) {
                             InterfaceBuilder interfaceBuilder = new InterfaceBuilder(optionalInf.get());
                             if (origSecurityEnabled || updatedSecurityEnabled) {
@@ -286,13 +284,17 @@ public class NeutronPortChangeListener extends AsyncDataTreeChangeListenerBase<P
                         } else {
                             LOG.warn("update: Interface {} is not present", portName);
                         }
-                    })));
+                    });
+                ListenableFutures.addErrorLogging(future, LOG,
+                        "update: Failed to update interface {} with networkId {}", portName, network);
+                return Collections.singletonList(future);
+            });
         }
     }
 
     private void handleFloatingIpPortUpdated(@Nullable Port original, Port update) {
         if ((original == null || NeutronConstants.FLOATING_IP_DEVICE_ID_PENDING.equals(original.getDeviceId()))
-            && !NeutronConstants.FLOATING_IP_DEVICE_ID_PENDING.equals(update.getDeviceId())) {
+                && !NeutronConstants.FLOATING_IP_DEVICE_ID_PENDING.equals(update.getDeviceId())) {
             // populate floating-ip uuid and floating-ip port attributes (uuid, mac and subnet id for the ONLY
             // fixed IP) to be used by NAT, depopulated in NATService once mac is retrieved in the removal path
             addToFloatingIpPortInfo(new Uuid(update.getDeviceId()), update.getUuid(), update.getFixedIps().get(0)
@@ -412,20 +414,23 @@ public class NeutronPortChangeListener extends AsyncDataTreeChangeListenerBase<P
              *  cleanup of router interface flows*/
             nvpnManager.deleteVpnInterface(routerPort.getUuid().getValue(),
                                            null /* vpn-id */, null /* wrtConfigTxn*/);
-            final Uuid internetVpnId = vpnInstanceInternetUuid;
             // update RouterInterfaces map
             ListenableFutures.addErrorLogging(txRunner.callWithNewWriteOnlyTransactionAndSubmit(CONFIGURATION,
                 confTx -> {
                     IpVersionChoice ipVersion = IpVersionChoice.UNDEFINED;
                     for (FixedIps portIP : portIps) {
                         Subnetmap sn = neutronvpnUtils.getSubnetmap(portIP.getSubnetId());
+                        if (null == sn) {
+                            LOG.error("Subnetmap for subnet {} not found", portIP.getSubnetId().getValue());
+                            continue;
+                        }
                         // router Port have either IPv4 or IPv6, never both
                         ipVersion = neutronvpnUtils.getIpVersionFromString(sn.getSubnetIp());
                         String ipValue = portIP.getIpAddress().stringValue();
                         neutronvpnUtils.removeVpnPortFixedIpToPort(vpnId.getValue(), ipValue, confTx);
                         // NOTE:  Please donot change the order of calls to removeSubnetFromVpn and
                         // and updateSubnetNodeWithFixedIP
-                        nvpnManager.removeSubnetFromVpn(vpnId, portIP.getSubnetId(), internetVpnId);
+                        nvpnManager.removeSubnetFromVpn(vpnId, sn, sn.getInternetVpnId());
                         nvpnManager.updateSubnetNodeWithFixedIp(portIP.getSubnetId(), null, null,
                             null, null, null);
                     }
@@ -516,19 +521,25 @@ public class NeutronPortChangeListener extends AsyncDataTreeChangeListenerBase<P
         // the MAC of the router's gw port is not available to be set when the
         // router is written. We catch that here.
         InstanceIdentifier<Routers> routersId = NeutronvpnUtils.buildExtRoutersIdentifier(routerId);
-        Optional<Routers> optionalRouter = MDSALUtil.read(dataBroker, LogicalDatastoreType.CONFIGURATION, routersId);
-        if (!optionalRouter.isPresent()) {
-            return;
-        }
+        Optional<Routers> optionalRouter = null;
+        try {
+            optionalRouter = MDSALUtil.read(dataBroker, LogicalDatastoreType.CONFIGURATION, routersId);
+            if (!optionalRouter.isPresent()) {
+                return;
+            }
+            Routers extRouters = optionalRouter.get();
+            if (extRouters.getExtGwMacAddress() != null) {
+                return;
+            }
 
-        Routers extRouters = optionalRouter.get();
-        if (extRouters.getExtGwMacAddress() != null) {
-            return;
+            RoutersBuilder builder = new RoutersBuilder(extRouters);
+            builder.setExtGwMacAddress(routerGwPort.getMacAddress().getValue());
+            MDSALUtil.syncWrite(dataBroker, LogicalDatastoreType.CONFIGURATION, routersId, builder.build());
+        } catch (ExecutionException | InterruptedException e) {
+            LOG.error("setExternalGwMac: failed to read EXT-Routers for router Id {} rout-Gw port {} due to exception",
+                    routerId, routerGwPort, e);
         }
 
-        RoutersBuilder builder = new RoutersBuilder(extRouters);
-        builder.setExtGwMacAddress(routerGwPort.getMacAddress().getValue());
-        MDSALUtil.syncWrite(dataBroker, LogicalDatastoreType.CONFIGURATION, routersId, builder.build());
     }
 
     @Nullable
@@ -555,7 +566,7 @@ public class NeutronPortChangeListener extends AsyncDataTreeChangeListenerBase<P
             LOG.error("failed to read host config from host {}", hostId, e);
             return null;
         }
-        return hostConfig.orNull();
+        return hostConfig.orElse(null);
     }
 
     private boolean isPortBound(final Port port) {
@@ -570,7 +581,7 @@ public class NeutronPortChangeListener extends AsyncDataTreeChangeListenerBase<P
             return false;
         }
         String vnicType = portBinding.getVnicType().trim().toLowerCase(Locale.getDefault());
-        return vnicType.equals(NeutronConstants.VNIC_TYPE_DIRECT);
+        return NeutronConstants.VNIC_TYPE_DIRECT.equals(vnicType);
     }
 
     private boolean isSupportedVnicTypeByHost(final Port port, final String vnicType) {
@@ -626,23 +637,26 @@ public class NeutronPortChangeListener extends AsyncDataTreeChangeListenerBase<P
     private void handleNeutronPortCreated(final Port port) {
         final String portName = port.getUuid().getValue();
         final Uuid portId = port.getUuid();
+        final String networkId = port.getNetworkId().getValue();
         final List<FixedIps> portIpAddrsList = port.nonnullFixedIps();
         if (NeutronConstants.IS_ODL_DHCP_PORT.test(port)) {
             return;
         }
-        jobCoordinator.enqueueJob("PORT- " + portName, () -> {
-            // add direct port to subnetMaps config DS
-            if (!(NeutronUtils.isPortVnicTypeNormal(port)
+        if (!(NeutronUtils.isPortVnicTypeNormal(port)
                 || isPortTypeSwitchdev(port)
                 && isSupportedVnicTypeByHost(port, NeutronConstants.VNIC_TYPE_DIRECT))) {
-                for (FixedIps ip: portIpAddrsList) {
-                    nvpnManager.updateSubnetmapNodeWithPorts(ip.getSubnetId(), null, portId);
-                }
-                LOG.info("Port {} is not a normal and not a direct with switchdev VNIC type ;"
-                         + "OF Port interfaces are not created", portName);
-                return Collections.emptyList();
+            for (FixedIps ip: portIpAddrsList) {
+                nvpnManager.updateSubnetmapNodeWithPorts(ip.getSubnetId(), null, portId);
             }
-            return Collections.singletonList(txRunner.callWithNewWriteOnlyTransactionAndSubmit(CONFIGURATION, tx -> {
+            LOG.info("Port {} is not a normal and not a direct with switchdev VNIC type ;"
+                    + "OF Port interfaces are not created", portName);
+            return;
+        }
+        jobCoordinator.enqueueJob("PORT- " + portName, () -> {
+            // add direct port to subnetMaps config DS
+            // TODO: for direct port as well, operations should be carried out per subnet based on port IP
+
+            ListenableFuture<Void> future = txRunner.callWithNewWriteOnlyTransactionAndSubmit(CONFIGURATION, tx -> {
                 LOG.info("Of-port-interface creation for port {}", portName);
                 // Create of-port interface for this neutron port
                 String portInterfaceName = createOfPortInterface(port, tx);
@@ -672,15 +686,16 @@ public class NeutronPortChangeListener extends AsyncDataTreeChangeListenerBase<P
                 if (!vpnIdList.isEmpty()) {
                     // create new vpn-interface for neutron port
                     LOG.debug("handleNeutronPortCreated: Adding VPN Interface for port {} from network {}", portName,
-                            port.getNetworkId().toString());
+                            networkId);
                     nvpnManager.createVpnInterface(vpnIdList, port, tx);
-                    if (!routerIds.isEmpty()) {
-                        for (Uuid routerId : routerIds) {
-                            nvpnManager.addToNeutronRouterInterfacesMap(routerId,port.getUuid().getValue());
-                        }
+                    for (Uuid routerId : routerIds) {
+                        nvpnManager.addToNeutronRouterInterfacesMap(routerId,port.getUuid().getValue());
                     }
                 }
-            }));
+            });
+            ListenableFutures.addErrorLogging(future, LOG,
+                    "handleNeutronPortCreated: Failed for port {} with networkId {}", portName, networkId);
+            return Collections.singletonList(future);
         });
     }
 
@@ -688,17 +703,19 @@ public class NeutronPortChangeListener extends AsyncDataTreeChangeListenerBase<P
         final String portName = port.getUuid().getValue();
         final Uuid portId = port.getUuid();
         final List<FixedIps> portIpsList = port.nonnullFixedIps();
-        jobCoordinator.enqueueJob("PORT- " + portName,
-            () -> Collections.singletonList(txRunner.callWithNewWriteOnlyTransactionAndSubmit(CONFIGURATION, confTx -> {
-                if (!(NeutronUtils.isPortVnicTypeNormal(port) || isPortTypeSwitchdev(port))) {
-                    for (FixedIps ip : portIpsList) {
-                        // remove direct port from subnetMaps config DS
-                        nvpnManager.removePortsFromSubnetmapNode(ip.getSubnetId(), null, portId);
-                    }
-                    LOG.info("Port {} is not a normal and not a direct with switchdev VNIC type ;"
-                            + "Skipping OF Port interfaces removal", portName);
-                    return;
-                }
+        if (!(NeutronUtils.isPortVnicTypeNormal(port) || isPortTypeSwitchdev(port))) {
+            for (FixedIps ip : portIpsList) {
+                // remove direct port from subnetMaps config DS
+                // TODO: for direct port as well, operations should be carried out per subnet based on port IP
+                nvpnManager.removePortsFromSubnetmapNode(ip.getSubnetId(), null, portId);
+            }
+            LOG.info("Port {} is not a normal and not a direct with switchdev VNIC type ;"
+                    + "Skipping OF Port interfaces removal", portName);
+            return;
+        }
+        jobCoordinator.enqueueJob("PORT- " + portName, () -> {
+            ListenableFuture<Void> future = txRunner.callWithNewWriteOnlyTransactionAndSubmit(CONFIGURATION, confTx -> {
+
                 Uuid vpnId = null;
                 Set<Uuid> routerIds = new HashSet<>();
                 Uuid internetVpnId = null;
@@ -745,7 +762,12 @@ public class NeutronPortChangeListener extends AsyncDataTreeChangeListenerBase<P
                 deleteOfPortInterface(port, confTx);
                 //dissociate fixedIP from floatingIP if associated
                 nvpnManager.dissociatefixedIPFromFloatingIP(port.getUuid().getValue());
-            })));
+            });
+            ListenableFutures.addErrorLogging(future, LOG,
+                    "handleNeutronPortDeleted: Failed to update interface {} with networkId", portName,
+                    port.getNetworkId().getValue());
+            return Collections.singletonList(future);
+        });
     }
 
 
@@ -882,15 +904,49 @@ public class NeutronPortChangeListener extends AsyncDataTreeChangeListenerBase<P
                             interfaceIdentifier);
             if (!optionalInf.isPresent()) {
                 wrtConfigTxn.put(interfaceIdentifier, inf);
+            } else if (isInterfaceUpdated(inf, optionalInf.get())) {
+               /*
+                Case where an update DTCN wasn't received by this class due to node going down
+                upon cluster reboot or any other unknown reason
+                In such a case, updates contained in the missed DTCN won't be processed and have to be handled
+                explicitly
+                Update of subports (vlanId, splithorizon tag) is handled here
+                Update of portSecurity (PortSecurityEnabled, SecurityGroups, AllowedAddressPairs) add is handled
+                Update of portSecurity update/removed is not handled
+                Update of parentrefs is not handled as parentrefs updation is handled by IFM Oxygen onwards
+                */
+                wrtConfigTxn.put(interfaceIdentifier, inf);
+                LOG.error("Interface {} is already present and is updated", infName);
             } else {
                 LOG.warn("Interface {} is already present", infName);
             }
-        } catch (ReadFailedException e) {
+        } catch (ExecutionException | InterruptedException e) {
             LOG.error("failed to create interface {}", infName, e);
         }
         return infName;
     }
 
+    // Not for generic use. For a special case where update DTCN isn't received
+    private static boolean isInterfaceUpdated(Interface newInterface, Interface oldInterface) {
+        if (newInterface.augmentation(SplitHorizon.class) != null) {
+            if (oldInterface.augmentation(SplitHorizon.class) == null) {
+                return true;
+            }
+            if (!newInterface.augmentation(SplitHorizon.class).equals(oldInterface
+                    .augmentation(SplitHorizon.class))) {
+                return true;
+            }
+        }
+        if (!newInterface.augmentation(IfL2vlan.class).equals(oldInterface.augmentation(IfL2vlan.class))) {
+            return true;
+        }
+        if (newInterface.augmentation(InterfaceAcl.class) != null && oldInterface
+                .augmentation(InterfaceAcl.class) == null) {
+            return true;
+        }
+        return false;
+    }
+
     private Interface createInterface(Port port) {
         String interfaceName = port.getUuid().getValue();
         IfL2vlan.L2vlanMode l2VlanMode = IfL2vlan.L2vlanMode.Trunk;
@@ -944,7 +1000,7 @@ public class NeutronPortChangeListener extends AsyncDataTreeChangeListenerBase<P
             } else {
                 LOG.warn("deleteOfPortInterface: Interface {} is not present", name);
             }
-        } catch (ReadFailedException e) {
+        } catch (ExecutionException | InterruptedException e) {
             LOG.error("deleteOfPortInterface: Failed to delete interface {}", name, e);
         }
     }