openstack/net-virt-providers/src/main/java/org/opendaylight/ovsdb/openstack/netvirt/providers/openflow13/PipelineOrchestratorImpl.java

   1 /*
   2  * Copyright (C) 2014 Red Hat, Inc.
   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  * Authors : Dave Tucker, Madhu Venugopal
   9  */
  10
  11 package org.opendaylight.ovsdb.openstack.netvirt.providers.openflow13;
  12
  13 import java.util.List;
  14 import java.util.Map;
  15 import java.util.concurrent.BlockingQueue;
  16 import java.util.concurrent.ExecutorService;
  17 import java.util.concurrent.Executors;
  18 import java.util.concurrent.LinkedBlockingQueue;
  19
  20 import org.opendaylight.controller.md.sal.binding.api.BindingTransactionChain;
  21 import org.opendaylight.controller.md.sal.common.api.data.AsyncTransaction;
  22 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  23 import org.opendaylight.controller.md.sal.common.api.data.TransactionChain;
  24 import org.opendaylight.controller.md.sal.common.api.data.TransactionChainListener;
  25 import org.opendaylight.controller.sal.binding.api.NotificationProviderService;
  26 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
  27 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeConnectorRemoved;
  28 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeConnectorUpdated;
  29 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
  30 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRemoved;
  31 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeUpdated;
  32 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.OpendaylightInventoryListener;
  33 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
  34 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
  35 import org.opendaylight.yangtools.yang.binding.DataObject;
  36 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  37 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier.InstanceIdentifierBuilder;
  38 import org.osgi.framework.ServiceReference;
  39 import org.slf4j.Logger;
  40 import org.slf4j.LoggerFactory;
  41
  42 import com.google.common.base.Optional;
  43 import com.google.common.collect.Lists;
  44 import com.google.common.collect.Maps;
  45 import com.google.common.util.concurrent.CheckedFuture;
  46 import com.google.common.util.concurrent.FutureCallback;
  47 import com.google.common.util.concurrent.Futures;
  48
  49 public class PipelineOrchestratorImpl implements PipelineOrchestrator, OpendaylightInventoryListener, TransactionChainListener {
  50
  51     private static final Logger logger = LoggerFactory.getLogger(PipelineOrchestratorImpl.class);
  52     private List<Service> staticPipeline = Lists.newArrayList(
  53                                                                 Service.CLASSIFIER,
  54                                                                 Service.ARP_RESPONDER,
  55                                                                 Service.INBOUND_NAT,
  56                                                                 Service.INGRESS_ACL,
  57                                                                 Service.LOAD_BALANCER,
  58                                                                 Service.ROUTING,
  59                                                                 Service.L2_REWRITE,
  60                                                                 Service.EGRESS_ACL,
  61                                                                 Service.OUTBOUND_NAT,
  62                                                                 Service.L2_FORWARDING
  63                                                               );
  64     Map<Service, AbstractServiceInstance> serviceRegistry = Maps.newConcurrentMap();
  65     private volatile MdsalConsumer mdsalConsumer;
  66     private volatile BlockingQueue<String> queue;
  67     private ExecutorService eventHandler;
  68     public PipelineOrchestratorImpl() {
  69     }
  70
  71     public void registerService(final ServiceReference ref, AbstractServiceInstance serviceInstance){
  72         Service service = (Service)ref.getProperty(AbstractServiceInstance.SERVICE_PROPERTY);
  73         serviceRegistry.put(service, serviceInstance);
  74     }
  75
  76     public void unregisterService(final ServiceReference ref) {
  77         serviceRegistry.remove(ref.getProperty(AbstractServiceInstance.SERVICE_PROPERTY));
  78     }
  79     @Override
  80     public Service getNextServiceInPipeline(Service service) {
  81         int index = staticPipeline.indexOf(service);
  82         if (index >= staticPipeline.size() - 1) return null;
  83         return staticPipeline.get(index + 1);
  84     }
  85
  86     @Override
  87     public AbstractServiceInstance getServiceInstance(Service service) {
  88         if (service == null) return null;
  89         return serviceRegistry.get(service);
  90     }
  91
  92     public void init() {
  93         eventHandler = Executors.newSingleThreadExecutor();
  94         this.queue = new LinkedBlockingQueue<String>();
  95         NotificationProviderService notificationService = mdsalConsumer.getNotificationService();
  96         if (notificationService != null) {
  97             notificationService.registerNotificationListener(this);
  98         }
  99     }
 100     public void start() {
 101         eventHandler.submit(new Runnable()  {
 102             @Override
 103             public void run() {
 104                 try {
 105                     while (true) {
 106                         String nodeId = queue.take();
 107                         for (Service service : staticPipeline) {
 108                             AbstractServiceInstance serviceInstance = getServiceInstance(service);
 109                             if (!serviceInstance.isBridgeInPipeline(nodeId)) {
 110                                 logger.debug("Bridge {} is not in pipeline", nodeId);
 111                                 continue;
 112                             }
 113
 114                             serviceInstance.programDefaultPipelineRule(nodeId);
 115                         }
 116                     }
 117                 } catch (Exception e) {
 118                     logger.warn("Processing interrupted, terminating ", e);
 119                 }
 120
 121                 while (!queue.isEmpty()) {
 122                     queue.poll();
 123                 }
 124                 queue = null;
 125             }
 126         });
 127     }
 128
 129     public void stop() {
 130         queue.clear();
 131         eventHandler.shutdownNow();
 132     }
 133
 134     void enqueue(String nodeId) {
 135         try {
 136             queue.put(new String(nodeId));
 137         } catch (InterruptedException e) {
 138             logger.warn("Failed to enqueue operation {}", nodeId, e);
 139         }
 140     }
 141
 142
 143     /**
 144      * Process the Node update notification. Check for Openflow node and make sure if the bridge is part of the Pipeline before
 145      * programming the Pipeline specific flows.
 146      */
 147     @Override
 148     public void onNodeUpdated(NodeUpdated nodeUpdated) {
 149         NodeRef ref = nodeUpdated.getNodeRef();
 150         InstanceIdentifier<Node> identifier = (InstanceIdentifier<Node>) ref.getValue();
 151         logger.debug("Node Update received for : "+identifier.toString());
 152         final NodeKey key = identifier.firstKeyOf(Node.class, NodeKey.class);
 153         final String nodeId = key.getId().getValue();
 154         if (key != null && key.getId().getValue().contains("openflow")) {
 155             InstanceIdentifierBuilder<Node> builder = ((InstanceIdentifier<Node>) ref.getValue()).builder();
 156             InstanceIdentifierBuilder<FlowCapableNode> augmentation = builder.augmentation(FlowCapableNode.class);
 157             final InstanceIdentifier<FlowCapableNode> path = augmentation.build();
 158             BindingTransactionChain txChain = mdsalConsumer.getDataBroker().createTransactionChain(this);
 159             CheckedFuture readFuture = txChain.newReadWriteTransaction().read(LogicalDatastoreType.OPERATIONAL, path);
 160             Futures.addCallback(readFuture, new FutureCallback<Optional<? extends DataObject>>() {
 161                 @Override
 162                 public void onSuccess(Optional<? extends DataObject> optional) {
 163                     if (!optional.isPresent()) {
 164                         enqueue(nodeId);
 165                     }
 166                 }
 167
 168                 @Override
 169                 public void onFailure(Throwable throwable) {
 170                     logger.debug(String.format("Can't retrieve node data for node %s. Writing node data with table0.", nodeId));
 171                     enqueue(nodeId);
 172                 }
 173             });
 174         }
 175     }
 176
 177     @Override
 178     public void onTransactionChainFailed(final TransactionChain<?, ?> chain, final AsyncTransaction<?, ?> transaction,
 179             final Throwable cause) {
 180         logger.error("Failed to export Flow Capable Inventory, Transaction {} failed.",transaction.getIdentifier(),cause);
 181     }
 182
 183     @Override
 184     public void onTransactionChainSuccessful(final TransactionChain<?, ?> chain) {
 185     }
 186
 187     @Override
 188     public void onNodeConnectorRemoved(NodeConnectorRemoved arg0) {
 189         // TODO Auto-generated method stub
 190
 191     }
 192
 193     @Override
 194     public void onNodeConnectorUpdated(NodeConnectorUpdated arg0) {
 195         // TODO Auto-generated method stub
 196
 197     }
 198
 199     @Override
 200     public void onNodeRemoved(NodeRemoved arg0) {
 201         // TODO Auto-generated method stub
 202
 203     }
 204
 205 }