nemo-renderers/openflow-renderer/src/main/java/org/opendaylight/nemo/renderer/openflow/physicalnetwork/PhysicalNetworkAdapter.java

   1 /*
   2  * Copyright (c) 2015 Huawei, 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.nemo.renderer.openflow.physicalnetwork;
  10
  11 import com.google.common.base.Optional;
  12 import com.google.common.util.concurrent.FutureCallback;
  13 import com.google.common.util.concurrent.Futures;
  14 import com.google.common.util.concurrent.ListenableFuture;
  15 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  16 import org.opendaylight.controller.md.sal.binding.api.DataChangeListener;
  17 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope;
  18 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  19 import org.opendaylight.controller.sal.binding.api.NotificationProviderService;
  20 import org.opendaylight.nemo.renderer.openflow.FlowUtils;
  21 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.yang.types.rev100924.MacAddress;
  22 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNodeConnector;
  23 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes;
  24 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnector;
  25 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnectorKey;
  26 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
  27 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
  28 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.generic.physical.network.rev151010.physical.network.physical.links.PhysicalLink;
  29 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.generic.physical.network.rev151010.physical.network.physical.links.PhysicalLinkBuilder;
  30 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.generic.physical.network.rev151010.physical.network.physical.links.PhysicalLinkKey;
  31 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.generic.physical.network.rev151010.physical.network.physical.nodes.PhysicalNode;
  32 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.generic.physical.network.rev151010.physical.network.physical.nodes.PhysicalNodeBuilder;
  33 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.generic.physical.network.rev151010.physical.network.physical.nodes.PhysicalNodeKey;
  34 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.generic.physical.network.rev151010.physical.node.instance.PhysicalPort;
  35 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.generic.physical.network.rev151010.physical.node.instance.PhysicalPortBuilder;
  36 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.generic.physical.network.rev151010.physical.node.instance.PhysicalPortKey;
  37 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.nemo.engine.common.rev151010.PhysicalLinkId;
  38 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.nemo.engine.common.rev151010.PhysicalNodeId;
  39 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.nemo.engine.common.rev151010.PhysicalPortId;
  40 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NetworkTopology;
  41 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.TopologyId;
  42 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.Topology;
  43 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyKey;
  44 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Link;
  45 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  46 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  47 import org.opendaylight.yangtools.yang.binding.NotificationListener;
  48 import org.slf4j.Logger;
  49 import org.slf4j.LoggerFactory;
  50
  51 import java.util.ArrayList;
  52 import java.util.List;
  53 import java.util.Timer;
  54 import java.util.TimerTask;
  55 import java.util.concurrent.CopyOnWriteArraySet;
  56 import java.util.concurrent.CountDownLatch;
  57
  58 public class PhysicalNetworkAdapter {
  59     private static final Logger log = LoggerFactory.getLogger(PhysicalNetworkAdapter.class);
  60     private static final String DEFAULT_TOPOLOGY_ID = "flow:1";
  61
  62     final private DataBroker dataBroker;
  63     private PhyConfigLoader phyConfigLoader;
  64     private DataBrokerAdapter dataBrokerAdapter;
  65     private PhysicalFlowUtils physicalFlowUtils;
  66     private FlowUtils ofFlowUtils;
  67     private NotificationProviderService notificationProviderService;
  68
  69     private CopyOnWriteArraySet<String> nodeIdSet;
  70     private CopyOnWriteArraySet<PhysicalLink> physicalLinkSet;
  71     private Timer phyTimer;
  72     private boolean running = false;
  73     private Integer mutex = 0;
  74     private ListenerRegistration<NotificationListener> ofPacketInListenerReg;
  75     private ListenerRegistration<DataChangeListener> ofNodesListenerReg;
  76     private ListenerRegistration<DataChangeListener> ofLinksListenerReg;
  77
  78     public PhysicalNetworkAdapter(DataBroker dataBroker
  79             , NotificationProviderService notificationProviderService
  80             , PhyConfigLoader phyConfigLoader
  81             , FlowUtils ofFlowUtils) {
  82         this.dataBroker = dataBroker;
  83         this.notificationProviderService = notificationProviderService;
  84         this.ofFlowUtils = ofFlowUtils;
  85         this.phyConfigLoader = phyConfigLoader;
  86         this.dataBrokerAdapter = new DataBrokerAdapter(dataBroker);
  87         physicalFlowUtils = new PhysicalFlowUtils(dataBroker);
  88         nodeIdSet = new CopyOnWriteArraySet<String>();
  89         physicalLinkSet = new CopyOnWriteArraySet<PhysicalLink>();
  90
  91         phyTimer = new Timer();
  92
  93
  94         registerListeners();
  95         initOFNodes();
  96         initOFLinks();
  97     }
  98
  99     public void close() {
 100         if (ofPacketInListenerReg != null)
 101             ofPacketInListenerReg.close();
 102         if (ofLinksListenerReg != null)
 103             ofLinksListenerReg.close();
 104         if (ofNodesListenerReg != null)
 105             ofNodesListenerReg.close();
 106         if (phyConfigLoader != null)
 107             phyConfigLoader.close();
 108         log.debug("Clear....\r\n{}", nodeIdSet);
 109         nodeIdSet.clear();
 110         physicalLinkSet.clear();
 111     }
 112
 113     public PhyConfigLoader getPhyConfigLoader() {
 114         return phyConfigLoader;
 115     }
 116
 117     /**
 118      * OFNode instance identifier
 119      *
 120      * @return
 121      */
 122     private InstanceIdentifier<Node> getOFNodeInstanceIdentifier() {
 123         return InstanceIdentifier.builder(Nodes.class).child(Node.class).build();
 124     }
 125
 126     private InstanceIdentifier<Link> getOFLinkInstanceIdentifier() {
 127         return InstanceIdentifier.builder(NetworkTopology.class).child(Topology.class, new TopologyKey(new TopologyId(DEFAULT_TOPOLOGY_ID))).child(Link.class).build();
 128     }
 129
 130     private InstanceIdentifier<Topology> getOFTopologyInstanceIdentifier() {
 131         return InstanceIdentifier.builder(NetworkTopology.class).child(Topology.class, new TopologyKey(new TopologyId(DEFAULT_TOPOLOGY_ID))).build();
 132     }
 133
 134     private InstanceIdentifier<Nodes> getOFNodesInstanceIdentifier() {
 135         return InstanceIdentifier.builder(Nodes.class).build();
 136     }
 137
 138     private InstanceIdentifier<NodeConnector> getOFPortInstanceIdentifier(NodeKey nodeKey, NodeConnectorKey connectorKey) {
 139         return InstanceIdentifier.builder(Nodes.class).child(Node.class, nodeKey).child(NodeConnector.class, connectorKey).build();
 140     }
 141
 142     private void registerListeners() {
 143         InstanceIdentifier<Node> nodeInsId = getOFNodeInstanceIdentifier();
 144         InstanceIdentifier<Link> linkInsId = getOFLinkInstanceIdentifier();
 145         ofNodesListenerReg = dataBroker.registerDataChangeListener(LogicalDatastoreType.OPERATIONAL, nodeInsId, new OFNodeListener(this), DataChangeScope.BASE);
 146         ofLinksListenerReg = dataBroker.registerDataChangeListener(LogicalDatastoreType.OPERATIONAL, linkInsId, new OFLinkListener(this), DataChangeScope.BASE);
 147         ofPacketInListenerReg = notificationProviderService.registerNotificationListener(new OFPacketInListener(ofFlowUtils));
 148     }
 149
 150     private void initOFLinks() {
 151         InstanceIdentifier<Topology> topologyInsId = getOFTopologyInstanceIdentifier();
 152         ListenableFuture<Optional<Topology>> topologyFuture = dataBroker.newReadOnlyTransaction().read(LogicalDatastoreType.OPERATIONAL, topologyInsId);
 153         Futures.addCallback(topologyFuture, new FutureCallback<Optional<Topology>>() {
 154             @Override
 155             public void onSuccess(Optional<Topology> result) {
 156                 if (result.isPresent() && result.get() instanceof Topology) {
 157                     Topology topology = result.get();
 158                     if (topology != null && topology.getLink() != null)
 159                         for (Link link : topology.getLink()) {
 160                             ofLinkAdded(link);
 161                         }
 162                 }
 163
 164                 return;
 165             }
 166
 167             @Override
 168             public void onFailure(Throwable t) {
 169                 log.error("Can not read the link info of topology {}: {}", DEFAULT_TOPOLOGY_ID, t);
 170                 return;
 171             }
 172         });
 173     }
 174
 175     private void initOFNodes() {
 176         InstanceIdentifier<Nodes> nodesInsId = getOFNodesInstanceIdentifier();
 177         ListenableFuture<Optional<Nodes>> nodesFuture = dataBroker.newReadOnlyTransaction().read(LogicalDatastoreType.OPERATIONAL, nodesInsId);
 178         Futures.addCallback(nodesFuture, new FutureCallback<Optional<Nodes>>() {
 179             @Override
 180             public void onSuccess(Optional<Nodes> result) {
 181                 if (result.isPresent() && result.get() instanceof Nodes) {
 182                     Nodes nodes = result.get();
 183                     if (nodes != null && nodes.getNode() != null)
 184                         for (Node node : nodes.getNode()) {
 185                             ofNodeAdded(node);
 186                         }
 187                 }
 188                 return;
 189             }
 190
 191             @Override
 192             public void onFailure(Throwable t) {
 193                 log.error("Can not read node information: {}", t);
 194                 return;
 195             }
 196         });
 197
 198     }
 199
 200     protected void ofNodeAdded(Node node) {
 201         log.debug("OF node added: {}.", node.getKey());
 202         String strNodeId = node.getId().getValue();
 203         // Add default flow entry. - Don't do this here, because it will
 204         // result in that the openflow plugin can't discover the topology.
 205         // Flow entries for LLDP are deployed manually through a shell script.
 206         // Flow entries for ARP are added in class FlowUtils.
 207 //        physicalFlowUtils.configArpPEntry(strNodeId);
 208 //        physicalFlowUtils.configLLDPEntry(strNodeId);
 209
 210         PhysicalNodeId nodeId = new PhysicalNodeId(strNodeId);
 211         PhysicalNodeBuilder nodeBuilder = new PhysicalNodeBuilder();
 212         nodeBuilder.setNodeId(nodeId);
 213         nodeBuilder.setKey(new PhysicalNodeKey(nodeId));
 214         List<PhysicalPort> physicalPortList = new ArrayList<PhysicalPort>();
 215         List<NodeConnector> nodeConnectors = node.getNodeConnector();
 216         if (nodeConnectors == null || nodeConnectors.size() == 0) {
 217             log.error("Node : {}, without port.", strNodeId);
 218         }
 219         if (nodeConnectors != null) {
 220             for (NodeConnector nodeConnector : nodeConnectors) {
 221                 PhysicalPort physicalPort = getPhysicalPort(node.getKey(), nodeConnector);
 222                 if (physicalPort != null) {
 223                     physicalPortList.add(physicalPort);
 224                 }
 225             }
 226         }
 227         nodeBuilder.setPhysicalPort(physicalPortList);
 228
 229         PhysicalNode confPhyNode = phyConfigLoader.getPhysicalNode(nodeId);
 230         if (confPhyNode != null) {
 231             nodeBuilder.setNodeType(confPhyNode.getNodeType());
 232             nodeBuilder.setAttribute(confPhyNode.getAttribute());
 233         } else {
 234             log.warn("Find one OF Node {},does not have info in config file.", node.getKey());
 235         }
 236         boolean result = dataBrokerAdapter.addPhysicalNode(nodeBuilder.build());
 237 //        if (result) {
 238         nodeIdSet.add(strNodeId);
 239         log.debug("Add....{}\r\n{}", strNodeId, nodeIdSet);
 240 //        }
 241     }
 242
 243     protected void ofNodeRemoved(Node node) {
 244         log.debug("OF node removed: {}.", node.getKey());
 245         String strNodeId = node.getId().getValue();
 246         PhysicalNodeId nodeId = new PhysicalNodeId(strNodeId);
 247         PhysicalNode confPhyNode = phyConfigLoader.getPhysicalNode(nodeId);
 248         if (confPhyNode == null) {
 249             log.warn("Find one OF Node removed {},does not have info in config file.", node.getKey());
 250         }
 251         boolean result = dataBrokerAdapter.removePhysicalNode(new PhysicalNodeKey(nodeId));
 252 //        if (result) {
 253         nodeIdSet.remove(strNodeId);
 254         log.debug("Remove....{}\r\n{}", strNodeId, nodeIdSet);
 255 //        }
 256     }
 257
 258     private PhysicalPort getPhysicalPort(NodeKey nodeKey, NodeConnector nodeConnector) {
 259         String strConnectorId = nodeConnector.getId().getValue();
 260         if (strConnectorId.contains("LOCAL"))
 261             return null;
 262         PhysicalPortId physicalPortId = new PhysicalPortId(strConnectorId);
 263         log.debug("Get port {} : {}.", nodeKey, nodeConnector.getId().getValue());
 264         FlowCapableNodeConnector flowCapableNodeConnector = getOFPort(nodeKey, nodeConnector.getKey());
 265         if (flowCapableNodeConnector != null) {
 266             PhysicalPortBuilder physicalPortBuilder = new PhysicalPortBuilder();
 267             physicalPortBuilder.setPortId(physicalPortId);
 268             physicalPortBuilder.setKey(new PhysicalPortKey(physicalPortId));
 269             physicalPortBuilder.setBandwidth(PhyConfigLoader.DEFAULT_LINK_BANDWIDTH);
 270             MacAddress macAddress = flowCapableNodeConnector.getHardwareAddress();
 271             physicalPortBuilder.setMacAddress(macAddress);
 272
 273             PhysicalPort confPhyPort = phyConfigLoader.getPhysicalPort(physicalPortId);
 274             if (confPhyPort != null) {
 275                 log.debug("Set port {} : {}.\r\n {} \r\n{}", nodeKey, nodeConnector.getId().getValue(), confPhyPort.getPortType().toString(), confPhyPort.getAttribute());
 276 //                long bandwidth = flowCapableNodeConnector.getMaximumSpeed() > 0 ? flowCapableNodeConnector.getMaximumSpeed() : confPhyPort.getBandwidth();
 277                 physicalPortBuilder.setPortType(confPhyPort.getPortType());
 278                 physicalPortBuilder.setAttribute(confPhyPort.getAttribute());
 279             } else {
 280                 log.warn("Can not get config info of {}-{} form data broker.", nodeKey.getId(), strConnectorId);
 281             }
 282             return physicalPortBuilder.build();
 283         } else {
 284             log.warn("Can not read OF port info of {}-{} form .", nodeKey.getId(), strConnectorId);
 285         }
 286         return null;
 287     }
 288
 289     private FlowCapableNodeConnector getOFPort(final NodeKey nodeKey, final NodeConnectorKey nodeConnectorKey) {
 290         final FlowCapableNodeConnector[] flowCapableNodeConnector = {null};
 291
 292         InstanceIdentifier<NodeConnector> nodeConnectorInsId = getOFPortInstanceIdentifier(nodeKey, nodeConnectorKey);
 293         final CountDownLatch downLatch = new CountDownLatch(1);
 294         ListenableFuture<Optional<NodeConnector>> nodeConnectorFuture = dataBroker.newReadOnlyTransaction().read(LogicalDatastoreType.OPERATIONAL, nodeConnectorInsId);
 295         Futures.addCallback(nodeConnectorFuture, new FutureCallback<Optional<NodeConnector>>() {
 296             @Override
 297             public void onSuccess(Optional<NodeConnector> result) {
 298                 if (result.isPresent() && result.get() instanceof NodeConnector) {
 299                     flowCapableNodeConnector[0] = result.get().getAugmentation(FlowCapableNodeConnector.class);
 300                     downLatch.countDown();
 301                 }
 302
 303                 return;
 304             }
 305
 306             @Override
 307             public void onFailure(Throwable t) {
 308                 log.error("Can not read the information of node connector {}-{} : {}", nodeKey, nodeConnectorKey, t);
 309                 downLatch.countDown();
 310                 return;
 311             }
 312         });
 313
 314         try {
 315             downLatch.await();
 316         } catch (InterruptedException e) {
 317             e.printStackTrace();
 318         }
 319
 320         return flowCapableNodeConnector[0];
 321     }
 322
 323     protected void ofLinkAdded(Link link) {
 324         log.debug("OF link added:{}.", link.getKey());
 325
 326         String srcNode = link.getSource().getSourceNode().getValue();
 327         String srcTp = link.getSource().getSourceTp().getValue();
 328         String dstNode = link.getDestination().getDestNode().getValue();
 329         String dstTp = link.getDestination().getDestTp().getValue();
 330
 331         String strLinkId = link.getLinkId().getValue();
 332         PhysicalLinkBuilder linkBuilder = new PhysicalLinkBuilder();
 333         linkBuilder.setLinkId(new PhysicalLinkId(strLinkId));
 334         linkBuilder.setSrcNodeId(new PhysicalNodeId(srcNode));
 335         linkBuilder.setSrcPortId(new PhysicalPortId(srcTp));
 336         linkBuilder.setDestNodeId(new PhysicalNodeId(dstNode));
 337         linkBuilder.setDestPortId(new PhysicalPortId(dstTp));
 338
 339         linkBuilder.setBandwidth(PhyConfigLoader.DEFAULT_LINK_BANDWIDTH);
 340         linkBuilder.setDelay(PhyConfigLoader.DEFAULT_LINK_DELAY);
 341         linkBuilder.setLossRate(PhyConfigLoader.DEFAULT_LINK_LOSS_RATE);
 342
 343         PhysicalLinkId physicalLinkId = new PhysicalLinkId(strLinkId);
 344         PhysicalLink physicalLink = phyConfigLoader.getPhysicalLink(physicalLinkId);
 345         if (physicalLink != null) {
 346             linkBuilder.setMetric(physicalLink.getMetric());
 347         } else {
 348             log.warn("Can not find conf info of {}.", link.getKey());
 349         }
 350         synchronized (mutex) {
 351             physicalLinkSet.add(linkBuilder.build());
 352             if (!running) {
 353                 phyTimer.schedule(new PhyTransmit(), 10, 500);
 354                 running = true;
 355             }
 356         }
 357     }
 358
 359     protected void ofLinkRemoved(Link link) {
 360         log.debug("OF link removed:{}.", link.getKey());
 361         String strLinkId = link.getLinkId().getValue();
 362         PhysicalLinkId linkId = new PhysicalLinkId(strLinkId);
 363         PhysicalLink confPhyLink = phyConfigLoader.getPhysicalLink(linkId);
 364         if (confPhyLink == null) {
 365             log.warn("Can not find conf info of {} while remove.", link);
 366         }
 367
 368         dataBrokerAdapter.removePhysicalLink(new PhysicalLinkKey(linkId));
 369     }
 370
 371     class PhyTransmit extends TimerTask {
 372
 373         @Override
 374         public void run() {
 375             synchronized (mutex) {
 376                 for (PhysicalLink physicalLink : physicalLinkSet) {
 377                     handleLink(physicalLink);
 378                 }
 379                 // Cancel timer. - Don't cancel timer, because this will result in
 380                 // that some physical links aren't wrote into data store sometimes.
 381 //                if (physicalLinkSet.size() == 0) {
 382 //                    phyTimer.cancel();
 383 //                    running = false;
 384 //                }
 385             }
 386         }
 387
 388         private void handleLink(PhysicalLink physicalLink) {
 389             String srcNodeId = physicalLink.getSrcNodeId().getValue();
 390             String dsrNodeId = physicalLink.getDestNodeId().getValue();
 391             if (nodeIdSet.contains(srcNodeId) && nodeIdSet.contains(dsrNodeId)) {
 392                 physicalLinkSet.remove(physicalLink);
 393                 log.debug("Put [{}]-[{}] to data broker.", srcNodeId, dsrNodeId);
 394                 dataBrokerAdapter.addPhysicalLink(physicalLink);
 395             }
 396         }
 397     }
 398 }