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.util.concurrent.FutureCallback;
  12 import com.google.common.util.concurrent.Futures;
  13 import com.google.common.util.concurrent.MoreExecutors;
  14 import java.util.HashMap;
  15 import java.util.List;
  16 import java.util.Map;
  17 import java.util.Objects;
  18 import java.util.Optional;
  19 import java.util.concurrent.ConcurrentHashMap;
  20 import java.util.concurrent.ExecutionException;
  21 import java.util.concurrent.ExecutorService;
  22 import java.util.concurrent.TimeUnit;
  23 import java.util.concurrent.TimeoutException;
  24 import java.util.stream.Collectors;
  25 import javax.annotation.Nonnull;
  26 import javax.annotation.Nullable;
  27 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipChange;
  28 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipListenerRegistration;
  29 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipService;
  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.MasterChecker;
  41 import org.opendaylight.openflowplugin.api.openflow.lifecycle.OwnershipChangeListener;
  42 import org.opendaylight.openflowplugin.api.openflow.role.RoleContext;
  43 import org.opendaylight.openflowplugin.api.openflow.role.RoleManager;
  44 import org.opendaylight.openflowplugin.api.openflow.rpc.RpcContext;
  45 import org.opendaylight.openflowplugin.api.openflow.rpc.RpcManager;
  46 import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsContext;
  47 import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsManager;
  48 import org.opendaylight.openflowplugin.impl.util.DeviceStateUtil;
  49 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
  50 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
  51 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
  52 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflowplugin.rf.state.rev170713.ResultState;
  53 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
  54 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  55 import org.slf4j.Logger;
  56 import org.slf4j.LoggerFactory;
  57
  58 public class ContextChainHolderImpl implements ContextChainHolder, MasterChecker {
  59     private static final Logger LOG = LoggerFactory.getLogger(ContextChainHolderImpl.class);
  60
  61     private static final String CONTEXT_CREATED_FOR_CONNECTION = " context created for connection: {}";
  62     private static final long REMOVE_DEVICE_FROM_DS_TIMEOUT = 5000L;
  63     private static final String ASYNC_SERVICE_ENTITY_TYPE = "org.opendaylight.mdsal.AsyncServiceCloseEntityType";
  64
  65     private final Map<DeviceInfo, ContextChain> contextChainMap = new ConcurrentHashMap<>();
  66     private final Map<DeviceInfo, ? super ConnectionContext> connectingDevices = new ConcurrentHashMap<>();
  67     private final EntityOwnershipListenerRegistration eosListenerRegistration;
  68     private final ClusterSingletonServiceProvider singletonServiceProvider;
  69     private final ExecutorService executorService;
  70     private final OwnershipChangeListener ownershipChangeListener;
  71     private DeviceManager deviceManager;
  72     private RpcManager rpcManager;
  73     private StatisticsManager statisticsManager;
  74     private RoleManager roleManager;
  75
  76     public ContextChainHolderImpl(final ExecutorService executorService,
  77                                   final ClusterSingletonServiceProvider singletonServiceProvider,
  78                                   final EntityOwnershipService entityOwnershipService,
  79                                   final OwnershipChangeListener ownershipChangeListener) {
  80         this.singletonServiceProvider = singletonServiceProvider;
  81         this.executorService = executorService;
  82         this.ownershipChangeListener = ownershipChangeListener;
  83         this.ownershipChangeListener.setMasterChecker(this);
  84         this.eosListenerRegistration = Objects.requireNonNull(entityOwnershipService
  85                 .registerListener(ASYNC_SERVICE_ENTITY_TYPE, this));
  86     }
  87
  88     @Override
  89     public <T extends OFPManager> void addManager(final T manager) {
  90         if (Objects.isNull(deviceManager) && manager instanceof DeviceManager) {
  91             LOG.trace("Context chain holder: Device manager OK.");
  92             deviceManager = (DeviceManager) manager;
  93         } else if (Objects.isNull(rpcManager) && manager instanceof RpcManager) {
  94             LOG.trace("Context chain holder: RPC manager OK.");
  95             rpcManager = (RpcManager) manager;
  96         } else if (Objects.isNull(statisticsManager) && manager instanceof StatisticsManager) {
  97             LOG.trace("Context chain holder: Statistics manager OK.");
  98             statisticsManager = (StatisticsManager) manager;
  99         } else if (Objects.isNull(roleManager) && manager instanceof RoleManager) {
 100             LOG.trace("Context chain holder: Role manager OK.");
 101             roleManager = (RoleManager) manager;
 102         }
 103     }
 104
 105     @VisibleForTesting
 106     void createContextChain(final ConnectionContext connectionContext) {
 107         final DeviceInfo deviceInfo = connectionContext.getDeviceInfo();
 108
 109         final DeviceContext deviceContext = deviceManager.createContext(connectionContext);
 110         deviceContext.registerMastershipWatcher(this);
 111         LOG.debug("Device" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfo);
 112
 113         final RpcContext rpcContext = rpcManager.createContext(deviceContext);
 114         rpcContext.registerMastershipWatcher(this);
 115         LOG.debug("RPC" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfo);
 116
 117         final StatisticsContext statisticsContext = statisticsManager.createContext(deviceContext,
 118                 ownershipChangeListener.isReconciliationFrameworkRegistered());
 119         statisticsContext.registerMastershipWatcher(this);
 120         LOG.debug("Statistics" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfo);
 121
 122         final RoleContext roleContext = roleManager.createContext(deviceContext);
 123         roleContext.registerMastershipWatcher(this);
 124         LOG.debug("Role" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfo);
 125
 126         final ContextChain contextChain = new ContextChainImpl(this, connectionContext,
 127                 executorService);
 128         contextChain.registerDeviceRemovedHandler(deviceManager);
 129         contextChain.registerDeviceRemovedHandler(rpcManager);
 130         contextChain.registerDeviceRemovedHandler(statisticsManager);
 131         contextChain.registerDeviceRemovedHandler(roleManager);
 132         contextChain.registerDeviceRemovedHandler(this);
 133         contextChain.addContext(deviceContext);
 134         contextChain.addContext(rpcContext);
 135         contextChain.addContext(statisticsContext);
 136         contextChain.addContext(roleContext);
 137         contextChainMap.put(deviceInfo, contextChain);
 138         connectingDevices.remove(deviceInfo);
 139         LOG.debug("Context chain" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfo);
 140
 141         deviceContext.onPublished();
 142         contextChain.registerServices(singletonServiceProvider);
 143     }
 144
 145     @Override
 146     public ConnectionStatus deviceConnected(final ConnectionContext connectionContext) throws Exception {
 147
 148         final DeviceInfo deviceInfo = connectionContext.getDeviceInfo();
 149         final ContextChain contextChain = contextChainMap.get(deviceInfo);
 150
 151         if (connectionContext.getFeatures().getAuxiliaryId() != 0) {
 152             if (contextChain == null) {
 153                 LOG.warn("An auxiliary connection for device {}, but no primary connection. Refusing connection.",
 154                         deviceInfo);
 155                 return ConnectionStatus.REFUSING_AUXILIARY_CONNECTION;
 156             } else {
 157                 if (contextChain.addAuxiliaryConnection(connectionContext)) {
 158                     LOG.info("An auxiliary connection was added to device: {}", deviceInfo);
 159                     return ConnectionStatus.MAY_CONTINUE;
 160                 } else {
 161                     LOG.warn("Not able to add auxiliary connection to the device {}", deviceInfo);
 162                     return ConnectionStatus.REFUSING_AUXILIARY_CONNECTION;
 163                 }
 164             }
 165         } else {
 166             LOG.info("Device {} connected.", deviceInfo);
 167             final boolean contextExists = contextChain != null;
 168             final boolean isClosing = contextExists && contextChain.isClosing();
 169
 170             if (!isClosing && connectingDevices.putIfAbsent(deviceInfo, connectionContext) != null) {
 171                 LOG.warn("Device {} is already trying to connect, wait until succeeded or disconnected.", deviceInfo);
 172                 return ConnectionStatus.ALREADY_CONNECTED;
 173             }
 174
 175             if (contextExists) {
 176                 if (isClosing) {
 177                     LOG.warn("Device {} is already in termination state, closing all incoming connections.",
 178                             deviceInfo);
 179                     return ConnectionStatus.CLOSING;
 180                 }
 181
 182                 LOG.warn("Device {} already connected. Closing previous connection", deviceInfo);
 183                 destroyContextChain(deviceInfo);
 184                 LOG.info("Old connection dropped, creating new context chain for device {}", deviceInfo);
 185                 createContextChain(connectionContext);
 186             } else {
 187                 LOG.info("No context chain found for device: {}, creating new.", deviceInfo);
 188                 createContextChain(connectionContext);
 189             }
 190
 191             return ConnectionStatus.MAY_CONTINUE;
 192         }
 193
 194     }
 195
 196     @Override
 197     public void onNotAbleToStartMastership(@Nonnull final DeviceInfo deviceInfo,
 198                                            @Nonnull final String reason,
 199                                            final boolean mandatory) {
 200         LOG.warn("Not able to set MASTER role on device {}, reason: {}", deviceInfo, reason);
 201
 202         if (!mandatory) {
 203             return;
 204         }
 205
 206         Optional.ofNullable(contextChainMap.get(deviceInfo)).ifPresent(contextChain -> {
 207             LOG.warn("This mastering is mandatory, destroying context chain and closing connection for device {}.",
 208                     deviceInfo);
 209             destroyContextChain(deviceInfo);
 210         });
 211     }
 212
 213     @Override
 214     public void onMasterRoleAcquired(@Nonnull final DeviceInfo deviceInfo,
 215                                      @Nonnull final ContextChainMastershipState mastershipState) {
 216         Optional.ofNullable(contextChainMap.get(deviceInfo)).ifPresent(contextChain -> {
 217             if (ownershipChangeListener.isReconciliationFrameworkRegistered() &&
 218                     !ContextChainMastershipState.INITIAL_SUBMIT.equals(mastershipState)) {
 219                 if (contextChain.isMastered(mastershipState, true)) {
 220                     Futures.addCallback(
 221                             ownershipChangeListener.becomeMasterBeforeSubmittedDS(deviceInfo),
 222                             reconciliationFrameworkCallback(deviceInfo, contextChain),
 223                             MoreExecutors.directExecutor());
 224                 }
 225             } else if (contextChain.isMastered(mastershipState, false)) {
 226                 LOG.info("Role MASTER was granted to device {}", deviceInfo);
 227                 ownershipChangeListener.becomeMaster(deviceInfo);
 228                 deviceManager.sendNodeAddedNotification(deviceInfo.getNodeInstanceIdentifier());
 229             }
 230         });
 231     }
 232
 233     @Override
 234     public void onSlaveRoleAcquired(final DeviceInfo deviceInfo) {
 235         ownershipChangeListener.becomeSlaveOrDisconnect(deviceInfo);
 236         LOG.info("Role SLAVE was granted to device {}", deviceInfo);
 237         Optional.ofNullable(contextChainMap.get(deviceInfo)).ifPresent(ContextChain::makeContextChainStateSlave);
 238     }
 239
 240     @Override
 241     public void onSlaveRoleNotAcquired(final DeviceInfo deviceInfo, final String reason) {
 242         LOG.warn("Not able to set SLAVE role on device {}, reason: {}", deviceInfo, reason);
 243         Optional.ofNullable(contextChainMap.get(deviceInfo)).ifPresent(contextChain -> destroyContextChain(deviceInfo));
 244     }
 245
 246     @Override
 247     public void onDeviceDisconnected(final ConnectionContext connectionContext) {
 248         final DeviceInfo deviceInfo = connectionContext.getDeviceInfo();
 249
 250         Optional.ofNullable(connectionContext.getDeviceInfo())
 251                 .map(contextChainMap::get)
 252                 .ifPresent(contextChain -> {
 253                     if (contextChain.auxiliaryConnectionDropped(connectionContext)) {
 254                         LOG.info("Auxiliary connection from device {} disconnected.", deviceInfo);
 255                     } else {
 256                         LOG.info("Device {} disconnected.", deviceInfo);
 257                         destroyContextChain(deviceInfo);
 258                     }
 259                 });
 260     }
 261
 262     @VisibleForTesting
 263     boolean checkAllManagers() {
 264         return Objects.nonNull(deviceManager)
 265                 && Objects.nonNull(rpcManager)
 266                 && Objects.nonNull(statisticsManager)
 267                 && Objects.nonNull(roleManager);
 268     }
 269
 270     @Override
 271     public void close() throws Exception {
 272         Map<DeviceInfo, ContextChain> copyOfChains = new HashMap<>(contextChainMap);
 273         copyOfChains.keySet().forEach(this::destroyContextChain);
 274         copyOfChains.clear();
 275         contextChainMap.clear();
 276         eosListenerRegistration.close();
 277     }
 278
 279     @Override
 280     public void ownershipChanged(EntityOwnershipChange entityOwnershipChange) {
 281         if (entityOwnershipChange.hasOwner()) {
 282             return;
 283         }
 284
 285         final String entityName = getEntityNameFromOwnershipChange(entityOwnershipChange);
 286
 287         if (Objects.nonNull(entityName)) {
 288             LOG.debug("Entity {} has no owner", entityName);
 289             final NodeId nodeId = new NodeId(entityName);
 290
 291             try {
 292                 final KeyedInstanceIdentifier<Node, NodeKey> nodeInstanceIdentifier =
 293                         DeviceStateUtil.createNodeInstanceIdentifier(nodeId);
 294
 295                 deviceManager.sendNodeRemovedNotification(nodeInstanceIdentifier);
 296
 297                 LOG.info("Try to remove device {} from operational DS", nodeId);
 298                 deviceManager
 299                         .removeDeviceFromOperationalDS(nodeInstanceIdentifier)
 300                         .get(REMOVE_DEVICE_FROM_DS_TIMEOUT, TimeUnit.MILLISECONDS);
 301                 LOG.info("Removing device from operational DS {} was successful", nodeId);
 302             } catch (TimeoutException | ExecutionException | NullPointerException | InterruptedException e) {
 303                 LOG.warn("Not able to remove device {} from operational DS. ",nodeId, e);
 304             }
 305         }
 306     }
 307
 308     private void destroyContextChain(final DeviceInfo deviceInfo) {
 309         ownershipChangeListener.becomeSlaveOrDisconnect(deviceInfo);
 310         Optional.ofNullable(contextChainMap.get(deviceInfo)).ifPresent(contextChain -> {
 311             deviceManager.sendNodeRemovedNotification(deviceInfo.getNodeInstanceIdentifier());
 312             contextChain.close();
 313         });
 314     }
 315
 316     @Override
 317     public List<DeviceInfo> listOfMasteredDevices() {
 318         return contextChainMap
 319                 .entrySet()
 320                 .stream()
 321                 .filter(deviceInfoContextChainEntry -> deviceInfoContextChainEntry
 322                         .getValue()
 323                         .isMastered(ContextChainMastershipState.CHECK, false))
 324                 .map(Map.Entry::getKey)
 325                 .collect(Collectors.toList());
 326     }
 327
 328     @Override
 329     public boolean isAnyDeviceMastered() {
 330         return contextChainMap
 331                 .entrySet()
 332                 .stream()
 333                 .findAny()
 334                 .filter(deviceInfoContextChainEntry -> deviceInfoContextChainEntry.getValue()
 335                         .isMastered(ContextChainMastershipState.CHECK, false))
 336                 .isPresent();
 337     }
 338
 339     private String getEntityNameFromOwnershipChange(final EntityOwnershipChange entityOwnershipChange) {
 340         final YangInstanceIdentifier.NodeIdentifierWithPredicates lastIdArgument =
 341                 (YangInstanceIdentifier.NodeIdentifierWithPredicates) entityOwnershipChange
 342                         .getEntity()
 343                         .getId()
 344                         .getLastPathArgument();
 345
 346         return lastIdArgument
 347                 .getKeyValues()
 348                 .values()
 349                 .iterator()
 350                 .next()
 351                 .toString();
 352     }
 353
 354     @Override
 355     public void onDeviceRemoved(final DeviceInfo deviceInfo) {
 356         contextChainMap.remove(deviceInfo);
 357         LOG.debug("Context chain removed for node {}", deviceInfo);
 358     }
 359
 360     private FutureCallback<ResultState> reconciliationFrameworkCallback(
 361             @Nonnull DeviceInfo deviceInfo,
 362             ContextChain contextChain) {
 363         return new FutureCallback<ResultState>() {
 364             @Override
 365             public void onSuccess(@Nullable ResultState result) {
 366                 if (ResultState.DONOTHING == result) {
 367                     LOG.info("Device {} connection is enabled by reconciliation framework.", deviceInfo);
 368                     contextChain.continueInitializationAfterReconciliation();
 369                 } else {
 370                     LOG.warn("Reconciliation framework failure for device {}", deviceInfo);
 371                     destroyContextChain(deviceInfo);
 372                 }
 373             }
 374
 375             @Override
 376             public void onFailure(@Nonnull Throwable throwable) {
 377                 LOG.warn("Reconciliation framework failure.");
 378                 destroyContextChain(deviceInfo);
 379             }
 380         };
 381     }
 382 }