2 * Copyright (C) 2014 Red Hat, Inc.
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
8 * Authors : Dave Tucker, Madhu Venugopal
11 package org.opendaylight.ovsdb.openstack.netvirt.providers.openflow13;
13 import java.util.List;
15 import java.util.concurrent.BlockingQueue;
16 import java.util.concurrent.ExecutorService;
17 import java.util.concurrent.Executors;
18 import java.util.concurrent.LinkedBlockingQueue;
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;
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;
49 public class PipelineOrchestratorImpl implements PipelineOrchestrator, OpendaylightInventoryListener, TransactionChainListener {
51 private static final Logger logger = LoggerFactory.getLogger(PipelineOrchestratorImpl.class);
52 private List<Service> staticPipeline = Lists.newArrayList(
54 Service.ARP_RESPONDER,
57 Service.LOAD_BALANCER,
60 Service.L2_FORWARDING,
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() {
71 public void registerService(final ServiceReference ref, AbstractServiceInstance serviceInstance){
72 Service service = (Service)ref.getProperty(AbstractServiceInstance.SERVICE_PROPERTY);
73 serviceRegistry.put(service, serviceInstance);
76 public void unregisterService(final ServiceReference ref) {
77 serviceRegistry.remove(ref.getProperty(AbstractServiceInstance.SERVICE_PROPERTY));
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);
87 public AbstractServiceInstance getServiceInstance(Service service) {
88 if (service == null) return null;
89 return serviceRegistry.get(service);
93 eventHandler = Executors.newSingleThreadExecutor();
94 this.queue = new LinkedBlockingQueue<String>();
95 NotificationProviderService notificationService = mdsalConsumer.getNotificationService();
96 if (notificationService != null) {
97 notificationService.registerNotificationListener(this);
100 public void start() {
101 eventHandler.submit(new Runnable() {
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);
114 serviceInstance.programDefaultPipelineRule(nodeId);
117 } catch (Exception e) {
118 logger.warn("Processing interrupted, terminating ", e);
122 while (!queue.isEmpty()) {
132 eventHandler.shutdownNow();
135 void enqueue(String nodeId) {
137 queue.put(new String(nodeId));
138 } catch (InterruptedException e) {
139 logger.warn("Failed to enqueue operation {}", nodeId, e);
145 * Process the Node update notification. Check for Openflow node and make sure if the bridge is part of the Pipeline before
146 * programming the Pipeline specific flows.
149 public void onNodeUpdated(NodeUpdated nodeUpdated) {
150 NodeRef ref = nodeUpdated.getNodeRef();
151 InstanceIdentifier<Node> identifier = (InstanceIdentifier<Node>) ref.getValue();
152 logger.debug("Node Update received for : "+identifier.toString());
153 final NodeKey key = identifier.firstKeyOf(Node.class, NodeKey.class);
154 final String nodeId = key.getId().getValue();
155 if (key != null && key.getId().getValue().contains("openflow")) {
156 InstanceIdentifierBuilder<Node> builder = ((InstanceIdentifier<Node>) ref.getValue()).builder();
157 InstanceIdentifierBuilder<FlowCapableNode> augmentation = builder.augmentation(FlowCapableNode.class);
158 final InstanceIdentifier<FlowCapableNode> path = augmentation.build();
159 BindingTransactionChain txChain = mdsalConsumer.getDataBroker().createTransactionChain(this);
160 CheckedFuture readFuture = txChain.newReadWriteTransaction().read(LogicalDatastoreType.OPERATIONAL, path);
161 Futures.addCallback(readFuture, new FutureCallback<Optional<? extends DataObject>>() {
163 public void onSuccess(Optional<? extends DataObject> optional) {
164 if (!optional.isPresent()) {
170 public void onFailure(Throwable throwable) {
171 logger.debug(String.format("Can't retrieve node data for node %s. Writing node data with table0.", nodeId));
179 public void onTransactionChainFailed(final TransactionChain<?, ?> chain, final AsyncTransaction<?, ?> transaction,
180 final Throwable cause) {
181 logger.error("Failed to export Flow Capable Inventory, Transaction {} failed.",transaction.getIdentifier(),cause);
185 public void onTransactionChainSuccessful(final TransactionChain<?, ?> chain) {
189 public void onNodeConnectorRemoved(NodeConnectorRemoved arg0) {
190 // TODO Auto-generated method stub
195 public void onNodeConnectorUpdated(NodeConnectorUpdated arg0) {
196 // TODO Auto-generated method stub
201 public void onNodeRemoved(NodeRemoved arg0) {
202 // TODO Auto-generated method stub