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