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.Preconditions;
  12 import com.google.common.base.Verify;
  13 import com.google.common.util.concurrent.FutureCallback;
  14 import com.google.common.util.concurrent.Futures;
  15 import com.google.common.util.concurrent.ListenableFuture;
  16 import io.netty.util.Timeout;
  17 import java.math.BigInteger;
  18 import java.util.HashMap;
  19 import java.util.List;
  20 import java.util.Map;
  21 import java.util.Objects;
  22 import java.util.concurrent.ExecutionException;
  23 import java.util.concurrent.TimeUnit;
  24 import java.util.concurrent.atomic.AtomicInteger;
  25 import javax.annotation.Nonnull;
  26 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  27 import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
  28 import org.opendaylight.controller.md.sal.binding.api.ReadOnlyTransaction;
  29 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  30 import org.opendaylight.controller.md.sal.common.api.data.TransactionChainClosedException;
  31 import org.opendaylight.mdsal.singleton.common.api.ServiceGroupIdentifier;
  32 import org.opendaylight.openflowjava.protocol.api.connection.ConnectionAdapter;
  33 import org.opendaylight.openflowjava.protocol.api.keys.MessageTypeKey;
  34 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
  35 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
  36 import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo;
  37 import org.opendaylight.openflowplugin.api.openflow.device.DeviceManager;
  38 import org.opendaylight.openflowplugin.api.openflow.device.DeviceState;
  39 import org.opendaylight.openflowplugin.api.openflow.device.MessageTranslator;
  40 import org.opendaylight.openflowplugin.api.openflow.device.RequestContext;
  41 import org.opendaylight.openflowplugin.api.openflow.device.TranslatorLibrary;
  42 import org.opendaylight.openflowplugin.api.openflow.device.Xid;
  43 import org.opendaylight.openflowplugin.api.openflow.device.handlers.ClusterInitializationPhaseHandler;
  44 import org.opendaylight.openflowplugin.api.openflow.device.handlers.MultiMsgCollector;
  45 import org.opendaylight.openflowplugin.api.openflow.lifecycle.LifecycleService;
  46 import org.opendaylight.openflowplugin.api.openflow.md.core.SwitchConnectionDistinguisher;
  47 import org.opendaylight.openflowplugin.api.openflow.md.core.TranslatorKey;
  48 import org.opendaylight.openflowplugin.api.openflow.registry.ItemLifeCycleRegistry;
  49 import org.opendaylight.openflowplugin.api.openflow.registry.flow.DeviceFlowRegistry;
  50 import org.opendaylight.openflowplugin.api.openflow.registry.flow.FlowDescriptor;
  51 import org.opendaylight.openflowplugin.api.openflow.registry.flow.FlowRegistryKey;
  52 import org.opendaylight.openflowplugin.api.openflow.registry.group.DeviceGroupRegistry;
  53 import org.opendaylight.openflowplugin.api.openflow.registry.meter.DeviceMeterRegistry;
  54 import org.opendaylight.openflowplugin.api.openflow.rpc.ItemLifeCycleKeeper;
  55 import org.opendaylight.openflowplugin.api.openflow.rpc.listener.ItemLifecycleListener;
  56 import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.MessageSpy;
  57 import org.opendaylight.openflowplugin.extension.api.ConvertorMessageFromOFJava;
  58 import org.opendaylight.openflowplugin.extension.api.ExtensionConverterProviderKeeper;
  59 import org.opendaylight.openflowplugin.extension.api.core.extension.ExtensionConverterProvider;
  60 import org.opendaylight.openflowplugin.extension.api.exception.ConversionException;
  61 import org.opendaylight.openflowplugin.extension.api.path.MessagePath;
  62 import org.opendaylight.openflowplugin.impl.common.ItemLifeCycleSourceImpl;
  63 import org.opendaylight.openflowplugin.impl.common.NodeStaticReplyTranslatorUtil;
  64 import org.opendaylight.openflowplugin.impl.device.listener.MultiMsgCollectorImpl;
  65 import org.opendaylight.openflowplugin.impl.registry.flow.DeviceFlowRegistryImpl;
  66 import org.opendaylight.openflowplugin.impl.registry.flow.FlowRegistryKeyFactory;
  67 import org.opendaylight.openflowplugin.impl.registry.group.DeviceGroupRegistryImpl;
  68 import org.opendaylight.openflowplugin.impl.registry.meter.DeviceMeterRegistryImpl;
  69 import org.opendaylight.openflowplugin.impl.util.DeviceInitializationUtils;
  70 import org.opendaylight.openflowplugin.openflow.md.core.sal.convertor.ConvertorExecutor;
  71 import org.opendaylight.openflowplugin.openflow.md.core.session.SwitchConnectionCookieOFImpl;
  72 import org.opendaylight.yang.gen.v1.urn.opendaylight.experimenter.message.service.rev151020.ExperimenterMessageFromDevBuilder;
  73 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
  74 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNodeConnector;
  75 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
  76 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow;
  77 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowKey;
  78 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeConnectorId;
  79 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
  80 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnector;
  81 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnectorBuilder;
  82 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnectorKey;
  83 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
  84 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.common.types.rev130731.PortReason;
  85 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.Error;
  86 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.ExperimenterMessage;
  87 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.FlowRemoved;
  88 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.MultipartReply;
  89 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OfHeader;
  90 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PacketIn;
  91 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PacketInMessage;
  92 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PortGrouping;
  93 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PortStatusMessage;
  94 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.experimenter.core.ExperimenterDataOfChoice;
  95 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.experimenter.types.rev151020.experimenter.core.message.ExperimenterMessageOfChoice;
  96 import org.opendaylight.yang.gen.v1.urn.opendaylight.packet.service.rev130709.PacketReceived;
  97 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.FlowCapableNodeConnectorStatisticsData;
  98 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.FlowCapableNodeConnectorStatisticsDataBuilder;
  99 import org.opendaylight.yangtools.yang.binding.DataObject;
 100 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
 101 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
 102 import org.slf4j.Logger;
 103 import org.slf4j.LoggerFactory;
 104
 105 /**
 106  *
 107  */
 108 public class DeviceContextImpl implements DeviceContext, ExtensionConverterProviderKeeper{
 109
 110     private static final Logger LOG = LoggerFactory.getLogger(DeviceContextImpl.class);
 111
 112     // TODO: drain factor should be parametrized
 113     private static final float REJECTED_DRAIN_FACTOR = 0.25f;
 114     // TODO: low water mark factor should be parametrized
 115     private static final float LOW_WATERMARK_FACTOR = 0.75f;
 116     // TODO: high water mark factor should be parametrized
 117     private static final float HIGH_WATERMARK_FACTOR = 0.95f;
 118     private boolean initialized;
 119     private static final Long RETRY_DELAY = 100L;
 120     private static final int RETRY_COUNT = 3;
 121
 122     private ConnectionContext primaryConnectionContext;
 123     private final DeviceState deviceState;
 124     private final DataBroker dataBroker;
 125     private final Map<SwitchConnectionDistinguisher, ConnectionContext> auxiliaryConnectionContexts;
 126     private TransactionChainManager transactionChainManager;
 127     private DeviceFlowRegistry deviceFlowRegistry;
 128     private DeviceGroupRegistry deviceGroupRegistry;
 129     private DeviceMeterRegistry deviceMeterRegistry;
 130     private final PacketInRateLimiter packetInLimiter;
 131     private final MessageSpy messageSpy;
 132     private final ItemLifeCycleKeeper flowLifeCycleKeeper;
 133     private NotificationPublishService notificationPublishService;
 134     private Timeout barrierTaskTimeout;
 135     private final MessageTranslator<PortGrouping, FlowCapableNodeConnector> portStatusTranslator;
 136     private final MessageTranslator<PacketInMessage, PacketReceived> packetInTranslator;
 137     private final MessageTranslator<FlowRemoved, org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowRemoved> flowRemovedTranslator;
 138     private final TranslatorLibrary translatorLibrary;
 139     private final ItemLifeCycleRegistry itemLifeCycleSourceRegistry;
 140     private ExtensionConverterProvider extensionConverterProvider;
 141     private final DeviceManager deviceManager;
 142     private boolean skipTableFeatures;
 143     private boolean switchFeaturesMandatory;
 144     private final DeviceInfo deviceInfo;
 145     private final ConvertorExecutor convertorExecutor;
 146     private volatile CONTEXT_STATE state;
 147     private ClusterInitializationPhaseHandler clusterInitializationPhaseHandler;
 148
 149     DeviceContextImpl(
 150             @Nonnull final ConnectionContext primaryConnectionContext,
 151             @Nonnull final DataBroker dataBroker,
 152             @Nonnull final MessageSpy messageSpy,
 153             @Nonnull final TranslatorLibrary translatorLibrary,
 154             @Nonnull final DeviceManager manager,
 155             final ConvertorExecutor convertorExecutor,
 156             final boolean skipTableFeatures) {
 157         this.primaryConnectionContext = primaryConnectionContext;
 158         this.deviceInfo = primaryConnectionContext.getDeviceInfo();
 159         this.deviceState = new DeviceStateImpl();
 160         this.dataBroker = dataBroker;
 161         this.auxiliaryConnectionContexts = new HashMap<>();
 162         this.messageSpy = Preconditions.checkNotNull(messageSpy);
 163         this.deviceManager = manager;
 164
 165         this.packetInLimiter = new PacketInRateLimiter(primaryConnectionContext.getConnectionAdapter(),
 166                 /*initial*/ 1000, /*initial*/2000, this.messageSpy, REJECTED_DRAIN_FACTOR);
 167
 168         this.translatorLibrary = translatorLibrary;
 169         this.portStatusTranslator = translatorLibrary.lookupTranslator(
 170                 new TranslatorKey(deviceInfo.getVersion(), PortGrouping.class.getName()));
 171         this.packetInTranslator = translatorLibrary.lookupTranslator(
 172                 new TranslatorKey(deviceInfo.getVersion(), PacketIn.class.getName()));
 173         this.flowRemovedTranslator = translatorLibrary.lookupTranslator(
 174                 new TranslatorKey(deviceInfo.getVersion(), FlowRemoved.class.getName()));
 175
 176         this.itemLifeCycleSourceRegistry = new ItemLifeCycleRegistryImpl();
 177         this.flowLifeCycleKeeper = new ItemLifeCycleSourceImpl();
 178         this.itemLifeCycleSourceRegistry.registerLifeCycleSource(flowLifeCycleKeeper);
 179         this.state = CONTEXT_STATE.INITIALIZATION;
 180         this.convertorExecutor = convertorExecutor;
 181         this.skipTableFeatures = skipTableFeatures;
 182         this.initialized = false;
 183     }
 184
 185     @Override
 186     public void initialSubmitTransaction() {
 187         transactionChainManager.initialSubmitWriteTransaction();
 188     }
 189
 190     @Override
 191     public void addAuxiliaryConnectionContext(final ConnectionContext connectionContext) {
 192         final SwitchConnectionDistinguisher connectionDistinguisher = createConnectionDistinguisher(connectionContext);
 193         auxiliaryConnectionContexts.put(connectionDistinguisher, connectionContext);
 194     }
 195
 196     private static SwitchConnectionDistinguisher createConnectionDistinguisher(final ConnectionContext connectionContext) {
 197         return new SwitchConnectionCookieOFImpl(connectionContext.getFeatures().getAuxiliaryId());
 198     }
 199
 200     @Override
 201     public void removeAuxiliaryConnectionContext(final ConnectionContext connectionContext) {
 202         final SwitchConnectionDistinguisher connectionDistinguisher = createConnectionDistinguisher(connectionContext);
 203         LOG.debug("auxiliary connection dropped: {}, nodeId:{}", connectionContext.getConnectionAdapter()
 204                 .getRemoteAddress(), getDeviceInfo().getLOGValue());
 205         auxiliaryConnectionContexts.remove(connectionDistinguisher);
 206     }
 207
 208     @Override
 209     public DeviceState getDeviceState() {
 210         return deviceState;
 211     }
 212
 213     @Override
 214     public ReadOnlyTransaction getReadTransaction() {
 215         return dataBroker.newReadOnlyTransaction();
 216     }
 217
 218     @Override
 219     public <T extends DataObject> void writeToTransaction(final LogicalDatastoreType store,
 220                                                           final InstanceIdentifier<T> path,
 221                                                           final T data){
 222         if (Objects.nonNull(transactionChainManager)) {
 223             transactionChainManager.writeToTransaction(store, path, data, false);
 224         }
 225     }
 226
 227     @Override
 228     public <T extends DataObject> void writeToTransactionWithParentsSlow(final LogicalDatastoreType store,
 229                                                                          final InstanceIdentifier<T> path,
 230                                                                          final T data){
 231         if (Objects.nonNull(transactionChainManager)) {
 232             transactionChainManager.writeToTransaction(store, path, data, true);
 233         }
 234     }
 235
 236     @Override
 237     public <T extends DataObject> void addDeleteToTxChain(final LogicalDatastoreType store, final InstanceIdentifier<T> path) {
 238         if (Objects.nonNull(transactionChainManager)) {
 239             transactionChainManager.addDeleteOperationTotTxChain(store, path);
 240         }
 241     }
 242
 243     @Override
 244     public boolean submitTransaction() {
 245         return Objects.nonNull(transactionChainManager) && transactionChainManager.submitWriteTransaction();
 246     }
 247
 248     @Override
 249     public ConnectionContext getPrimaryConnectionContext() {
 250         return primaryConnectionContext;
 251     }
 252
 253     @Override
 254     public ConnectionContext getAuxiliaryConnectionContexts(final BigInteger cookie) {
 255         return auxiliaryConnectionContexts.get(new SwitchConnectionCookieOFImpl(cookie.longValue()));
 256     }
 257
 258     @Override
 259     public DeviceFlowRegistry getDeviceFlowRegistry() {
 260         return deviceFlowRegistry;
 261     }
 262
 263     @Override
 264     public DeviceGroupRegistry getDeviceGroupRegistry() {
 265         return deviceGroupRegistry;
 266     }
 267
 268     @Override
 269     public DeviceMeterRegistry getDeviceMeterRegistry() {
 270         return deviceMeterRegistry;
 271     }
 272
 273     @Override
 274     public void processReply(final OfHeader ofHeader) {
 275         if (ofHeader instanceof Error) {
 276             messageSpy.spyMessage(ofHeader.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
 277         } else {
 278             messageSpy.spyMessage(ofHeader.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
 279         }
 280     }
 281
 282     @Override
 283     public void processReply(final Xid xid, final List<MultipartReply> ofHeaderList) {
 284         for (final MultipartReply multipartReply : ofHeaderList) {
 285             messageSpy.spyMessage(multipartReply.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
 286         }
 287     }
 288
 289     @Override
 290     public void processFlowRemovedMessage(final FlowRemoved flowRemoved) {
 291         //1. translate to general flow (table, priority, match, cookie)
 292         final org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowRemoved flowRemovedNotification =
 293                 flowRemovedTranslator.translate(flowRemoved, deviceInfo, null);
 294
 295         if(!deviceManager.getIsNotificationFlowRemovedOff()) {
 296             // Trigger off a notification
 297             notificationPublishService.offerNotification(flowRemovedNotification);
 298         } else if(LOG.isDebugEnabled()) {
 299             LOG.debug("For nodeId={} isNotificationFlowRemovedOff={}", getDeviceInfo().getLOGValue(), deviceManager.getIsNotificationFlowRemovedOff());
 300         }
 301
 302         final ItemLifecycleListener itemLifecycleListener = flowLifeCycleKeeper.getItemLifecycleListener();
 303         if (itemLifecycleListener != null) {
 304             //2. create registry key
 305             final FlowRegistryKey flowRegKey = FlowRegistryKeyFactory.create(flowRemovedNotification);
 306             //3. lookup flowId
 307             final FlowDescriptor flowDescriptor = deviceFlowRegistry.retrieveIdForFlow(flowRegKey);
 308             //4. if flowId present:
 309             if (flowDescriptor != null) {
 310                 // a) construct flow path
 311                 final KeyedInstanceIdentifier<Flow, FlowKey> flowPath = getDeviceInfo().getNodeInstanceIdentifier()
 312                         .augmentation(FlowCapableNode.class)
 313                         .child(Table.class, flowDescriptor.getTableKey())
 314                         .child(Flow.class, new FlowKey(flowDescriptor.getFlowId()));
 315                 // b) notify listener
 316                 itemLifecycleListener.onRemoved(flowPath);
 317             } else {
 318                 LOG.debug("flow id not found: nodeId={} tableId={}, priority={}",
 319                         getDeviceInfo().getNodeId(), flowRegKey.getTableId(), flowRemovedNotification.getPriority());
 320             }
 321         }
 322     }
 323
 324     @Override
 325     public void processPortStatusMessage(final PortStatusMessage portStatus) {
 326         messageSpy.spyMessage(portStatus.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
 327         try {
 328             updatePortStatusMessage(portStatus);
 329         } catch (final Exception e) {
 330             LOG.warn("Error processing port status message for port {} on device {} : {}", portStatus.getPortNo().toString(),
 331                     getDeviceInfo().getNodeId().toString(), e);
 332             retryProcessPortStatusMessage(portStatus,RETRY_COUNT);
 333         }
 334     }
 335
 336     private void retryProcessPortStatusMessage(final PortStatusMessage portStatus, final int retryCount){
 337         Thread thread = new Thread(new PortStatusRetryRunnable(portStatus, retryCount));
 338         thread.start();
 339     }
 340
 341     private class PortStatusRetryRunnable implements Runnable {
 342         private final PortStatusMessage portStatusMessage;
 343         private final int retryCount;
 344         private int count ;
 345
 346         PortStatusRetryRunnable (final PortStatusMessage portStatus,final int retry){
 347             portStatusMessage = portStatus;
 348             retryCount = retry;
 349             count = 0;
 350         }
 351         @Override
 352         public void run() {
 353             while(count < retryCount) {
 354                 try {
 355                     TimeUnit.MILLISECONDS.sleep(RETRY_DELAY);
 356                 } catch (InterruptedException e) {
 357                     LOG.warn("Caught interrupted exception while sleeping during a port retry task");
 358                 }
 359
 360                 try {
 361                     updatePortStatusMessage(portStatusMessage);
 362                     break;
 363                 } catch (final Exception e) {
 364                     LOG.warn("Error processing port status message for {} on port {} on device {} : {}",count,
 365                             portStatusMessage.getPortNo().toString(),getDeviceInfo().getNodeId().toString(), e);
 366                     count ++ ;
 367                 }
 368             }
 369             LOG.warn("Failed to update port status for port {} on device {} even after 3 retries",
 370                     portStatusMessage.getPortNo().toString(),getDeviceInfo().getNodeId().toString());
 371         }
 372     }
 373
 374     void updatePortStatusMessage(final PortStatusMessage portStatusMessage){
 375         final FlowCapableNodeConnector flowCapableNodeConnector = portStatusTranslator.translate(portStatusMessage,
 376                 getDeviceInfo(), null);
 377
 378         final KeyedInstanceIdentifier<NodeConnector, NodeConnectorKey> iiToNodeConnector =
 379                 provideIIToNodeConnector(portStatusMessage.getPortNo(), portStatusMessage.getVersion());
 380
 381         if (portStatusMessage.getReason().equals(PortReason.OFPPRADD) || portStatusMessage.getReason().
 382                 equals(PortReason.OFPPRMODIFY)) {
 383             // because of ADD status node connector has to be created
 384             final NodeConnectorBuilder nConnectorBuilder = new NodeConnectorBuilder().setKey(iiToNodeConnector.getKey());
 385             nConnectorBuilder.addAugmentation(FlowCapableNodeConnectorStatisticsData.class,
 386                     new FlowCapableNodeConnectorStatisticsDataBuilder().build());
 387             nConnectorBuilder.addAugmentation(FlowCapableNodeConnector.class, flowCapableNodeConnector);
 388             writeToTransaction(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector, nConnectorBuilder.build());
 389         } else if (portStatusMessage.getReason().equals(PortReason.OFPPRDELETE)) {
 390             addDeleteToTxChain(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector);
 391         }
 392         submitTransaction();
 393     }
 394
 395     private KeyedInstanceIdentifier<NodeConnector, NodeConnectorKey> provideIIToNodeConnector(final long portNo, final short version) {
 396         final InstanceIdentifier<Node> iiToNodes = getDeviceInfo().getNodeInstanceIdentifier();
 397         final BigInteger dataPathId = getDeviceInfo().getDatapathId();
 398         final NodeConnectorId nodeConnectorId = NodeStaticReplyTranslatorUtil.nodeConnectorId(dataPathId.toString(), portNo, version);
 399         return iiToNodes.child(NodeConnector.class, new NodeConnectorKey(nodeConnectorId));
 400     }
 401
 402     @Override
 403     public void processPacketInMessage(final PacketInMessage packetInMessage) {
 404         messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH);
 405         final ConnectionAdapter connectionAdapter = getPrimaryConnectionContext().getConnectionAdapter();
 406         final PacketReceived packetReceived = packetInTranslator.translate(packetInMessage, getDeviceInfo(), null);
 407
 408         if (packetReceived == null) {
 409             LOG.debug("Received a null packet from switch {}", connectionAdapter.getRemoteAddress());
 410             messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_TRANSLATE_SRC_FAILURE);
 411             return;
 412         } else {
 413             messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_TRANSLATE_OUT_SUCCESS);
 414         }
 415
 416         if (!packetInLimiter.acquirePermit()) {
 417             LOG.debug("Packet limited");
 418             // TODO: save packet into emergency slot if possible
 419             messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PACKET_IN_LIMIT_REACHED_AND_DROPPED);
 420             return;
 421         }
 422
 423         final ListenableFuture<?> offerNotification = notificationPublishService.offerNotification(packetReceived);
 424         if (NotificationPublishService.REJECTED.equals(offerNotification)) {
 425             LOG.debug("notification offer rejected");
 426             messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_NOTIFICATION_REJECTED);
 427             packetInLimiter.drainLowWaterMark();
 428             packetInLimiter.releasePermit();
 429             return;
 430         }
 431
 432         Futures.addCallback(offerNotification, new FutureCallback<Object>() {
 433             @Override
 434             public void onSuccess(final Object result) {
 435                 messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
 436                 packetInLimiter.releasePermit();
 437             }
 438
 439             @Override
 440             public void onFailure(final Throwable t) {
 441                 messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_NOTIFICATION_REJECTED);
 442                 LOG.debug("notification offer failed: {}", t.getMessage());
 443                 LOG.trace("notification offer failed..", t);
 444                 packetInLimiter.releasePermit();
 445             }
 446         });
 447     }
 448
 449     @Override
 450     public void processExperimenterMessage(final ExperimenterMessage notification) {
 451         // lookup converter
 452         final ExperimenterDataOfChoice vendorData = notification.getExperimenterDataOfChoice();
 453         final MessageTypeKey<? extends ExperimenterDataOfChoice> key = new MessageTypeKey<>(
 454                 getDeviceInfo().getVersion(),
 455                 (Class<? extends ExperimenterDataOfChoice>) vendorData.getImplementedInterface());
 456         final ConvertorMessageFromOFJava<ExperimenterDataOfChoice, MessagePath> messageConverter = extensionConverterProvider.getMessageConverter(key);
 457         if (messageConverter == null) {
 458             LOG.warn("custom converter for {}[OF:{}] not found",
 459                     notification.getExperimenterDataOfChoice().getImplementedInterface(),
 460                     getDeviceInfo().getVersion());
 461             return;
 462         }
 463         // build notification
 464         final ExperimenterMessageOfChoice messageOfChoice;
 465         try {
 466             messageOfChoice = messageConverter.convert(vendorData, MessagePath.MESSAGE_NOTIFICATION);
 467             final ExperimenterMessageFromDevBuilder experimenterMessageFromDevBld = new ExperimenterMessageFromDevBuilder()
 468                     .setNode(new NodeRef(getDeviceInfo().getNodeInstanceIdentifier()))
 469                     .setExperimenterMessageOfChoice(messageOfChoice);
 470             // publish
 471             notificationPublishService.offerNotification(experimenterMessageFromDevBld.build());
 472         } catch (final ConversionException e) {
 473             LOG.error("Conversion of experimenter notification failed", e);
 474         }
 475     }
 476
 477     @Override
 478     public TranslatorLibrary oook() {
 479         return translatorLibrary;
 480     }
 481
 482     @Override
 483     public synchronized void close() {
 484         LOG.debug("closing deviceContext: {}, nodeId:{}",
 485                 getPrimaryConnectionContext().getConnectionAdapter().getRemoteAddress(),
 486                 getDeviceInfo().getLOGValue());
 487         // NOOP
 488         throw new UnsupportedOperationException("Autocloseble.close will be removed soon");
 489     }
 490
 491     @Override
 492     public void setCurrentBarrierTimeout(final Timeout timeout) {
 493         barrierTaskTimeout = timeout;
 494     }
 495
 496     @Override
 497     public Timeout getBarrierTaskTimeout() {
 498         return barrierTaskTimeout;
 499     }
 500
 501     @Override
 502     public void setNotificationPublishService(final NotificationPublishService notificationPublishService) {
 503         this.notificationPublishService = notificationPublishService;
 504     }
 505
 506     @Override
 507     public MessageSpy getMessageSpy() {
 508         return messageSpy;
 509     }
 510
 511     @Override
 512     public void onPublished() {
 513         Verify.verify(CONTEXT_STATE.INITIALIZATION.equals(getState()));
 514         setState(CONTEXT_STATE.WORKING);
 515         primaryConnectionContext.getConnectionAdapter().setPacketInFiltering(false);
 516         for (final ConnectionContext switchAuxConnectionContext : auxiliaryConnectionContexts.values()) {
 517             switchAuxConnectionContext.getConnectionAdapter().setPacketInFiltering(false);
 518         }
 519     }
 520
 521     @Override
 522     public MultiMsgCollector getMultiMsgCollector(final RequestContext<List<MultipartReply>> requestContext) {
 523         return new MultiMsgCollectorImpl(this, requestContext);
 524     }
 525
 526     @Override
 527     public void updatePacketInRateLimit(final long upperBound) {
 528         packetInLimiter.changeWaterMarks((int) (LOW_WATERMARK_FACTOR * upperBound), (int) (HIGH_WATERMARK_FACTOR * upperBound));
 529     }
 530
 531     @Override
 532     public ItemLifeCycleRegistry getItemLifeCycleSourceRegistry() {
 533         return itemLifeCycleSourceRegistry;
 534     }
 535
 536     @Override
 537     public void setExtensionConverterProvider(final ExtensionConverterProvider extensionConverterProvider) {
 538         this.extensionConverterProvider = extensionConverterProvider;
 539     }
 540
 541     @Override
 542     public ExtensionConverterProvider getExtensionConverterProvider() {
 543         return extensionConverterProvider;
 544     }
 545
 546     @Override
 547     public synchronized void shutdownConnection() {
 548         if (LOG.isDebugEnabled()) {
 549             LOG.debug("Shutdown method for node {}", getDeviceInfo().getLOGValue());
 550         }
 551         if (CONTEXT_STATE.TERMINATION.equals(getState())) {
 552             LOG.debug("DeviceCtx for Node {} is in termination process.", getDeviceInfo().getLOGValue());
 553             return;
 554         }
 555
 556         if (ConnectionContext.CONNECTION_STATE.RIP.equals(getPrimaryConnectionContext().getConnectionState())) {
 557             LOG.debug("ConnectionCtx for Node {} is in RIP state.", getDeviceInfo().getLOGValue());
 558             return;
 559         }
 560         /* Terminate Auxiliary Connection */
 561         for (final ConnectionContext connectionContext : auxiliaryConnectionContexts.values()) {
 562             LOG.debug("Closing auxiliary connection {}", connectionContext.getNodeId());
 563             connectionContext.closeConnection(false);
 564         }
 565         /* Terminate Primary Connection */
 566         getPrimaryConnectionContext().closeConnection(true);
 567         /* Close all Group Registry */
 568         deviceGroupRegistry.close();
 569         deviceFlowRegistry.close();
 570         deviceMeterRegistry.close();
 571     }
 572
 573     @Override
 574     public ListenableFuture<Void> shuttingDownDataStoreTransactions() {
 575         ListenableFuture<Void> future = Futures.immediateFuture(null);
 576         if (Objects.nonNull(this.transactionChainManager)) {
 577             future = this.transactionChainManager.shuttingDown();
 578         }
 579         return future;
 580     }
 581
 582     @VisibleForTesting
 583     TransactionChainManager getTransactionChainManager() {
 584         return this.transactionChainManager;
 585     }
 586
 587     @Override
 588     public void setSwitchFeaturesMandatory(boolean switchFeaturesMandatory) {
 589         this.switchFeaturesMandatory = switchFeaturesMandatory;
 590     }
 591
 592     @Override
 593     public CONTEXT_STATE getState() {
 594         return this.state;
 595     }
 596
 597     @Override
 598     public void setState(CONTEXT_STATE state) {
 599         this.state = state;
 600     }
 601
 602     @Override
 603     public ListenableFuture<Void> stopClusterServices(boolean deviceDisconnected) {
 604         ListenableFuture<Void> future = Futures.immediateFuture(null);
 605         if (Objects.nonNull(this.transactionChainManager)) {
 606             future = this.transactionChainManager.deactivateTransactionManager();
 607         }
 608         return future;
 609     }
 610
 611     @Override
 612     public ServiceGroupIdentifier getServiceIdentifier() {
 613         return this.deviceInfo.getServiceIdentifier();
 614     }
 615
 616     @Override
 617     public DeviceInfo getDeviceInfo() {
 618         return this.deviceInfo;
 619     }
 620
 621     @Override
 622     public void putLifecycleServiceIntoTxChainManager(final LifecycleService lifecycleService){
 623         if (Objects.nonNull(this.transactionChainManager)) {
 624             this.transactionChainManager.setLifecycleService(lifecycleService);
 625         }
 626     }
 627
 628     @Override
 629     public void replaceConnectionContext(final ConnectionContext connectionContext){
 630         // Act like we are initializing the context
 631         setState(CONTEXT_STATE.INITIALIZATION);
 632         this.primaryConnectionContext = connectionContext;
 633         this.onPublished();
 634     }
 635
 636     @Override
 637     public boolean isSkipTableFeatures() {
 638         return this.skipTableFeatures;
 639     }
 640
 641     @Override
 642     public void setLifecycleInitializationPhaseHandler(final ClusterInitializationPhaseHandler handler) {
 643         this.clusterInitializationPhaseHandler = handler;
 644     }
 645
 646     @Override
 647     public boolean onContextInstantiateService(final ConnectionContext connectionContext) {
 648
 649         if (getPrimaryConnectionContext().getConnectionState().equals(ConnectionContext.CONNECTION_STATE.RIP)) {
 650             LOG.warn("Connection on device {} was interrupted, will stop starting master services.", deviceInfo.getLOGValue());
 651             return false;
 652         }
 653
 654         LOG.info("Starting device context cluster services for node {}", deviceInfo.getLOGValue());
 655
 656         lazyTransactionManagerInitialization();
 657
 658         this.transactionChainManager.activateTransactionManager();
 659
 660         try {
 661             DeviceInitializationUtils.initializeNodeInformation(this, switchFeaturesMandatory, this.convertorExecutor);
 662         } catch (ExecutionException | InterruptedException e) {
 663             LOG.warn("Device {} cannot be initialized: ", deviceInfo.getLOGValue(), e);
 664             return false;
 665         }
 666
 667         return this.clusterInitializationPhaseHandler.onContextInstantiateService(getPrimaryConnectionContext());
 668     }
 669
 670     @VisibleForTesting
 671     void lazyTransactionManagerInitialization() {
 672         if (!this.initialized) {
 673             if (LOG.isDebugEnabled()) {
 674                 LOG.debug("Transaction chain manager for node {} created", deviceInfo.getLOGValue());
 675             }
 676             this.transactionChainManager = new TransactionChainManager(dataBroker, deviceInfo);
 677             this.deviceFlowRegistry = new DeviceFlowRegistryImpl(dataBroker, deviceInfo.getNodeInstanceIdentifier());
 678             this.deviceGroupRegistry = new DeviceGroupRegistryImpl();
 679             this.deviceMeterRegistry = new DeviceMeterRegistryImpl();
 680             this.initialized = true;
 681         }
 682     }
 683 }