renderers/vpp/src/main/java/org/opendaylight/groupbasedpolicy/renderer/vpp/manager/VppNodeManager.java

   1 /*
   2  * Copyright (c) 2016 Cisco Systems, Inc. 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.groupbasedpolicy.renderer.vpp.manager;
  10
  11 import static org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.NetconfNodeConnectionStatus.ConnectionStatus.Connected;
  12 import static org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.NetconfNodeConnectionStatus.ConnectionStatus.Connecting;
  13
  14 import java.util.ArrayList;
  15 import java.util.Arrays;
  16 import java.util.HashMap;
  17 import java.util.List;
  18 import java.util.Map;
  19 import java.util.concurrent.Callable;
  20 import java.util.concurrent.ExecutionException;
  21 import java.util.concurrent.ExecutorService;
  22 import java.util.concurrent.Executors;
  23 import java.util.concurrent.Future;
  24 import java.util.stream.Collectors;
  25
  26 import javax.annotation.Nonnull;
  27 import javax.annotation.Nullable;
  28
  29 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  30 import org.opendaylight.controller.md.sal.binding.api.MountPoint;
  31 import org.opendaylight.controller.md.sal.binding.api.MountPointService;
  32 import org.opendaylight.controller.md.sal.binding.api.ReadOnlyTransaction;
  33 import org.opendaylight.controller.md.sal.binding.api.ReadWriteTransaction;
  34 import org.opendaylight.controller.md.sal.binding.api.WriteTransaction;
  35 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  36 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
  37 import org.opendaylight.controller.sal.binding.api.BindingAwareBroker;
  38 import org.opendaylight.groupbasedpolicy.renderer.vpp.nat.NatUtil;
  39 import org.opendaylight.groupbasedpolicy.renderer.vpp.util.VppIidFactory;
  40 import org.opendaylight.groupbasedpolicy.util.DataStoreHelper;
  41 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.iana._if.type.rev140508.EthernetCsmacd;
  42 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.IpAddress;
  43 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.Ipv4Address;
  44 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.interfaces.rev140508.Interfaces;
  45 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.ip.rev140616.Interface1;
  46 import org.opendaylight.yang.gen.v1.urn.opendaylight.groupbasedpolicy.renderer.rev151103.renderers.renderer.renderer.nodes.RendererNode;
  47 import org.opendaylight.yang.gen.v1.urn.opendaylight.groupbasedpolicy.renderer.rev151103.renderers.renderer.renderer.nodes.RendererNodeBuilder;
  48 import org.opendaylight.yang.gen.v1.urn.opendaylight.groupbasedpolicy.renderer.rev151103.renderers.renderer.renderer.nodes.RendererNodeKey;
  49 import org.opendaylight.yang.gen.v1.urn.opendaylight.groupbasedpolicy.vpp_renderer.rev160425.VppInterfaceAugmentation;
  50 import org.opendaylight.yang.gen.v1.urn.opendaylight.groupbasedpolicy.vpp_renderer.rev160425.VppInterfaceAugmentationBuilder;
  51 import org.opendaylight.yang.gen.v1.urn.opendaylight.groupbasedpolicy.vpp_renderer.rev160425.renderers.renderer.renderer.nodes.renderer.node.PhysicalInterface;
  52 import org.opendaylight.yang.gen.v1.urn.opendaylight.groupbasedpolicy.vpp_renderer.rev160425.renderers.renderer.renderer.nodes.renderer.node.PhysicalInterfaceBuilder;
  53 import org.opendaylight.yang.gen.v1.urn.opendaylight.groupbasedpolicy.vpp_renderer.rev160425.renderers.renderer.renderer.nodes.renderer.node.PhysicalInterfaceKey;
  54 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.NetconfNode;
  55 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.NetconfNodeConnectionStatus;
  56 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.connection.status.available.capabilities.AvailableCapability;
  57 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NetworkTopology;
  58 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NodeId;
  59 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.TopologyId;
  60 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.Topology;
  61 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyKey;
  62 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
  63 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.NodeKey;
  64 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  65 import org.slf4j.Logger;
  66 import org.slf4j.LoggerFactory;
  67
  68 import com.google.common.base.Optional;
  69 import com.google.common.base.Preconditions;
  70 import com.google.common.base.Splitter;
  71 import com.google.common.base.Strings;
  72 import com.google.common.collect.Lists;
  73 import com.google.common.collect.Sets;
  74 import com.google.common.util.concurrent.CheckedFuture;
  75
  76 public class VppNodeManager {
  77
  78     private static final short DURATION = 3000;
  79     private static final TopologyId TOPOLOGY_ID = new TopologyId("topology-netconf");
  80     private static final Logger LOG = LoggerFactory.getLogger(VppNodeManager.class);
  81     private static final String V3PO_CAPABILITY = "(urn:opendaylight:params:xml:ns:yang:v3po?revision=2016-12-14)v3po";
  82     private static final String INTERFACES_CAPABILITY = "(urn:ietf:params:xml:ns:yang:ietf-interfaces?revision=2014-05-08)ietf-interfaces";
  83     private static final NodeId CONTROLLER_CONFIG_NODE = new NodeId("controller-config");
  84     private static final String NO_PUBLIC_INT_SPECIFIED = "unspecified";
  85     private final Map<NodeId, PhysicalInterfaceKey> extInterfaces = new HashMap<>();
  86     private final DataBroker dataBroker;
  87     private final List<String> requiredCapabilities;
  88     private final MountPointService mountService;
  89
  90     public VppNodeManager(@Nonnull final DataBroker dataBroker,
  91             @Nonnull final BindingAwareBroker.ProviderContext session, @Nullable String physicalInterfaces) {
  92         this.dataBroker = Preconditions.checkNotNull(dataBroker);
  93         this.mountService = Preconditions.checkNotNull(session.getSALService(MountPointService.class));
  94         requiredCapabilities = initializeRequiredCapabilities();
  95         if (!Strings.isNullOrEmpty(physicalInterfaces) && physicalInterfaces != NO_PUBLIC_INT_SPECIFIED) {
  96             loadPhysicalInterfaces(physicalInterfaces);
  97         }
  98     }
  99
 100     /**
 101      * Caches list of physical interfaces.
 102      */
 103     public void loadPhysicalInterfaces(@Nonnull String physicalInterfaces) {
 104         for (String intfOnNode : Sets.newConcurrentHashSet(Splitter.on(",").split(physicalInterfaces))) {
 105             List<String> entries = Lists.newArrayList(Splitter.on(":").split(intfOnNode));
 106             if (entries.size() != 2) {
 107                 LOG.warn("Cannot resolve {} initial configuration for physical interfaces.", intfOnNode);
 108                 continue;
 109             }
 110             NodeId nodeId = new NodeId(entries.get(0));
 111             PhysicalInterfaceKey infaceKey = new PhysicalInterfaceKey(entries.get(1));
 112             LOG.info("Interface " + infaceKey + " on node " + nodeId + "will be considered as external");
 113             extInterfaces.put(nodeId, infaceKey);
 114         }
 115     }
 116
 117     /**
 118      * Synchronizes nodes to DataStore based on their modification state which results in
 119      * create/update/remove of Node.
 120      */
 121     public void syncNodes(Node dataAfter, Node dataBefore) {
 122         if (isControllerConfigNode(dataAfter, dataBefore)) {
 123             LOG.trace("{} is ignored by VPP-renderer", CONTROLLER_CONFIG_NODE);
 124             return;
 125         }
 126         // New node
 127         if (dataBefore == null && dataAfter != null) {
 128             createNode(dataAfter);
 129         }
 130         // Connected/disconnected node
 131         if (dataBefore != null && dataAfter != null) {
 132             updateNode(dataAfter);
 133         }
 134         // Removed node
 135         if (dataBefore != null && dataAfter == null) {
 136             removeNode(dataBefore);
 137         }
 138     }
 139
 140     private boolean isControllerConfigNode(Node dataAfter, Node dataBefore) {
 141         if (dataAfter != null) {
 142             return CONTROLLER_CONFIG_NODE.equals(dataAfter.getNodeId());
 143         }
 144         return CONTROLLER_CONFIG_NODE.equals(dataBefore.getNodeId());
 145     }
 146
 147     private void createNode(Node node) {
 148         LOG.info("Registering new node {}", node.getNodeId().getValue());
 149         NetconfNode netconfNode = getNodeAugmentation(node);
 150         if (netconfNode == null) {
 151             return;
 152         }
 153         NetconfNodeConnectionStatus.ConnectionStatus connectionStatus = netconfNode.getConnectionStatus();
 154         switch (connectionStatus) {
 155             case Connecting:
 156                 LOG.info("Connecting device {} ...", node.getNodeId().getValue());
 157                 break;
 158             case Connected:
 159                 resolveConnectedNode(node, netconfNode);
 160                 break;
 161             default:
 162                 break;
 163         }
 164     }
 165
 166     private void updateNode(Node node) {
 167         LOG.info("Updating node {}", node.getNodeId());
 168         NetconfNode netconfNode = getNodeAugmentation(node);
 169         if (netconfNode == null || netconfNode.getConnectionStatus() == null) {
 170             return;
 171         }
 172         NetconfNodeConnectionStatus.ConnectionStatus afterNodeStatus = netconfNode.getConnectionStatus();
 173         if (afterNodeStatus.equals(Connected)) {
 174             resolveConnectedNode(node, netconfNode);
 175         }
 176         if (afterNodeStatus.equals(Connecting)) {
 177             resolveDisconnectedNode(node);
 178             LOG.info("Node {} is disconnected, removing from available nodes", node.getNodeId().getValue());
 179         }
 180     }
 181
 182     private void removeNode(Node node) {
 183         resolveDisconnectedNode(node);
 184         LOG.info("Node {} is removed", node.getNodeId().getValue());
 185     }
 186
 187     private void resolveConnectedNode(Node node, NetconfNode netconfNode) {
 188         InstanceIdentifier<Node> mountPointIid = getMountpointIid(node);
 189         RendererNode rendererNode = remapNode(mountPointIid);
 190         if (!isCapableNetconfDevice(node, netconfNode)) {
 191             LOG.warn("Node {} is not connected.", node.getNodeId().getValue());
 192             return;
 193         }
 194         final DataBroker mountpoint = getNodeMountPoint(mountPointIid);
 195         if (mountpoint == null) {
 196             LOG.warn("Mountpoint not available for node {}", node.getNodeId().getValue());
 197             return;
 198         }
 199         final WriteTransaction wTx = dataBroker.newWriteOnlyTransaction();
 200         wTx.put(LogicalDatastoreType.OPERATIONAL, VppIidFactory.getRendererNodeIid(rendererNode), rendererNode, true);
 201         DataStoreHelper.submitToDs(wTx);
 202         LOG.info("Node {} is capable and ready.", node.getNodeId().getValue());
 203         syncPhysicalInterfacesInLocalDs(mountpoint, mountPointIid);
 204         NatUtil.resolveOutboundNatInterface(mountpoint, mountPointIid, node.getNodeId(), extInterfaces);
 205     }
 206
 207     private void resolveDisconnectedNode(Node node) {
 208         InstanceIdentifier<Node> mountPointIid = getMountpointIid(node);
 209         RendererNode rendererNode = remapNode(mountPointIid);
 210         final WriteTransaction wTx = dataBroker.newWriteOnlyTransaction();
 211         wTx.delete(LogicalDatastoreType.OPERATIONAL, VppIidFactory.getRendererNodeIid(rendererNode));
 212         CheckedFuture<Void, TransactionCommitFailedException> submitFuture = wTx.submit();
 213         try {
 214             submitFuture.checkedGet();
 215         } catch (TransactionCommitFailedException e) {
 216             LOG.error("Write transaction failed to {}", e.getMessage());
 217         } catch (Exception e) {
 218             LOG.error("Failed to .. {}", e.getMessage());
 219         }
 220     }
 221
 222     @Nullable
 223     private DataBroker getNodeMountPoint(InstanceIdentifier<Node> mountPointIid) {
 224         final Future<Optional<MountPoint>> futureOptionalObject = getMountpointFromSal(mountPointIid);
 225         try {
 226             final Optional<MountPoint> optionalObject = futureOptionalObject.get();
 227             LOG.debug("Optional mountpoint object: {}", optionalObject);
 228             MountPoint mountPoint;
 229             if (optionalObject.isPresent()) {
 230                 mountPoint = optionalObject.get();
 231                 if (mountPoint != null) {
 232                     Optional<DataBroker> optionalDataBroker = mountPoint.getService(DataBroker.class);
 233                     if (optionalDataBroker.isPresent()) {
 234                         return optionalDataBroker.get();
 235                     } else {
 236                         LOG.warn("Cannot obtain data broker from mountpoint {}", mountPoint);
 237                     }
 238                 } else {
 239                     LOG.warn("Cannot obtain mountpoint with IID {}", mountPointIid);
 240                 }
 241             }
 242             return null;
 243         } catch (ExecutionException | InterruptedException e) {
 244             LOG.warn("Unable to obtain mountpoint ... {}", e);
 245             return null;
 246         }
 247     }
 248
 249     private RendererNode remapNode(InstanceIdentifier<Node> path) {
 250         RendererNodeBuilder rendererNodeBuilder = new RendererNodeBuilder();
 251         rendererNodeBuilder.setKey(new RendererNodeKey(path)).setNodePath(path);
 252         return rendererNodeBuilder.build();
 253     }
 254
 255     private InstanceIdentifier<Node> getMountpointIid(Node node) {
 256         return InstanceIdentifier.builder(NetworkTopology.class)
 257             .child(Topology.class, new TopologyKey(TOPOLOGY_ID))
 258             .child(Node.class, new NodeKey(node.getNodeId()))
 259             .build();
 260     }
 261
 262     private boolean isCapableNetconfDevice(Node node, NetconfNode netconfAugmentation) {
 263         if (netconfAugmentation.getAvailableCapabilities() == null
 264                 || netconfAugmentation.getAvailableCapabilities().getAvailableCapability() == null
 265                 || netconfAugmentation.getAvailableCapabilities().getAvailableCapability().isEmpty()) {
 266             LOG.warn("Node {} does not contain any capabilities", node.getNodeId().getValue());
 267             return false;
 268         }
 269         if (!capabilityCheck(netconfAugmentation.getAvailableCapabilities().getAvailableCapability())) {
 270             LOG.warn("Node {} does not contain all capabilities required by vpp-renderer", node.getNodeId().getValue());
 271             return false;
 272         }
 273         return true;
 274     }
 275
 276     private boolean capabilityCheck(final List<AvailableCapability> capabilities) {
 277         final List<String> availableCapabilities = capabilities.stream()
 278                 .map(AvailableCapability::getCapability)
 279                 .collect(Collectors.toList());
 280         return requiredCapabilities.stream()
 281                 .allMatch(availableCapabilities::contains);
 282     }
 283
 284     private NetconfNode getNodeAugmentation(Node node) {
 285         NetconfNode netconfNode = node.getAugmentation(NetconfNode.class);
 286         if (netconfNode == null) {
 287             LOG.warn("Node {} is not a netconf device", node.getNodeId().getValue());
 288             return null;
 289         }
 290         return netconfNode;
 291     }
 292
 293     /**
 294      * Initialize all common capabilities required by VPP renderer. Any connected node is examined
 295      * whether it's
 296      * an appropriate device to handle configuration created by this renderer. A device must support
 297      * all capabilities
 298      * in list below.
 299      *
 300      * @return list of string representations of required capabilities
 301      */
 302     private List<String> initializeRequiredCapabilities() {
 303         // Required device capabilities
 304
 305         String[] capabilityEntries = {V3PO_CAPABILITY, INTERFACES_CAPABILITY};
 306         return Arrays.asList(capabilityEntries);
 307     }
 308
 309     // TODO bug 7699
 310     // This works as a workaround for mountpoint registration in cluster. If application is registered on different
 311     // node as netconf service, it obtains mountpoint registered by SlaveSalFacade (instead of MasterSalFacade). However
 312     // this service registers mountpoint a moment later then connectionStatus is set to "Connected". If NodeManager hits
 313     // state where device is connected but mountpoint is not yet available, try to get it again in a while
 314     private Future<Optional<MountPoint>> getMountpointFromSal(final InstanceIdentifier<Node> iid) {
 315         final ExecutorService executorService = Executors.newSingleThreadExecutor();
 316         final Callable<Optional<MountPoint>> task = () -> {
 317             byte attempt = 0;
 318             do {
 319                 try {
 320                     final Optional<MountPoint> optionalMountpoint = mountService.getMountPoint(iid);
 321                     if (optionalMountpoint.isPresent()) {
 322                         return optionalMountpoint;
 323                     }
 324                     LOG.warn("Mountpoint {} is not registered yet", iid);
 325                     Thread.sleep(DURATION);
 326                 } catch (InterruptedException e) {
 327                     LOG.warn("Thread interrupted to ", e);
 328                 }
 329                 attempt ++;
 330             } while (attempt <= 3);
 331             return Optional.absent();
 332         };
 333         return executorService.submit(task);
 334     }
 335
 336     private void syncPhysicalInterfacesInLocalDs(DataBroker mountPointDataBroker, InstanceIdentifier<Node> nodeIid) {
 337         ReadWriteTransaction rwTx = dataBroker.newReadWriteTransaction();
 338         ReadOnlyTransaction rTx = mountPointDataBroker.newReadOnlyTransaction();
 339         Optional<Interfaces> readIfaces = DataStoreHelper.readFromDs(LogicalDatastoreType.CONFIGURATION,
 340                 InstanceIdentifier.create(Interfaces.class), rTx);
 341         if (readIfaces.isPresent()) {
 342             InstanceIdentifier<RendererNode> rendererNodeIid = VppIidFactory.getRendererNodesIid()
 343                 .builder()
 344                 .child(RendererNode.class, new RendererNodeKey(nodeIid))
 345                 .build();
 346             Optional<RendererNode> optRendNode = DataStoreHelper.readFromDs(LogicalDatastoreType.OPERATIONAL,
 347                     rendererNodeIid, rwTx);
 348             NodeId nodeId = nodeIid.firstKeyOf(Node.class).getNodeId();
 349             RendererNode rendNode = new RendererNodeBuilder(optRendNode.get())
 350                 .addAugmentation(VppInterfaceAugmentation.class, resolveTerminationPoints(nodeId, readIfaces.get()))
 351                 .build();
 352             rwTx.put(LogicalDatastoreType.OPERATIONAL, VppIidFactory.getRendererNodeIid(optRendNode.get()), rendNode,
 353                     true);
 354         }
 355         rTx.close();
 356         DataStoreHelper.submitToDs(rwTx);
 357     }
 358
 359     private VppInterfaceAugmentation resolveTerminationPoints(NodeId nodeId, Interfaces interfaces) {
 360         List<PhysicalInterface> phIfaces = new ArrayList<>();
 361         if (interfaces != null && interfaces.getInterface() != null) {
 362             interfaces.getInterface()
 363                 .stream()
 364                 .filter(iface -> iface.getType().equals(EthernetCsmacd.class))
 365                 .filter(iface -> iface.getAugmentation(Interface1.class) != null)
 366                 .forEach(iface -> {
 367                     PhysicalInterfaceBuilder phIface = new PhysicalInterfaceBuilder();
 368                     phIface.setInterfaceName(iface.getName());
 369                     phIface.setType(iface.getType());
 370                     phIface.setAddress(resolveIpAddress(iface.getAugmentation(Interface1.class)));
 371
 372                     if (extInterfaces.get(nodeId) != null && extInterfaces.get(nodeId)
 373                         .getInterfaceName()
 374                         .equals(phIface.getInterfaceName())) {
 375                         phIface.setExternal(true);
 376                         LOG.trace("Assigning external Interface, interface name: {}, extInterface name: {}",
 377                             phIface.getInterfaceName(), extInterfaces.get(nodeId).getInterfaceName());
 378                     }
 379                     phIfaces.add(phIface.build());
 380                 });
 381         }
 382         return new VppInterfaceAugmentationBuilder().setPhysicalInterface(phIfaces).build();
 383     }
 384
 385     private List<IpAddress> resolveIpAddress(Interface1 iface) {
 386         if (iface.getIpv4() != null && iface.getIpv4().getAddress() != null) {
 387             return iface.getIpv4().getAddress().stream().map(ipv4 -> {
 388                 return new IpAddress(new Ipv4Address(ipv4.getIp().getValue()));
 389             }).collect(Collectors.toList());
 390         } else if (iface.getIpv6() != null && iface.getIpv6().getAddress() != null) {
 391             return iface.getIpv6().getAddress().stream().map(ipv6 -> {
 392                 return new IpAddress(new Ipv4Address(ipv6.getIp().getValue()));
 393             }).collect(Collectors.toList());
 394         }
 395         return Lists.newArrayList();
 396     }
 397 }