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