openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/device/DeviceContextImpl.java

   1 /**
   2  * Copyright (c) 2015 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 package org.opendaylight.openflowplugin.impl.device;
   9
  10 import com.google.common.annotations.VisibleForTesting;
  11 import com.google.common.base.Function;
  12 import com.google.common.base.Optional;
  13 import com.google.common.base.Preconditions;
  14 import com.google.common.util.concurrent.AsyncFunction;
  15 import com.google.common.util.concurrent.CheckedFuture;
  16 import com.google.common.util.concurrent.FutureCallback;
  17 import com.google.common.util.concurrent.FutureFallback;
  18 import com.google.common.util.concurrent.Futures;
  19 import com.google.common.util.concurrent.ListenableFuture;
  20 import io.netty.util.HashedWheelTimer;
  21 import io.netty.util.Timeout;
  22 import java.math.BigInteger;
  23 import java.util.Collection;
  24 import java.util.Collections;
  25 import java.util.HashMap;
  26 import java.util.HashSet;
  27 import java.util.LinkedList;
  28 import java.util.List;
  29 import java.util.Map;
  30 import java.util.concurrent.ConcurrentHashMap;
  31 import javax.annotation.CheckForNull;
  32 import javax.annotation.Nonnull;
  33 import javax.annotation.Nullable;
  34 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  35 import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
  36 import org.opendaylight.controller.md.sal.binding.api.NotificationService;
  37 import org.opendaylight.controller.md.sal.binding.api.ReadOnlyTransaction;
  38 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  39 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
  40 import org.opendaylight.openflowjava.protocol.api.connection.ConnectionAdapter;
  41 import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueue;
  42 import org.opendaylight.openflowjava.protocol.api.keys.MessageTypeKey;
  43 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
  44 import org.opendaylight.openflowplugin.api.openflow.connection.OutboundQueueProvider;
  45 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
  46 import org.opendaylight.openflowplugin.api.openflow.device.DeviceState;
  47 import org.opendaylight.openflowplugin.api.openflow.device.MessageTranslator;
  48 import org.opendaylight.openflowplugin.api.openflow.device.RequestContext;
  49 import org.opendaylight.openflowplugin.api.openflow.device.TranslatorLibrary;
  50 import org.opendaylight.openflowplugin.api.openflow.device.Xid;
  51 import org.opendaylight.openflowplugin.api.openflow.device.handlers.DeviceTerminationPhaseHandler;
  52 import org.opendaylight.openflowplugin.api.openflow.device.handlers.MultiMsgCollector;
  53 import org.opendaylight.openflowplugin.api.openflow.md.core.SwitchConnectionDistinguisher;
  54 import org.opendaylight.openflowplugin.api.openflow.md.core.TranslatorKey;
  55 import org.opendaylight.openflowplugin.api.openflow.registry.ItemLifeCycleRegistry;
  56 import org.opendaylight.openflowplugin.api.openflow.registry.flow.DeviceFlowRegistry;
  57 import org.opendaylight.openflowplugin.api.openflow.registry.flow.FlowDescriptor;
  58 import org.opendaylight.openflowplugin.api.openflow.registry.flow.FlowRegistryKey;
  59 import org.opendaylight.openflowplugin.api.openflow.registry.group.DeviceGroupRegistry;
  60 import org.opendaylight.openflowplugin.api.openflow.registry.meter.DeviceMeterRegistry;
  61 import org.opendaylight.openflowplugin.api.openflow.rpc.ItemLifeCycleKeeper;
  62 import org.opendaylight.openflowplugin.api.openflow.rpc.RpcContext;
  63 import org.opendaylight.openflowplugin.api.openflow.rpc.listener.ItemLifecycleListener;
  64 import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsContext;
  65 import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.MessageSpy;
  66 import org.opendaylight.openflowplugin.extension.api.ConvertorMessageFromOFJava;
  67 import org.opendaylight.openflowplugin.extension.api.ExtensionConverterProviderKeeper;
  68 import org.opendaylight.openflowplugin.extension.api.core.extension.ExtensionConverterProvider;
  69 import org.opendaylight.openflowplugin.extension.api.exception.ConversionException;
  70 import org.opendaylight.openflowplugin.extension.api.path.MessagePath;
  71 import org.opendaylight.openflowplugin.impl.common.ItemLifeCycleSourceImpl;
  72 import org.opendaylight.openflowplugin.impl.common.NodeStaticReplyTranslatorUtil;
  73 import org.opendaylight.openflowplugin.impl.device.listener.MultiMsgCollectorImpl;
  74 import org.opendaylight.openflowplugin.impl.registry.flow.DeviceFlowRegistryImpl;
  75 import org.opendaylight.openflowplugin.impl.registry.flow.FlowRegistryKeyFactory;
  76 import org.opendaylight.openflowplugin.impl.registry.group.DeviceGroupRegistryImpl;
  77 import org.opendaylight.openflowplugin.impl.registry.meter.DeviceMeterRegistryImpl;
  78 import org.opendaylight.openflowplugin.impl.util.DeviceInitializationUtils;
  79 import org.opendaylight.openflowplugin.impl.util.MdSalRegistrationUtils;
  80 import org.opendaylight.openflowplugin.openflow.md.core.session.SwitchConnectionCookieOFImpl;
  81 import org.opendaylight.yang.gen.v1.urn.opendaylight.experimenter.message.service.rev151020.ExperimenterMessageFromDevBuilder;
  82 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
  83 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNodeConnector;
  84 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
  85 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow;
  86 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowKey;
  87 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeConnectorId;
  88 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeConnectorRef;
  89 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
  90 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnector;
  91 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnectorBuilder;
  92 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnectorKey;
  93 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
  94 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.common.types.rev130731.PortReason;
  95 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.Error;
  96 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.ExperimenterMessage;
  97 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.FlowRemoved;
  98 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.MultipartReply;
  99 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OfHeader;
 100 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PacketIn;
 101 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PacketInMessage;
 102 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PortGrouping;
 103 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PortStatusMessage;
 104 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.experimenter.core.ExperimenterDataOfChoice;
 105 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.experimenter.types.rev151020.experimenter.core.message.ExperimenterMessageOfChoice;
 106 import org.opendaylight.yang.gen.v1.urn.opendaylight.packet.service.rev130709.PacketReceived;
 107 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.FlowCapableNodeConnectorStatisticsData;
 108 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.FlowCapableNodeConnectorStatisticsDataBuilder;
 109 import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.OfpRole;
 110 import org.opendaylight.yangtools.yang.binding.DataObject;
 111 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
 112 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
 113 import org.slf4j.Logger;
 114 import org.slf4j.LoggerFactory;
 115
 116 /**
 117  *
 118  */
 119 public class DeviceContextImpl implements DeviceContext, ExtensionConverterProviderKeeper {
 120
 121     private static final Logger LOG = LoggerFactory.getLogger(DeviceContextImpl.class);
 122
 123     // TODO: drain factor should be parametrized
 124     public static final float REJECTED_DRAIN_FACTOR = 0.25f;
 125     // TODO: low water mark factor should be parametrized
 126     private static final float LOW_WATERMARK_FACTOR = 0.75f;
 127     // TODO: high water mark factor should be parametrized
 128     private static final float HIGH_WATERMARK_FACTOR = 0.95f;
 129
 130     private final ConnectionContext primaryConnectionContext;
 131     private final DeviceState deviceState;
 132     private final DataBroker dataBroker;
 133     private final HashedWheelTimer hashedWheelTimer;
 134     private final Map<SwitchConnectionDistinguisher, ConnectionContext> auxiliaryConnectionContexts;
 135     private final TransactionChainManager transactionChainManager;
 136     private final DeviceFlowRegistry deviceFlowRegistry;
 137     private final DeviceGroupRegistry deviceGroupRegistry;
 138     private final DeviceMeterRegistry deviceMeterRegistry;
 139     private final Collection<DeviceTerminationPhaseHandler> closeHandlers = new HashSet<>();
 140     private final PacketInRateLimiter packetInLimiter;
 141     private final MessageSpy messageSpy;
 142     private final ItemLifeCycleKeeper flowLifeCycleKeeper;
 143     private NotificationPublishService notificationPublishService;
 144     private NotificationService notificationService;
 145     private final OutboundQueue outboundQueueProvider;
 146     private Timeout barrierTaskTimeout;
 147     private final MessageTranslator<PortGrouping, FlowCapableNodeConnector> portStatusTranslator;
 148     private final MessageTranslator<PacketInMessage, PacketReceived> packetInTranslator;
 149     private final MessageTranslator<FlowRemoved, org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowRemoved> flowRemovedTranslator;
 150     private final TranslatorLibrary translatorLibrary;
 151     private final Map<Long, NodeConnectorRef> nodeConnectorCache;
 152     private final ItemLifeCycleRegistry itemLifeCycleSourceRegistry;
 153     private RpcContext rpcContext;
 154     private ExtensionConverterProvider extensionConverterProvider;
 155
 156     private final boolean switchFeaturesMandatory;
 157     private StatisticsContext statisticsContext;
 158
 159
 160     @VisibleForTesting
 161     DeviceContextImpl(@Nonnull final ConnectionContext primaryConnectionContext,
 162                       @Nonnull final DeviceState deviceState,
 163                       @Nonnull final DataBroker dataBroker,
 164                       @Nonnull final HashedWheelTimer hashedWheelTimer,
 165                       @Nonnull final MessageSpy _messageSpy,
 166                       @Nonnull final OutboundQueueProvider outboundQueueProvider,
 167                       @Nonnull final TranslatorLibrary translatorLibrary,
 168                       final boolean switchFeaturesMandatory) {
 169         this.switchFeaturesMandatory = switchFeaturesMandatory;
 170         this.primaryConnectionContext = Preconditions.checkNotNull(primaryConnectionContext);
 171         this.deviceState = Preconditions.checkNotNull(deviceState);
 172         this.dataBroker = Preconditions.checkNotNull(dataBroker);
 173         this.hashedWheelTimer = Preconditions.checkNotNull(hashedWheelTimer);
 174         this.outboundQueueProvider = Preconditions.checkNotNull(outboundQueueProvider);
 175         primaryConnectionContext.setDeviceDisconnectedHandler(DeviceContextImpl.this);
 176         this.transactionChainManager = new TransactionChainManager(dataBroker, deviceState);
 177         auxiliaryConnectionContexts = new HashMap<>();
 178         deviceFlowRegistry = new DeviceFlowRegistryImpl();
 179         deviceGroupRegistry = new DeviceGroupRegistryImpl();
 180         deviceMeterRegistry = new DeviceMeterRegistryImpl();
 181         messageSpy = _messageSpy;
 182
 183         packetInLimiter = new PacketInRateLimiter(primaryConnectionContext.getConnectionAdapter(),
 184                 /*initial*/ 1000, /*initial*/2000, messageSpy, REJECTED_DRAIN_FACTOR);
 185
 186         this.translatorLibrary = translatorLibrary;
 187         portStatusTranslator = translatorLibrary.lookupTranslator(
 188                 new TranslatorKey(deviceState.getVersion(), PortGrouping.class.getName()));
 189         packetInTranslator = translatorLibrary.lookupTranslator(
 190                 new TranslatorKey(deviceState.getVersion(), PacketIn.class.getName()));
 191         flowRemovedTranslator = translatorLibrary.lookupTranslator(
 192                 new TranslatorKey(deviceState.getVersion(), FlowRemoved.class.getName()));
 193
 194
 195         nodeConnectorCache = new ConcurrentHashMap<>();
 196
 197         itemLifeCycleSourceRegistry = new ItemLifeCycleRegistryImpl();
 198         flowLifeCycleKeeper = new ItemLifeCycleSourceImpl();
 199         itemLifeCycleSourceRegistry.registerLifeCycleSource(flowLifeCycleKeeper);
 200     }
 201
 202     /**
 203      * This method is called from {@link DeviceManagerImpl} only. So we could say "posthandshake process finish"
 204      * and we are able to set a scheduler for an automatic transaction submitting by time (0,5sec).
 205      */
 206     void initialSubmitTransaction() {
 207         transactionChainManager.initialSubmitWriteTransaction();
 208     }
 209
 210     @Override
 211     public Long reservedXidForDeviceMessage() {
 212         return outboundQueueProvider.reserveEntry();
 213     }
 214
 215     @Override
 216     public void addAuxiliaryConnectionContext(final ConnectionContext connectionContext) {
 217         final SwitchConnectionDistinguisher connectionDistinguisher = createConnectionDistinguisher(connectionContext);
 218         auxiliaryConnectionContexts.put(connectionDistinguisher, connectionContext);
 219     }
 220
 221     private static SwitchConnectionDistinguisher createConnectionDistinguisher(final ConnectionContext connectionContext) {
 222         return new SwitchConnectionCookieOFImpl(connectionContext.getFeatures().getAuxiliaryId());
 223     }
 224
 225     @Override
 226     public void removeAuxiliaryConnectionContext(final ConnectionContext connectionContext) {
 227         final SwitchConnectionDistinguisher connectionDistinguisher = createConnectionDistinguisher(connectionContext);
 228         if (null != connectionDistinguisher) {
 229             auxiliaryConnectionContexts.remove(connectionDistinguisher);
 230         }
 231     }
 232
 233     @Override
 234     public DeviceState getDeviceState() {
 235         return deviceState;
 236     }
 237
 238     @Override
 239     public ReadOnlyTransaction getReadTransaction() {
 240         return dataBroker.newReadOnlyTransaction();
 241     }
 242
 243     @Override
 244     public ListenableFuture<Void> onClusterRoleChange(final OfpRole oldRole, @CheckForNull final OfpRole role) {
 245         LOG.trace("onClusterRoleChange {} for node:", role, deviceState.getNodeId());
 246         Preconditions.checkArgument(role != null);
 247         if (role.equals(oldRole)) {
 248             LOG.debug("Demanded role change for device {} is not changed. OldRole: {}, NewRole {}", deviceState.getNodeId(), oldRole, role);
 249             return Futures.immediateFuture(null);
 250         }
 251         if (OfpRole.BECOMEMASTER.equals(role)) {
 252             if (!deviceState.deviceSynchronized()) {
 253                 //TODO: no necessary code for yet - it needs for initialization phase only
 254                 LOG.debug("Setup Empty TxManager {} for initialization phase", getDeviceState().getNodeId());
 255                 transactionChainManager.activateTransactionManager();
 256                 return Futures.immediateCheckedFuture(null);
 257             }
 258             /* Relevant for no initial Slave-to-Master scenario in cluster */
 259             final ListenableFuture<Void> deviceInitialization = asyncClusterRoleChange();
 260             Futures.addCallback(deviceInitialization, new FutureCallback<Void>() {
 261
 262                 @Override
 263                 public void onSuccess(@Nullable final Void aVoid) {
 264                     //No operation
 265                 }
 266
 267                 @Override
 268                 public void onFailure(final Throwable throwable) {
 269                     LOG.debug("Device {} init unexpected fail. Unregister RPCs", getDeviceState().getNodeId());
 270                     MdSalRegistrationUtils.unregisterServices(getRpcContext());
 271                 }
 272
 273             });
 274
 275             return deviceInitialization;
 276
 277         } else if (OfpRole.BECOMESLAVE.equals(role)) {
 278             if (null != rpcContext) {
 279                 MdSalRegistrationUtils.registerSlaveServices(rpcContext, role);
 280             }
 281             return transactionChainManager.deactivateTransactionManager();
 282         } else {
 283             LOG.warn("Unknown OFCluster Role {} for Node {}", role, deviceState.getNodeId());
 284             if (null != rpcContext) {
 285                 MdSalRegistrationUtils.unregisterServices(rpcContext);
 286             }
 287             return transactionChainManager.deactivateTransactionManager();
 288         }
 289     }
 290
 291     /*
 292      * we don't have active TxManager so anything will not be stored to DS yet, but we have
 293      * check all NodeInformation for statistics otherwise statistics will not contains
 294      * all possible MultipartTypes for polling in StatTypeList
 295      */
 296     private ListenableFuture<Void> asyncClusterRoleChange() {
 297         if (statisticsContext == null) {
 298             final String errMsg = String.format("DeviceCtx %s is up but we are missing StatisticsContext", deviceState.getNodeId());
 299             LOG.warn(errMsg);
 300             return Futures.immediateFailedFuture(new IllegalStateException(errMsg));
 301         }
 302         if (rpcContext == null) {
 303             final String errMsg = String.format("DeviceCtx %s is up but we are missing RpcContext", deviceState.getNodeId());
 304             LOG.warn(errMsg);
 305             return Futures.immediateFailedFuture(new IllegalStateException(errMsg));
 306         }
 307
 308         final InstanceIdentifier<FlowCapableNode> ofNodeII = deviceState.getNodeInstanceIdentifier()
 309                 .augmentation(FlowCapableNode.class);
 310         final ReadOnlyTransaction readTx = getReadTransaction();
 311         final CheckedFuture<Optional<FlowCapableNode>, ReadFailedException> readOfNodeFuture = readTx.read(
 312                 LogicalDatastoreType.OPERATIONAL, ofNodeII);
 313
 314         final ListenableFuture<Void> nodeInitInfoFuture = Futures.transform(readOfNodeFuture,
 315                 new AsyncFunction<Optional<FlowCapableNode>, Void>() {
 316                     @Override
 317                     public ListenableFuture<Void> apply(final Optional<FlowCapableNode> input) throws Exception {
 318                         if (!input.isPresent() || input.get().getTable() == null || input.get().getTable().isEmpty()) {
 319                             /* Last master close fail scenario so we would like to activate TxManager */
 320                             LOG.debug("Operational DS for Device {} has to be replaced", deviceState.getNodeId());
 321                             getDeviceState().setDeviceSynchronized(false);
 322                             transactionChainManager.activateTransactionManager();
 323                         }
 324                         return DeviceInitializationUtils.initializeNodeInformation(DeviceContextImpl.this, switchFeaturesMandatory);
 325                     }
 326                 });
 327
 328         final ListenableFuture<Boolean> statPollFuture = Futures.transform(nodeInitInfoFuture,
 329                 new AsyncFunction<Void, Boolean>() {
 330
 331                     @Override
 332                     public ListenableFuture<Boolean> apply(final Void input) throws Exception {
 333                         getStatisticsContext().statListForCollectingInitialization();
 334                         if (getDeviceState().deviceSynchronized()) {
 335                             return Futures.immediateFuture(Boolean.TRUE);
 336                         }
 337                         return getStatisticsContext().gatherDynamicData();
 338                     }
 339                 });
 340
 341         return Futures.transform(statPollFuture, new Function<Boolean, Void>() {
 342
 343             @Override
 344             public Void apply(final Boolean input) {
 345                 if (ConnectionContext.CONNECTION_STATE.RIP.equals(getPrimaryConnectionContext().getConnectionState())) {
 346                     final String errMsg = String.format("We lost connection for Device %s, context has to be closed.",
 347                             getDeviceState().getNodeId());
 348                     LOG.warn(errMsg);
 349                     transactionChainManager.clearUnsubmittedTransaction();
 350                     throw new IllegalStateException(errMsg);
 351                 }
 352                 if (!input.booleanValue()) {
 353                     final String errMsg = String.format("Get Initial Device %s information fails",
 354                             getDeviceState().getNodeId());
 355                     LOG.warn(errMsg);
 356                     transactionChainManager.clearUnsubmittedTransaction();
 357                     throw new IllegalStateException(errMsg);
 358                 }
 359                 LOG.debug("Get Initial Device {} information is successful", getDeviceState().getNodeId());
 360                 getDeviceState().setDeviceSynchronized(true);
 361                 transactionChainManager.activateTransactionManager();
 362                 MdSalRegistrationUtils.registerMasterServices(getRpcContext(), DeviceContextImpl.this, OfpRole.BECOMEMASTER);
 363                 getRpcContext().registerStatCompatibilityServices();
 364                 initialSubmitTransaction();
 365                 getDeviceState().setStatisticsPollingEnabledProp(true);
 366                 return null;
 367             }
 368         });
 369     }
 370
 371     @Override
 372     public <T extends DataObject> void writeToTransaction(final LogicalDatastoreType store,
 373                                                           final InstanceIdentifier<T> path, final T data) {
 374         transactionChainManager.writeToTransaction(store, path, data);
 375     }
 376
 377     @Override
 378     public <T extends DataObject> void addDeleteToTxChain(final LogicalDatastoreType store, final InstanceIdentifier<T> path) {
 379         transactionChainManager.addDeleteOperationTotTxChain(store, path);
 380     }
 381
 382     @Override
 383     public boolean submitTransaction() {
 384         return transactionChainManager.submitWriteTransaction();
 385     }
 386
 387     @Override
 388     public ConnectionContext getPrimaryConnectionContext() {
 389         return primaryConnectionContext;
 390     }
 391
 392     @Override
 393     public ConnectionContext getAuxiliaryConnectiobContexts(final BigInteger cookie) {
 394         return auxiliaryConnectionContexts.get(new SwitchConnectionCookieOFImpl(cookie.longValue()));
 395     }
 396
 397     @Override
 398     public DeviceFlowRegistry getDeviceFlowRegistry() {
 399         return deviceFlowRegistry;
 400     }
 401
 402     @Override
 403     public DeviceGroupRegistry getDeviceGroupRegistry() {
 404         return deviceGroupRegistry;
 405     }
 406
 407     @Override
 408     public DeviceMeterRegistry getDeviceMeterRegistry() {
 409         return deviceMeterRegistry;
 410     }
 411
 412     @Override
 413     public void processReply(final OfHeader ofHeader) {
 414         if (ofHeader instanceof Error) {
 415             messageSpy.spyMessage(ofHeader.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
 416         } else {
 417             messageSpy.spyMessage(ofHeader.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
 418         }
 419     }
 420
 421     @Override
 422     public void processReply(final Xid xid, final List<MultipartReply> ofHeaderList) {
 423         for (final MultipartReply multipartReply : ofHeaderList) {
 424             messageSpy.spyMessage(multipartReply.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
 425         }
 426     }
 427
 428     @Override
 429     public void processFlowRemovedMessage(final FlowRemoved flowRemoved) {
 430         final ItemLifecycleListener itemLifecycleListener = flowLifeCycleKeeper.getItemLifecycleListener();
 431         if (itemLifecycleListener != null) {
 432             //1. translate to general flow (table, priority, match, cookie)
 433             final org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowRemoved flowRemovedNotification =
 434                     flowRemovedTranslator.translate(flowRemoved, this, null);
 435             //2. create registry key
 436             final FlowRegistryKey flowRegKey = FlowRegistryKeyFactory.create(flowRemovedNotification);
 437             //3. lookup flowId
 438             final FlowDescriptor flowDescriptor = deviceFlowRegistry.retrieveIdForFlow(flowRegKey);
 439             //4. if flowId present:
 440             if (flowDescriptor != null) {
 441                 // a) construct flow path
 442                 final KeyedInstanceIdentifier<Flow, FlowKey> flowPath = getDeviceState().getNodeInstanceIdentifier()
 443                         .augmentation(FlowCapableNode.class)
 444                         .child(Table.class, flowDescriptor.getTableKey())
 445                         .child(Flow.class, new FlowKey(flowDescriptor.getFlowId()));
 446                 // b) notify listener
 447                 itemLifecycleListener.onRemoved(flowPath);
 448             } else {
 449                 LOG.debug("flow id not found: nodeId={} tableId={}, priority={}",
 450                         getDeviceState().getNodeId(), flowRegKey.getTableId(), flowRemovedNotification.getPriority());
 451             }
 452         }
 453     }
 454
 455     @Override
 456     public void processPortStatusMessage(final PortStatusMessage portStatus) {
 457         messageSpy.spyMessage(portStatus.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
 458         final FlowCapableNodeConnector flowCapableNodeConnector = portStatusTranslator.translate(portStatus, this, null);
 459
 460         final KeyedInstanceIdentifier<NodeConnector, NodeConnectorKey> iiToNodeConnector = provideIIToNodeConnector(portStatus.getPortNo(), portStatus.getVersion());
 461         if (portStatus.getReason().equals(PortReason.OFPPRADD) || portStatus.getReason().equals(PortReason.OFPPRMODIFY)) {
 462             // because of ADD status node connector has to be created
 463             final NodeConnectorBuilder nConnectorBuilder = new NodeConnectorBuilder().setKey(iiToNodeConnector.getKey());
 464             nConnectorBuilder.addAugmentation(FlowCapableNodeConnectorStatisticsData.class, new FlowCapableNodeConnectorStatisticsDataBuilder().build());
 465             nConnectorBuilder.addAugmentation(FlowCapableNodeConnector.class, flowCapableNodeConnector);
 466             writeToTransaction(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector, nConnectorBuilder.build());
 467         } else if (portStatus.getReason().equals(PortReason.OFPPRDELETE)) {
 468             addDeleteToTxChain(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector);
 469         }
 470         submitTransaction();
 471     }
 472
 473     private KeyedInstanceIdentifier<NodeConnector, NodeConnectorKey> provideIIToNodeConnector(final long portNo, final short version) {
 474         final InstanceIdentifier<Node> iiToNodes = deviceState.getNodeInstanceIdentifier();
 475         final BigInteger dataPathId = deviceState.getFeatures().getDatapathId();
 476         final NodeConnectorId nodeConnectorId = NodeStaticReplyTranslatorUtil.nodeConnectorId(dataPathId.toString(), portNo, version);
 477         return iiToNodes.child(NodeConnector.class, new NodeConnectorKey(nodeConnectorId));
 478     }
 479
 480     @Override
 481     public void processPacketInMessage(final PacketInMessage packetInMessage) {
 482         messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH);
 483         final ConnectionAdapter connectionAdapter = getPrimaryConnectionContext().getConnectionAdapter();
 484         final PacketReceived packetReceived = packetInTranslator.translate(packetInMessage, this, null);
 485
 486         if (packetReceived == null) {
 487             LOG.debug("Received a null packet from switch {}", connectionAdapter.getRemoteAddress());
 488             messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_TRANSLATE_SRC_FAILURE);
 489             return;
 490         } else {
 491             messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_TRANSLATE_OUT_SUCCESS);
 492         }
 493
 494         if (!packetInLimiter.acquirePermit()) {
 495             LOG.debug("Packet limited");
 496             // TODO: save packet into emergency slot if possible
 497             messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PACKET_IN_LIMIT_REACHED_AND_DROPPED);
 498             return;
 499         }
 500
 501         final ListenableFuture<? extends Object> offerNotification = notificationPublishService.offerNotification(packetReceived);
 502         if (NotificationPublishService.REJECTED.equals(offerNotification)) {
 503             LOG.debug("notification offer rejected");
 504             messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_NOTIFICATION_REJECTED);
 505             packetInLimiter.drainLowWaterMark();
 506             packetInLimiter.releasePermit();
 507             return;
 508         }
 509
 510         Futures.addCallback(offerNotification, new FutureCallback<Object>() {
 511             @Override
 512             public void onSuccess(final Object result) {
 513                 messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
 514                 packetInLimiter.releasePermit();
 515             }
 516
 517             @Override
 518             public void onFailure(final Throwable t) {
 519                 messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_NOTIFICATION_REJECTED);
 520                 LOG.debug("notification offer failed: {}", t.getMessage());
 521                 LOG.trace("notification offer failed..", t);
 522                 packetInLimiter.releasePermit();
 523             }
 524         });
 525     }
 526
 527     @Override
 528     public void processExperimenterMessage(final ExperimenterMessage notification) {
 529         // lookup converter
 530         final ExperimenterDataOfChoice vendorData = notification.getExperimenterDataOfChoice();
 531         final MessageTypeKey<? extends ExperimenterDataOfChoice> key = new MessageTypeKey<>(
 532                 deviceState.getVersion(),
 533                 (Class<? extends ExperimenterDataOfChoice>) vendorData.getImplementedInterface());
 534         final ConvertorMessageFromOFJava<ExperimenterDataOfChoice, MessagePath> messageConverter = extensionConverterProvider.getMessageConverter(key);
 535         if (messageConverter == null) {
 536             LOG.warn("custom converter for {}[OF:{}] not found",
 537                     notification.getExperimenterDataOfChoice().getImplementedInterface(),
 538                     deviceState.getVersion());
 539             return;
 540         }
 541         // build notification
 542         final ExperimenterMessageOfChoice messageOfChoice;
 543         try {
 544             messageOfChoice = messageConverter.convert(vendorData, MessagePath.MESSAGE_NOTIFICATION);
 545             final ExperimenterMessageFromDevBuilder experimenterMessageFromDevBld = new ExperimenterMessageFromDevBuilder()
 546                 .setNode(new NodeRef(deviceState.getNodeInstanceIdentifier()))
 547                     .setExperimenterMessageOfChoice(messageOfChoice);
 548             // publish
 549             notificationPublishService.offerNotification(experimenterMessageFromDevBld.build());
 550         } catch (final ConversionException e) {
 551             LOG.warn("Conversion of experimenter notification failed", e);
 552         }
 553     }
 554
 555     @Override
 556     public TranslatorLibrary oook() {
 557         return translatorLibrary;
 558     }
 559
 560     @Override
 561     public HashedWheelTimer getTimer() {
 562         return hashedWheelTimer;
 563     }
 564
 565     @Override
 566     public synchronized void close() {
 567         LOG.debug("closing deviceContext: {}, nodeId:{}",
 568                 getPrimaryConnectionContext().getConnectionAdapter().getRemoteAddress(),
 569                 getDeviceState().getNodeId());
 570
 571         if (deviceState.isValid()) {
 572             primaryConnectionContext.closeConnection(false);
 573             tearDown();
 574         }
 575     }
 576
 577     private synchronized void tearDown() {
 578         LOG.trace("tearDown method for node {}", deviceState.getNodeId());
 579         if (deviceState.isValid()) {
 580             deviceState.setValid(false);
 581
 582             for (final ConnectionContext connectionContext : auxiliaryConnectionContexts.values()) {
 583                 connectionContext.closeConnection(false);
 584             }
 585
 586             deviceGroupRegistry.close();
 587             deviceFlowRegistry.close();
 588             deviceMeterRegistry.close();
 589
 590             final ListenableFuture<Void> future = transactionChainManager.shuttingDown();
 591             Futures.addCallback(future, new FutureCallback<Void>() {
 592
 593                 @Override
 594                 public void onSuccess(final Void result) {
 595                     LOG.info("TxChain {} was shutdown successful.", getDeviceState().getNodeId());
 596                     tearDownClean();
 597                 }
 598
 599                 @Override
 600                 public void onFailure(final Throwable t) {
 601                     LOG.warn("Shutdown TxChain {} fail.", getDeviceState().getNodeId(), t);
 602                     tearDownClean();
 603                 }
 604             });
 605         }
 606     }
 607
 608     private void tearDownClean() {
 609         LOG.info("Closing transaction chain manager without cleaning inventory operational");
 610         transactionChainManager.close();
 611
 612         final LinkedList<DeviceTerminationPhaseHandler> reversedCloseHandlers = new LinkedList<>(closeHandlers);
 613         Collections.reverse(reversedCloseHandlers);
 614         for (final DeviceTerminationPhaseHandler deviceContextClosedHandler : reversedCloseHandlers) {
 615             deviceContextClosedHandler.onDeviceContextLevelDown(this);
 616         }
 617         closeHandlers.clear();
 618     }
 619
 620     @Override
 621     public void onDeviceDisconnected(final ConnectionContext connectionContext) {
 622         LOG.info("ConnectionEvent: Device disconnected from controller, Device:{}, NodeId:{}",
 623                 connectionContext.getConnectionAdapter().getRemoteAddress(), connectionContext.getNodeId());
 624         if (getPrimaryConnectionContext().equals(connectionContext)) {
 625             try {
 626                 tearDown();
 627             } catch (final Exception e) {
 628                 LOG.trace("Error closing device context.");
 629             }
 630         } else {
 631             LOG.debug("auxiliary connection dropped: {}, nodeId:{}",
 632                     connectionContext.getConnectionAdapter().getRemoteAddress(),
 633                     getDeviceState().getNodeId());
 634             final SwitchConnectionDistinguisher connectionDistinguisher = createConnectionDistinguisher(connectionContext);
 635             auxiliaryConnectionContexts.remove(connectionDistinguisher);
 636         }
 637     }
 638
 639     @Override
 640     public void setCurrentBarrierTimeout(final Timeout timeout) {
 641         barrierTaskTimeout = timeout;
 642     }
 643
 644     @Override
 645     public Timeout getBarrierTaskTimeout() {
 646         return barrierTaskTimeout;
 647     }
 648
 649     @Override
 650     public void setNotificationService(final NotificationService notificationService) {
 651         this.notificationService = notificationService;
 652     }
 653
 654     @Override
 655     public void setNotificationPublishService(final NotificationPublishService notificationPublishService) {
 656         this.notificationPublishService = notificationPublishService;
 657     }
 658
 659     @Override
 660     public MessageSpy getMessageSpy() {
 661         return messageSpy;
 662     }
 663
 664     @Override
 665     public void addDeviceContextClosedHandler(final DeviceTerminationPhaseHandler deviceContextClosedHandler) {
 666         closeHandlers.add(deviceContextClosedHandler);
 667     }
 668
 669     @Override
 670     public void onPublished() {
 671         primaryConnectionContext.getConnectionAdapter().setPacketInFiltering(false);
 672         for (final ConnectionContext switchAuxConnectionContext : auxiliaryConnectionContexts.values()) {
 673             switchAuxConnectionContext.getConnectionAdapter().setPacketInFiltering(false);
 674         }
 675     }
 676
 677     @Override
 678     public MultiMsgCollector getMultiMsgCollector(final RequestContext<List<MultipartReply>> requestContext) {
 679         return new MultiMsgCollectorImpl(this, requestContext);
 680     }
 681
 682     @Override
 683     public NodeConnectorRef lookupNodeConnectorRef(final Long portNumber) {
 684         return nodeConnectorCache.get(portNumber);
 685     }
 686
 687     @Override
 688     public void storeNodeConnectorRef(final Long portNumber, final NodeConnectorRef nodeConnectorRef) {
 689         nodeConnectorCache.put(
 690                 Preconditions.checkNotNull(portNumber),
 691                 Preconditions.checkNotNull(nodeConnectorRef));
 692     }
 693
 694     @Override
 695     public void updatePacketInRateLimit(final long upperBound) {
 696         packetInLimiter.changeWaterMarks((int) (LOW_WATERMARK_FACTOR * upperBound), (int) (HIGH_WATERMARK_FACTOR * upperBound));
 697     }
 698
 699     @Override
 700     public ItemLifeCycleRegistry getItemLifeCycleSourceRegistry() {
 701         return itemLifeCycleSourceRegistry;
 702     }
 703
 704     @Override
 705     public void setRpcContext(final RpcContext rpcContext) {
 706         this.rpcContext = rpcContext;
 707     }
 708
 709     @Override
 710     public RpcContext getRpcContext() {
 711         return rpcContext;
 712     }
 713
 714     @Override
 715     public void setExtensionConverterProvider(final ExtensionConverterProvider extensionConverterProvider) {
 716         this.extensionConverterProvider = extensionConverterProvider;
 717     }
 718
 719     @Override
 720     public ExtensionConverterProvider getExtensionConverterProvider() {
 721         return extensionConverterProvider;
 722     }
 723
 724     @Override
 725     public void setStatisticsContext(final StatisticsContext statisticsContext) {
 726         this.statisticsContext = statisticsContext;
 727     }
 728
 729     @Override
 730     public StatisticsContext getStatisticsContext() {
 731         return statisticsContext;
 732     }
 733 }