Bug 1811 Fixed the pipeline by moving L2Fwd to the end.
[ovsdb.git] / 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                     e.printStackTrace();
120                 }
121
122                 while (!queue.isEmpty()) {
123                     queue.poll();
124                 }
125                 queue = null;
126             }
127         });
128     }
129
130     public void stop() {
131         queue.clear();
132         eventHandler.shutdownNow();
133     }
134
135     void enqueue(String nodeId) {
136         try {
137             queue.put(new String(nodeId));
138         } catch (InterruptedException e) {
139             logger.warn("Failed to enqueue operation {}", nodeId, e);
140         }
141     }
142
143
144     /**
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.
147      */
148     @Override
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>>() {
162                 @Override
163                 public void onSuccess(Optional<? extends DataObject> optional) {
164                     if (!optional.isPresent()) {
165                         enqueue(nodeId);
166                     }
167                 }
168
169                 @Override
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));
172                     enqueue(nodeId);
173                 }
174             });
175         }
176     }
177
178     @Override
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);
182     }
183
184     @Override
185     public void onTransactionChainSuccessful(final TransactionChain<?, ?> chain) {
186     }
187
188     @Override
189     public void onNodeConnectorRemoved(NodeConnectorRemoved arg0) {
190         // TODO Auto-generated method stub
191
192     }
193
194     @Override
195     public void onNodeConnectorUpdated(NodeConnectorUpdated arg0) {
196         // TODO Auto-generated method stub
197
198     }
199
200     @Override
201     public void onNodeRemoved(NodeRemoved arg0) {
202         // TODO Auto-generated method stub
203
204     }
205
206 }