bb333de44e3f8551d61af9bac6c0dc93cdebfb3e
[openflowplugin.git] / openflowplugin / src / main / java / org / opendaylight / openflowplugin / openflow / md / core / MDController.java
1 /**
2  * Copyright (c) 2013 Cisco Systems, 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.openflowplugin.openflow.md.core;
10
11 import java.util.ArrayList;
12 import java.util.Collection;
13 import java.util.LinkedHashSet;
14 import java.util.List;
15 import java.util.Map;
16 import java.util.concurrent.BlockingQueue;
17 import java.util.concurrent.ConcurrentHashMap;
18 import java.util.concurrent.ConcurrentMap;
19 import java.util.concurrent.ExecutionException;
20 import java.util.concurrent.Future;
21 import java.util.concurrent.LinkedBlockingQueue;
22 import java.util.concurrent.RejectedExecutionException;
23 import java.util.concurrent.RejectedExecutionHandler;
24 import java.util.concurrent.ThreadPoolExecutor;
25 import java.util.concurrent.TimeUnit;
26 import java.util.concurrent.TimeoutException;
27
28 import org.opendaylight.openflowjava.protocol.api.connection.ConnectionConfiguration;
29 import org.opendaylight.openflowjava.protocol.spi.connection.SwitchConnectionProvider;
30 import org.opendaylight.openflowplugin.openflow.md.OFConstants;
31 import org.opendaylight.openflowplugin.openflow.md.core.session.OFSessionUtil;
32 import org.opendaylight.openflowplugin.openflow.md.core.translator.ErrorTranslator;
33 import org.opendaylight.openflowplugin.openflow.md.core.translator.ErrorV10Translator;
34 import org.opendaylight.openflowplugin.openflow.md.core.translator.ExperimenterTranslator;
35 import org.opendaylight.openflowplugin.openflow.md.core.translator.FeaturesV10ToNodeConnectorUpdatedTranslator;
36 import org.opendaylight.openflowplugin.openflow.md.core.translator.FlowRemovedTranslator;
37 import org.opendaylight.openflowplugin.openflow.md.core.translator.MultiPartMessageDescToNodeUpdatedTranslator;
38 import org.opendaylight.openflowplugin.openflow.md.core.translator.MultiPartReplyPortToNodeConnectorUpdatedTranslator;
39 import org.opendaylight.openflowplugin.openflow.md.core.translator.MultipartReplyTableFeaturesToTableUpdatedTranslator;
40 import org.opendaylight.openflowplugin.openflow.md.core.translator.MultipartReplyTranslator;
41 import org.opendaylight.openflowplugin.openflow.md.core.translator.PacketInTranslator;
42 import org.opendaylight.openflowplugin.openflow.md.core.translator.PacketInV10Translator;
43 import org.opendaylight.openflowplugin.openflow.md.core.translator.PortStatusMessageToNodeConnectorUpdatedTranslator;
44 import org.opendaylight.openflowplugin.openflow.md.lldp.LLDPSpeakerPopListener;
45 import org.opendaylight.openflowplugin.openflow.md.queue.PopListener;
46 import org.opendaylight.openflowplugin.openflow.md.util.OpenflowPortsUtil;
47 import org.opendaylight.openflowplugin.statistics.MessageSpy;
48 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.NodeErrorNotification;
49 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.SwitchFlowRemoved;
50 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.AggregateFlowStatisticsUpdate;
51 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.FlowsStatisticsUpdate;
52 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.FlowTableStatisticsUpdate;
53 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.GroupDescStatsUpdated;
54 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.GroupFeaturesUpdated;
55 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.GroupStatisticsUpdated;
56 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeConnectorUpdated;
57 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeUpdated;
58 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.MeterConfigStatsUpdated;
59 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.MeterFeaturesUpdated;
60 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.MeterStatisticsUpdated;
61 import org.opendaylight.yang.gen.v1.urn.opendaylight.node.error.service.rev140410.BadActionErrorNotification;
62 import org.opendaylight.yang.gen.v1.urn.opendaylight.node.error.service.rev140410.BadInstructionErrorNotification;
63 import org.opendaylight.yang.gen.v1.urn.opendaylight.node.error.service.rev140410.BadMatchErrorNotification;
64 import org.opendaylight.yang.gen.v1.urn.opendaylight.node.error.service.rev140410.BadRequestErrorNotification;
65 import org.opendaylight.yang.gen.v1.urn.opendaylight.node.error.service.rev140410.ExperimenterErrorNotification;
66 import org.opendaylight.yang.gen.v1.urn.opendaylight.node.error.service.rev140410.FlowModErrorNotification;
67 import org.opendaylight.yang.gen.v1.urn.opendaylight.node.error.service.rev140410.GroupModErrorNotification;
68 import org.opendaylight.yang.gen.v1.urn.opendaylight.node.error.service.rev140410.HelloFailedErrorNotification;
69 import org.opendaylight.yang.gen.v1.urn.opendaylight.node.error.service.rev140410.MeterModErrorNotification;
70 import org.opendaylight.yang.gen.v1.urn.opendaylight.node.error.service.rev140410.PortModErrorNotification;
71 import org.opendaylight.yang.gen.v1.urn.opendaylight.node.error.service.rev140410.QueueOpErrorNotification;
72 import org.opendaylight.yang.gen.v1.urn.opendaylight.node.error.service.rev140410.RoleRequestErrorNotification;
73 import org.opendaylight.yang.gen.v1.urn.opendaylight.node.error.service.rev140410.SwitchConfigErrorNotification;
74 import org.opendaylight.yang.gen.v1.urn.opendaylight.node.error.service.rev140410.TableFeaturesErrorNotification;
75 import org.opendaylight.yang.gen.v1.urn.opendaylight.node.error.service.rev140410.TableModErrorNotification;
76 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.ErrorMessage;
77 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.ExperimenterMessage;
78 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.FlowRemovedMessage;
79 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.GetFeaturesOutput;
80 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.MultipartReplyMessage;
81 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OfHeader;
82 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PacketInMessage;
83 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PortStatusMessage;
84 import org.opendaylight.yang.gen.v1.urn.opendaylight.packet.service.rev130709.PacketReceived;
85 import org.opendaylight.yang.gen.v1.urn.opendaylight.packet.service.rev130709.TransmitPacketInput;
86 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.NodeConnectorStatisticsUpdate;
87 import org.opendaylight.yang.gen.v1.urn.opendaylight.queue.statistics.rev131216.QueueStatisticsUpdate;
88 import org.opendaylight.yang.gen.v1.urn.opendaylight.table.service.rev131026.TableUpdated;
89 import org.opendaylight.yangtools.yang.binding.DataContainer;
90 import org.opendaylight.yangtools.yang.binding.DataObject;
91 import org.slf4j.Logger;
92 import org.slf4j.LoggerFactory;
93
94 import com.google.common.collect.Lists;
95 import com.google.common.util.concurrent.ForwardingBlockingQueue;
96 import com.google.common.util.concurrent.Futures;
97 import com.google.common.util.concurrent.ListenableFuture;
98 import com.google.common.util.concurrent.ListeningExecutorService;
99 import com.google.common.util.concurrent.MoreExecutors;
100
101 /**
102  *
103  */
104 public class MDController implements IMDController, AutoCloseable {
105
106     private static final Logger LOG = LoggerFactory.getLogger(MDController.class);
107
108     private Collection<SwitchConnectionProvider> switchConnectionProviders;
109
110     private ConcurrentMap<TranslatorKey, Collection<IMDMessageTranslator<OfHeader, List<DataObject>>>> messageTranslators;
111     private Map<Class<? extends DataObject>, Collection<PopListener<DataObject>>> popListeners;
112     private MessageSpy<DataContainer> messageSpyCounter;
113
114     final private int OF10 = OFConstants.OFP_VERSION_1_0;
115     final private int OF13 = OFConstants.OFP_VERSION_1_3;
116
117     private ErrorHandlerSimpleImpl errorHandler;
118
119     /**
120      * @return translator mapping
121      */
122     public Map<TranslatorKey, Collection<IMDMessageTranslator<OfHeader, List<DataObject>>>> getMessageTranslators() {
123         return messageTranslators;
124     }
125
126     /**
127      * provisioning of translator mapping
128      */
129     public void init() {
130         LOG.debug("init");
131
132         OpenflowPortsUtil.init();
133
134         messageTranslators = new ConcurrentHashMap<>();
135         popListeners = new ConcurrentHashMap<>();
136         //TODO: move registration to factory
137         addMessageTranslator(ErrorMessage.class, OF10, new ErrorV10Translator());
138         addMessageTranslator(ErrorMessage.class, OF13, new ErrorTranslator());
139         addMessageTranslator(FlowRemovedMessage.class, OF10, new FlowRemovedTranslator());
140         addMessageTranslator(FlowRemovedMessage.class, OF13, new FlowRemovedTranslator());
141         addMessageTranslator(PacketInMessage.class,OF10, new PacketInV10Translator());
142         addMessageTranslator(PacketInMessage.class,OF13, new PacketInTranslator());
143         addMessageTranslator(PortStatusMessage.class,OF10, new PortStatusMessageToNodeConnectorUpdatedTranslator());
144         addMessageTranslator(PortStatusMessage.class,OF13, new PortStatusMessageToNodeConnectorUpdatedTranslator());
145         addMessageTranslator(MultipartReplyMessage.class,OF13,new MultiPartReplyPortToNodeConnectorUpdatedTranslator());
146         addMessageTranslator(MultipartReplyMessage.class,OF10, new MultiPartMessageDescToNodeUpdatedTranslator());
147         addMessageTranslator(MultipartReplyMessage.class,OF13, new MultiPartMessageDescToNodeUpdatedTranslator());
148         addMessageTranslator(ExperimenterMessage.class, OF10, new ExperimenterTranslator());
149         addMessageTranslator(MultipartReplyMessage.class,OF10, new MultipartReplyTranslator());
150         addMessageTranslator(MultipartReplyMessage.class,OF13, new MultipartReplyTranslator());
151         addMessageTranslator(MultipartReplyMessage.class,OF13,new MultipartReplyTableFeaturesToTableUpdatedTranslator());
152         addMessageTranslator(GetFeaturesOutput.class,OF10, new FeaturesV10ToNodeConnectorUpdatedTranslator());
153
154         NotificationPopListener<DataObject> notificationPopListener = new NotificationPopListener<DataObject>();
155         notificationPopListener.setNotificationProviderService(
156                 OFSessionUtil.getSessionManager().getNotificationProviderService());
157         notificationPopListener.setMessageSpy(messageSpyCounter);
158
159         //TODO: move registration to factory
160         addMessagePopListener(NodeErrorNotification.class, notificationPopListener);
161         addMessagePopListener(BadActionErrorNotification.class, notificationPopListener);
162         addMessagePopListener(BadInstructionErrorNotification.class, notificationPopListener);
163         addMessagePopListener(BadMatchErrorNotification.class, notificationPopListener);
164         addMessagePopListener(BadRequestErrorNotification.class, notificationPopListener);
165         addMessagePopListener(ExperimenterErrorNotification.class, notificationPopListener);
166         addMessagePopListener(FlowModErrorNotification.class, notificationPopListener);
167         addMessagePopListener(GroupModErrorNotification.class, notificationPopListener);
168         addMessagePopListener(HelloFailedErrorNotification.class, notificationPopListener);
169         addMessagePopListener(MeterModErrorNotification.class, notificationPopListener);
170         addMessagePopListener(PortModErrorNotification.class, notificationPopListener);
171         addMessagePopListener(QueueOpErrorNotification.class, notificationPopListener);
172         addMessagePopListener(RoleRequestErrorNotification.class, notificationPopListener);
173         addMessagePopListener(SwitchConfigErrorNotification.class, notificationPopListener);
174         addMessagePopListener(TableFeaturesErrorNotification.class, notificationPopListener);
175         addMessagePopListener(TableModErrorNotification.class, notificationPopListener);
176         addMessagePopListener(NodeConnectorUpdated.class,notificationPopListener);
177         addMessagePopListener(PacketReceived.class,notificationPopListener);
178         addMessagePopListener(TransmitPacketInput.class, notificationPopListener);
179         addMessagePopListener(NodeUpdated.class, notificationPopListener);
180
181         addMessagePopListener(SwitchFlowRemoved.class, notificationPopListener);
182         addMessagePopListener(TableUpdated.class, notificationPopListener);
183
184         //Notification registration for flow statistics
185         addMessagePopListener(FlowsStatisticsUpdate.class, notificationPopListener);
186         addMessagePopListener(AggregateFlowStatisticsUpdate.class, notificationPopListener);
187
188         //Notification registrations for group-statistics
189         addMessagePopListener(GroupStatisticsUpdated.class, notificationPopListener);
190         addMessagePopListener(GroupFeaturesUpdated.class, notificationPopListener);
191         addMessagePopListener(GroupDescStatsUpdated.class, notificationPopListener);
192
193         //Notification registrations for meter-statistics
194         addMessagePopListener(MeterStatisticsUpdated.class, notificationPopListener);
195         addMessagePopListener(MeterConfigStatsUpdated.class, notificationPopListener);
196         addMessagePopListener(MeterFeaturesUpdated.class, notificationPopListener);
197
198         //Notification registration for port-statistics
199         addMessagePopListener(NodeConnectorStatisticsUpdate.class, notificationPopListener);
200
201         //Notification registration for flow-table statistics
202         addMessagePopListener(FlowTableStatisticsUpdate.class, notificationPopListener);
203
204         //Notification registration for queue-statistics
205         addMessagePopListener(QueueStatisticsUpdate.class, notificationPopListener);
206
207         //Notification for LLDPSpeaker
208         LLDPSpeakerPopListener<DataObject> lldpPopListener  = new LLDPSpeakerPopListener<DataObject>();
209         addMessagePopListener(NodeConnectorUpdated.class,lldpPopListener);
210
211         // Push the updated Listeners to Session Manager which will be then picked up by ConnectionConductor eventually
212         OFSessionUtil.getSessionManager().setTranslatorMapping(messageTranslators);
213         OFSessionUtil.getSessionManager().setPopListenerMapping(popListeners);
214         OFSessionUtil.getSessionManager().setMessageSpy(messageSpyCounter);
215
216         // prepare worker pool for rpc
217         // TODO: get size from configSubsystem
218         int rpcThreadLimit = 10;
219         ListeningExecutorService rpcPoolDelegator = createRpcPoolSpyDecorated(rpcThreadLimit, messageSpyCounter);
220         OFSessionUtil.getSessionManager().setRpcPool(rpcPoolDelegator);
221
222     }
223
224     /**
225      * @param rpcThreadLimit
226      * @param messageSpy
227      * @return
228      */
229     private static ListeningExecutorService createRpcPoolSpyDecorated(final int rpcThreadLimit, final MessageSpy<DataContainer> messageSpy) {
230         final BlockingQueue<Runnable> delegate = new LinkedBlockingQueue<>(100000);
231         final BlockingQueue<Runnable> queue = new ForwardingBlockingQueue<Runnable>() {
232             @Override
233             protected BlockingQueue<Runnable> delegate() {
234                 return delegate;
235             }
236
237             @Override
238             public boolean offer(final Runnable r) {
239                 // ThreadPoolExecutor will spawn a new thread after core size is reached only
240                 // if the queue.offer returns false.
241                 return false;
242             }
243         };
244
245         ThreadPoolLoggingExecutor rpcPool = new ThreadPoolLoggingExecutor(rpcThreadLimit, rpcThreadLimit, 0L,
246                 TimeUnit.MILLISECONDS, queue, "OFRpc");
247         rpcPool.setRejectedExecutionHandler(new RejectedExecutionHandler() {
248                         @Override
249                         public void rejectedExecution(final Runnable r, final ThreadPoolExecutor executor) {
250                                 try {
251                                         executor.getQueue().put(r);
252                                 } catch (InterruptedException e) {
253                                         throw new RejectedExecutionException("Interrupted while waiting on queue", e);
254                                 }
255
256                         }
257                 });
258         ListeningExecutorService listeningRpcPool = MoreExecutors.listeningDecorator(rpcPool);
259         RpcListeningExecutorService rpcPoolDecorated = new RpcListeningExecutorService(listeningRpcPool);
260         rpcPoolDecorated.setMessageSpy(messageSpy);
261         return rpcPoolDecorated;
262     }
263
264     /**
265      * @param switchConnectionProviders
266      *            the switchConnectionProviders to set
267      */
268     public void setSwitchConnectionProviders(final Collection<SwitchConnectionProvider> switchConnectionProviders) {
269         this.switchConnectionProviders = switchConnectionProviders;
270     }
271
272     /**
273      * Function called by dependency manager after "init ()" is called and after
274      * the services provided by the class are registered in the service registry
275      *
276      */
277     public void start() {
278         LOG.debug("starting ..");
279         LOG.debug("switchConnectionProvider: " + switchConnectionProviders);
280         // setup handler
281         SwitchConnectionHandlerImpl switchConnectionHandler = new SwitchConnectionHandlerImpl();
282         switchConnectionHandler.setMessageSpy(messageSpyCounter);
283
284         errorHandler = new ErrorHandlerSimpleImpl();
285
286         switchConnectionHandler.setErrorHandler(errorHandler);
287         switchConnectionHandler.init();
288
289         List<ListenableFuture<Boolean>> starterChain = new ArrayList<>(switchConnectionProviders.size());
290         for (SwitchConnectionProvider switchConnectionPrv : switchConnectionProviders) {
291             switchConnectionPrv.setSwitchConnectionHandler(switchConnectionHandler);
292             ListenableFuture<Boolean> isOnlineFuture = switchConnectionPrv.startup();
293             starterChain.add(isOnlineFuture);
294         }
295
296         Future<List<Boolean>> srvStarted = Futures.allAsList(starterChain);
297     }
298
299     /**
300      * @return wished connections configurations
301      * @deprecated use configSubsystem
302      */
303     @Deprecated
304     private static Collection<ConnectionConfiguration> getConnectionConfiguration() {
305         // TODO:: get config from state manager
306         ConnectionConfiguration configuration = ConnectionConfigurationFactory.getDefault();
307         ConnectionConfiguration configurationLegacy = ConnectionConfigurationFactory.getLegacy();
308         return Lists.newArrayList(configuration, configurationLegacy);
309     }
310
311     /**
312      * Function called by the dependency manager before the services exported by
313      * the component are unregistered, this will be followed by a "destroy ()"
314      * calls
315      *
316      */
317     public void stop() {
318         LOG.debug("stopping");
319         List<ListenableFuture<Boolean>> stopChain = new ArrayList<>(switchConnectionProviders.size());
320         try {
321             for (SwitchConnectionProvider switchConnectionPrv : switchConnectionProviders) {
322                 ListenableFuture<Boolean> shutdown =  switchConnectionPrv.shutdown();
323                 stopChain.add(shutdown);
324             }
325             Futures.allAsList(stopChain).get(5000, TimeUnit.MILLISECONDS);
326         } catch (InterruptedException | ExecutionException | TimeoutException e) {
327             LOG.error(e.getMessage(), e);
328         }
329         close();
330     }
331
332     /**
333      * Function called by the dependency manager when at least one dependency
334      * become unsatisfied or when the component is shutting down because for
335      * example bundle is being stopped.
336      *
337      */
338     public void destroy() {
339         close();
340     }
341
342     @Override
343     public void addMessageTranslator(final Class<? extends DataObject> messageType, final int version, final IMDMessageTranslator<OfHeader, List<DataObject>> translator) {
344         TranslatorKey tKey = new TranslatorKey(version, messageType.getName());
345
346         Collection<IMDMessageTranslator<OfHeader, List<DataObject>>> existingValues = messageTranslators.get(tKey);
347         if (existingValues == null) {
348             existingValues = new LinkedHashSet<>();
349             messageTranslators.put(tKey, existingValues);
350         }
351         existingValues.add(translator);
352         LOG.debug("{} is now translated by {}", messageType, translator);
353     }
354
355     @Override
356     public void removeMessageTranslator(final Class<? extends DataObject> messageType, final int version, final IMDMessageTranslator<OfHeader, List<DataObject>> translator) {
357         TranslatorKey tKey = new TranslatorKey(version, messageType.getName());
358         Collection<IMDMessageTranslator<OfHeader, List<DataObject>>> values = messageTranslators.get(tKey);
359         if (values != null) {
360             values.remove(translator);
361             if (values.isEmpty()) {
362                 messageTranslators.remove(tKey);
363             }
364             LOG.debug("{} is now removed from translators", translator);
365          }
366     }
367
368     @Override
369     public void addMessagePopListener(final Class<? extends DataObject> messageType, final PopListener<DataObject> popListener) {
370         Collection<PopListener<DataObject>> existingValues = popListeners.get(messageType);
371         if (existingValues == null) {
372             existingValues = new LinkedHashSet<>();
373             popListeners.put(messageType, existingValues);
374         }
375         existingValues.add(popListener);
376         LOG.debug("{} is now popListened by {}", messageType, popListener);
377     }
378
379     @Override
380     public void removeMessagePopListener(final Class<? extends DataObject> messageType, final PopListener<DataObject> popListener) {
381         Collection<PopListener<DataObject>> values = popListeners.get(messageType);
382         if (values != null) {
383             values.remove(popListener);
384             if (values.isEmpty()) {
385                 popListeners.remove(messageType);
386             }
387             LOG.debug("{} is now removed from popListeners", popListener);
388          }
389     }
390
391     /**
392      * @param messageSpyCounter the messageSpyCounter to set
393      */
394     public void setMessageSpyCounter(
395             final MessageSpy<DataContainer> messageSpyCounter) {
396         this.messageSpyCounter = messageSpyCounter;
397     }
398
399     @Override
400     public void close() {
401         LOG.debug("close");
402         messageSpyCounter = null;
403         messageTranslators = null;
404         popListeners = null;
405         for (SwitchConnectionProvider switchConnectionPrv : switchConnectionProviders) {
406             switchConnectionPrv.setSwitchConnectionHandler(null);
407         }
408         switchConnectionProviders = null;
409         OpenflowPortsUtil.close();
410         OFSessionUtil.releaseSessionManager();
411         errorHandler = null;
412     }
413 }