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,
59 Service.L3_FORWARDING,
65 Map<Service, AbstractServiceInstance> serviceRegistry = Maps.newConcurrentMap();
66 private volatile MdsalConsumer mdsalConsumer;
67 private volatile BlockingQueue<String> queue;
68 private ExecutorService eventHandler;
69 public PipelineOrchestratorImpl() {
72 public void registerService(final ServiceReference ref, AbstractServiceInstance serviceInstance){
73 Service service = (Service)ref.getProperty(AbstractServiceInstance.SERVICE_PROPERTY);
74 serviceRegistry.put(service, serviceInstance);
77 public void unregisterService(final ServiceReference ref) {
78 serviceRegistry.remove(ref.getProperty(AbstractServiceInstance.SERVICE_PROPERTY));
81 public Service getNextServiceInPipeline(Service service) {
82 int index = staticPipeline.indexOf(service);
83 if (index >= staticPipeline.size() - 1) return null;
84 return staticPipeline.get(index + 1);
88 public AbstractServiceInstance getServiceInstance(Service service) {
89 if (service == null) return null;
90 return serviceRegistry.get(service);
94 eventHandler = Executors.newSingleThreadExecutor();
95 this.queue = new LinkedBlockingQueue<String>();
96 NotificationProviderService notificationService = mdsalConsumer.getNotificationService();
97 if (notificationService != null) {
98 notificationService.registerNotificationListener(this);
101 public void start() {
102 eventHandler.submit(new Runnable() {
107 String nodeId = queue.take();
109 * Since we are hooking on OpendaylightInventoryListener and as observed in
110 * Bug 1997 multiple Threads trying to write to a same table at the same time
111 * causes programming issues. Hence delaying the programming by a second to
112 * avoid the clash. This hack/workaround should be removed once Bug 1997 is resolved.
115 for (Service service : staticPipeline) {
116 AbstractServiceInstance serviceInstance = getServiceInstance(service);
117 serviceInstance.programDefaultPipelineRule(nodeId);
120 } catch (Exception e) {
121 logger.warn("Processing interrupted, terminating ", e);
124 while (!queue.isEmpty()) {
134 eventHandler.shutdownNow();
137 void enqueue(String nodeId) {
139 queue.put(new String(nodeId));
140 } catch (InterruptedException e) {
141 logger.warn("Failed to enqueue operation {}", nodeId, e);
147 * Process the Node update notification. Check for Openflow node and make sure if the bridge is part of the Pipeline before
148 * programming the Pipeline specific flows.
151 public void onNodeUpdated(NodeUpdated nodeUpdated) {
152 NodeRef ref = nodeUpdated.getNodeRef();
153 InstanceIdentifier<Node> identifier = (InstanceIdentifier<Node>) ref.getValue();
154 logger.debug("Node Update received for : "+identifier.toString());
155 final NodeKey key = identifier.firstKeyOf(Node.class, NodeKey.class);
156 final String nodeId = key.getId().getValue();
157 if (key != null && key.getId().getValue().contains("openflow")) {
158 InstanceIdentifierBuilder<Node> builder = ((InstanceIdentifier<Node>) ref.getValue()).builder();
159 InstanceIdentifierBuilder<FlowCapableNode> augmentation = builder.augmentation(FlowCapableNode.class);
160 final InstanceIdentifier<FlowCapableNode> path = augmentation.build();
161 BindingTransactionChain txChain = mdsalConsumer.getDataBroker().createTransactionChain(this);
162 CheckedFuture readFuture = txChain.newReadWriteTransaction().read(LogicalDatastoreType.OPERATIONAL, path);
163 Futures.addCallback(readFuture, new FutureCallback<Optional<? extends DataObject>>() {
165 public void onSuccess(Optional<? extends DataObject> optional) {
166 if (!optional.isPresent()) {
172 public void onFailure(Throwable throwable) {
173 logger.debug(String.format("Can't retrieve node data for node %s. Writing node data with table0.", nodeId));
181 public void onTransactionChainFailed(final TransactionChain<?, ?> chain, final AsyncTransaction<?, ?> transaction,
182 final Throwable cause) {
183 logger.error("Failed to export Flow Capable Inventory, Transaction {} failed.",transaction.getIdentifier(),cause);
187 public void onTransactionChainSuccessful(final TransactionChain<?, ?> chain) {
191 public void onNodeConnectorRemoved(NodeConnectorRemoved arg0) {
192 // TODO Auto-generated method stub
197 public void onNodeConnectorUpdated(NodeConnectorUpdated arg0) {
198 // TODO Auto-generated method stub
203 public void onNodeRemoved(NodeRemoved arg0) {
204 // TODO Auto-generated method stub