openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/lifecycle/ContextChainHolderImpl.java

   1 /*
   2  * Copyright (c) 2016 Pantheon Technologies s.r.o. 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.lifecycle;
   9
  10 import com.google.common.annotations.VisibleForTesting;
  11 import com.google.common.base.Verify;
  12 import com.google.common.util.concurrent.FutureCallback;
  13 import com.google.common.util.concurrent.Futures;
  14 import com.google.common.util.concurrent.ListenableFuture;
  15 import io.netty.util.HashedWheelTimer;
  16 import java.util.Collections;
  17 import java.util.HashMap;
  18 import java.util.Map;
  19 import java.util.Objects;
  20 import java.util.Optional;
  21 import java.util.concurrent.ExecutorService;
  22 import java.util.concurrent.TimeUnit;
  23 import java.util.concurrent.TimeoutException;
  24 import javax.annotation.Nonnull;
  25 import javax.annotation.Nullable;
  26 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipChange;
  27 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipListenerRegistration;
  28 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipService;
  29 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
  30 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceProvider;
  31 import org.opendaylight.openflowplugin.api.openflow.OFPManager;
  32 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
  33 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionStatus;
  34 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
  35 import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo;
  36 import org.opendaylight.openflowplugin.api.openflow.device.DeviceManager;
  37 import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChain;
  38 import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChainHolder;
  39 import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChainMastershipState;
  40 import org.opendaylight.openflowplugin.api.openflow.lifecycle.LifecycleService;
  41 import org.opendaylight.openflowplugin.api.openflow.rpc.RpcContext;
  42 import org.opendaylight.openflowplugin.api.openflow.rpc.RpcManager;
  43 import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsContext;
  44 import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsManager;
  45 import org.opendaylight.openflowplugin.impl.util.DeviceStateUtil;
  46 import org.opendaylight.openflowplugin.impl.util.ItemScheduler;
  47 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
  48 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
  49 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
  50 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
  51 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  52 import org.slf4j.Logger;
  53 import org.slf4j.LoggerFactory;
  54
  55 public class ContextChainHolderImpl implements ContextChainHolder {
  56     private static final Logger LOG = LoggerFactory.getLogger(ContextChainHolderImpl.class);
  57
  58     private static final String CONTEXT_CREATED_FOR_CONNECTION = " context created for connection: {}";
  59     private static final long CHECK_ROLE_MASTER_TIMEOUT = 20000L;
  60     private static final long CHECK_ROLE_MASTER_TOLERANCE = CHECK_ROLE_MASTER_TIMEOUT / 2;
  61     private static final long REMOVE_DEVICE_FROM_DS_TIMEOUT = 5000L;
  62     private static final String ASYNC_SERVICE_ENTITY_TYPE = "org.opendaylight.mdsal.AsyncServiceCloseEntityType";
  63
  64     private final Map<DeviceInfo, ContextChain> contextChainMap = Collections.synchronizedMap(new HashMap<>());
  65     private final EntityOwnershipListenerRegistration eosListenerRegistration;
  66     private final ClusterSingletonServiceProvider singletonServiceProvider;
  67     private final ItemScheduler<DeviceInfo, ContextChain> scheduler;
  68     private final ExecutorService executorService;
  69     private DeviceManager deviceManager;
  70     private RpcManager rpcManager;
  71     private StatisticsManager statisticsManager;
  72
  73     public ContextChainHolderImpl(final HashedWheelTimer timer,
  74                                   final ExecutorService executorService,
  75                                   final ClusterSingletonServiceProvider singletonServiceProvider,
  76                                   final EntityOwnershipService entityOwnershipService) {
  77         this.singletonServiceProvider = singletonServiceProvider;
  78         this.executorService = executorService;
  79         this.eosListenerRegistration = Verify.verifyNotNull(entityOwnershipService.registerListener
  80                 (ASYNC_SERVICE_ENTITY_TYPE, this));
  81
  82         this.scheduler = new ItemScheduler<>(
  83                 timer,
  84                 CHECK_ROLE_MASTER_TIMEOUT,
  85                 CHECK_ROLE_MASTER_TOLERANCE,
  86                 ContextChain::makeDeviceSlave);
  87     }
  88
  89     @Override
  90     public <T extends OFPManager> void addManager(final T manager) {
  91         if (Objects.isNull(deviceManager) && manager instanceof DeviceManager) {
  92             LOG.trace("Context chain holder: Device manager OK.");
  93             deviceManager = (DeviceManager) manager;
  94         } else if (Objects.isNull(rpcManager) && manager instanceof RpcManager) {
  95             LOG.trace("Context chain holder: RPC manager OK.");
  96             rpcManager = (RpcManager) manager;
  97         } else if (Objects.isNull(statisticsManager) && manager instanceof StatisticsManager) {
  98             LOG.trace("Context chain holder: Statistics manager OK.");
  99             statisticsManager = (StatisticsManager) manager;
 100         }
 101     }
 102
 103     @Override
 104     public ContextChain createContextChain(final ConnectionContext connectionContext) {
 105         final DeviceInfo deviceInfo = connectionContext.getDeviceInfo();
 106         final String deviceInfoLOGValue = deviceInfo.getLOGValue();
 107         final ContextChain contextChain = new ContextChainImpl(connectionContext);
 108
 109         if (LOG.isDebugEnabled()) {
 110             LOG.debug("Context chain" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfoLOGValue);
 111         }
 112
 113         final LifecycleService lifecycleService = new LifecycleServiceImpl(this, executorService);
 114         lifecycleService.registerDeviceRemovedHandler(deviceManager);
 115         lifecycleService.registerDeviceRemovedHandler(rpcManager);
 116         lifecycleService.registerDeviceRemovedHandler(statisticsManager);
 117
 118         if (LOG.isDebugEnabled()) {
 119             LOG.debug("Lifecycle services" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfoLOGValue);
 120         }
 121
 122         final DeviceContext deviceContext = deviceManager.createContext(connectionContext);
 123
 124         if (LOG.isDebugEnabled()) {
 125             LOG.debug("Device" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfoLOGValue);
 126         }
 127
 128         final RpcContext rpcContext = rpcManager.createContext(connectionContext.getDeviceInfo(), deviceContext);
 129
 130         if (LOG.isDebugEnabled()) {
 131             LOG.debug("RPC" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfoLOGValue);
 132         }
 133
 134         final StatisticsContext statisticsContext = statisticsManager.createContext(deviceContext);
 135
 136         if (LOG.isDebugEnabled()) {
 137             LOG.debug("Statistics" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfoLOGValue);
 138         }
 139
 140         deviceContext.setLifecycleInitializationPhaseHandler(statisticsContext);
 141         statisticsContext.setLifecycleInitializationPhaseHandler(rpcContext);
 142         statisticsContext.setInitialSubmitHandler(deviceContext);
 143
 144         contextChain.addLifecycleService(lifecycleService);
 145         contextChain.addContext(deviceContext);
 146         contextChain.addContext(rpcContext);
 147         contextChain.addContext(statisticsContext);
 148
 149         LOG.info("Starting timer for setting SLAVE role on node {} if no role will be set in {}s.",
 150                 deviceInfo.getLOGValue(), CHECK_ROLE_MASTER_TIMEOUT / 1000L);
 151         scheduler.add(deviceInfo, contextChain);
 152         scheduler.startIfNotRunning();
 153         deviceContext.onPublished();
 154         contextChain.registerServices(singletonServiceProvider);
 155         return contextChain;
 156     }
 157
 158     @Override
 159     public synchronized void destroyContextChain(final DeviceInfo deviceInfo) {
 160         Optional.ofNullable(contextChainMap.remove(deviceInfo)).ifPresent(contextChain -> {
 161             deviceManager.sendNodeRemovedNotification(deviceInfo.getNodeInstanceIdentifier());
 162             contextChain.close();
 163         });
 164     }
 165
 166     @Override
 167     public ConnectionStatus deviceConnected(final ConnectionContext connectionContext) throws Exception {
 168         final DeviceInfo deviceInfo = connectionContext.getDeviceInfo();
 169         final ContextChain contextChain = contextChainMap.get(deviceInfo);
 170         LOG.info("Device {} connected.", deviceInfo.getLOGValue());
 171
 172         if (contextChain != null) {
 173             if (contextChain.addAuxiliaryConnection(connectionContext)) {
 174                 LOG.info("An auxiliary connection was added to device: {}", deviceInfo.getLOGValue());
 175                 return ConnectionStatus.MAY_CONTINUE;
 176             } else {
 177                 LOG.warn("Device {} already connected. Closing all connection to the device.", deviceInfo.getLOGValue());
 178                 destroyContextChain(deviceInfo);
 179                 return ConnectionStatus.ALREADY_CONNECTED;
 180             }
 181         } else {
 182             if (LOG.isDebugEnabled()) {
 183                 LOG.debug("No context chain found for device: {}, creating new.", deviceInfo.getLOGValue());
 184             }
 185             contextChainMap.put(deviceInfo, createContextChain(connectionContext));
 186         }
 187
 188         return ConnectionStatus.MAY_CONTINUE;
 189     }
 190
 191     @Override
 192     public void onNotAbleToStartMastership(final DeviceInfo deviceInfo, @Nonnull final String reason, final boolean mandatory) {
 193         LOG.warn("Not able to set MASTER role on device {}, reason: {}", deviceInfo.getLOGValue(), reason);
 194
 195         if (!mandatory) {
 196             return;
 197         }
 198
 199         Optional.ofNullable(contextChainMap.get(deviceInfo)).ifPresent(contextChain -> {
 200             LOG.warn("This mastering is mandatory, destroying context chain and closing connection for device {}.", deviceInfo.getLOGValue());
 201             addDestroyChainCallback(contextChain.stopChain(), deviceInfo);
 202         });
 203     }
 204
 205     @Override
 206     public void onMasterRoleAcquired(final DeviceInfo deviceInfo, @Nonnull final ContextChainMastershipState mastershipState) {
 207         scheduler.remove(deviceInfo);
 208
 209         Optional.ofNullable(contextChainMap.get(deviceInfo)).ifPresent(contextChain -> {
 210             if (contextChain.isMastered(mastershipState)) {
 211                 LOG.info("Role MASTER was granted to device {}", deviceInfo.getLOGValue());
 212                 deviceManager.sendNodeAddedNotification(deviceInfo.getNodeInstanceIdentifier());
 213             }
 214         });
 215     }
 216
 217     @Override
 218     public void onSlaveRoleAcquired(final DeviceInfo deviceInfo) {
 219         Optional.ofNullable(contextChainMap.get(deviceInfo)).ifPresent(ContextChain::makeContextChainStateSlave);
 220     }
 221
 222     @Override
 223     public void onSlaveRoleNotAcquired(final DeviceInfo deviceInfo) {
 224         Optional.ofNullable(contextChainMap.get(deviceInfo)).ifPresent(contextChain -> destroyContextChain(deviceInfo));
 225     }
 226
 227     @Override
 228     public void onDeviceDisconnected(final ConnectionContext connectionContext) {
 229         final DeviceInfo deviceInfo = connectionContext.getDeviceInfo();
 230
 231         if (Objects.isNull(deviceInfo)) {
 232             return;
 233         }
 234
 235         Optional.ofNullable(contextChainMap.get(deviceInfo)).ifPresent(contextChain -> {
 236             if (contextChain.auxiliaryConnectionDropped(connectionContext)) {
 237                 LOG.info("Auxiliary connection from device {} disconnected.", deviceInfo.getLOGValue());
 238             } else {
 239                 LOG.info("Device {} disconnected.", deviceInfo.getLOGValue());
 240                 addDestroyChainCallback(contextChain.connectionDropped(), deviceInfo);
 241             }
 242         });
 243     }
 244
 245     @VisibleForTesting
 246     boolean checkAllManagers() {
 247         return Objects.nonNull(deviceManager) && Objects.nonNull(rpcManager) && Objects.nonNull(statisticsManager);
 248     }
 249
 250     @Override
 251     public void close() throws Exception {
 252         scheduler.close();
 253
 254         contextChainMap.forEach((deviceInfo, contextChain) -> {
 255             if (contextChain.isMastered(ContextChainMastershipState.CHECK)) {
 256                 addDestroyChainCallback(contextChain.stopChain(), deviceInfo);
 257             } else {
 258                 destroyContextChain(deviceInfo);
 259             }
 260         });
 261
 262         contextChainMap.clear();
 263         eosListenerRegistration.close();
 264     }
 265
 266     @Override
 267     public void ownershipChanged(EntityOwnershipChange entityOwnershipChange) {
 268         if (entityOwnershipChange.hasOwner()) {
 269             return;
 270         }
 271
 272         final String entityName = getEntityNameFromOwnershipChange(entityOwnershipChange);
 273
 274         if (Objects.nonNull(entityName)) {
 275             if (LOG.isDebugEnabled()) {
 276                 LOG.debug("Entity {} has no owner", entityName);
 277             }
 278
 279             final NodeId nodeId = new NodeId(entityName);
 280
 281             try {
 282                 final KeyedInstanceIdentifier<Node, NodeKey> nodeInstanceIdentifier =
 283                         DeviceStateUtil.createNodeInstanceIdentifier(nodeId);
 284
 285                 deviceManager.sendNodeRemovedNotification(nodeInstanceIdentifier);
 286
 287                 LOG.info("Removing device {} from operational DS", nodeId);
 288                 deviceManager
 289                         .removeDeviceFromOperationalDS(nodeInstanceIdentifier)
 290                         .checkedGet(REMOVE_DEVICE_FROM_DS_TIMEOUT, TimeUnit.MILLISECONDS);
 291             } catch (TimeoutException | TransactionCommitFailedException e) {
 292                 LOG.info("Not able to remove device {} from operational DS. Probably removed by another cluster node.",
 293                         nodeId);
 294             }
 295         }
 296     }
 297
 298     private String getEntityNameFromOwnershipChange(final EntityOwnershipChange entityOwnershipChange) {
 299         final YangInstanceIdentifier.NodeIdentifierWithPredicates lastIdArgument =
 300                 (YangInstanceIdentifier.NodeIdentifierWithPredicates) entityOwnershipChange
 301                         .getEntity()
 302                         .getId()
 303                         .getLastPathArgument();
 304
 305         return lastIdArgument
 306                 .getKeyValues()
 307                 .values()
 308                 .iterator()
 309                 .next()
 310                 .toString();
 311     }
 312
 313     private void addDestroyChainCallback(final ListenableFuture<Void> future, final DeviceInfo deviceInfo) {
 314         scheduler.remove(deviceInfo);
 315
 316         Futures.addCallback(future, new FutureCallback<Void>() {
 317             @Override
 318             public void onSuccess(@Nullable final Void aVoid) {
 319                 destroyContextChain(deviceInfo);
 320             }
 321
 322             @Override
 323             public void onFailure(@Nonnull final Throwable throwable) {
 324                 destroyContextChain(deviceInfo);
 325             }
 326         });
 327     }
 328 }