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 edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
  15 import java.util.HashMap;
  16 import java.util.List;
  17 import java.util.Map;
  18 import java.util.Objects;
  19 import java.util.Optional;
  20 import java.util.concurrent.ConcurrentHashMap;
  21 import java.util.concurrent.ExecutionException;
  22 import java.util.concurrent.ExecutorService;
  23 import java.util.concurrent.TimeUnit;
  24 import java.util.concurrent.TimeoutException;
  25 import java.util.stream.Collectors;
  26 import javax.annotation.Nonnull;
  27 import org.opendaylight.mdsal.eos.binding.api.EntityOwnershipChange;
  28 import org.opendaylight.mdsal.eos.binding.api.EntityOwnershipListenerRegistration;
  29 import org.opendaylight.mdsal.eos.binding.api.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.openflow.protocol.rev130731.FeaturesReply;
  53 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.mdsal.core.general.entity.rev150930.Entity;
  54 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflowplugin.rf.state.rev170713.ResultState;
  55 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
  56 import org.slf4j.Logger;
  57 import org.slf4j.LoggerFactory;
  58
  59 public class ContextChainHolderImpl implements ContextChainHolder, MasterChecker {
  60     private static final Logger LOG = LoggerFactory.getLogger(ContextChainHolderImpl.class);
  61
  62     private static final String CONTEXT_CREATED_FOR_CONNECTION = " context created for connection: {}";
  63     private static final long REMOVE_DEVICE_FROM_DS_TIMEOUT = 5000L;
  64     private static final String ASYNC_SERVICE_ENTITY_TYPE = "org.opendaylight.mdsal.AsyncServiceCloseEntityType";
  65
  66     private final Map<DeviceInfo, ContextChain> contextChainMap = new ConcurrentHashMap<>();
  67     private final Map<DeviceInfo, ? super ConnectionContext> connectingDevices = new ConcurrentHashMap<>();
  68     private final EntityOwnershipListenerRegistration eosListenerRegistration;
  69     private final ClusterSingletonServiceProvider singletonServiceProvider;
  70     private final ExecutorService executorService;
  71     private final OwnershipChangeListener ownershipChangeListener;
  72     private DeviceManager deviceManager;
  73     private RpcManager rpcManager;
  74     private StatisticsManager statisticsManager;
  75     private RoleManager roleManager;
  76
  77     public ContextChainHolderImpl(final ExecutorService executorService,
  78                                   final ClusterSingletonServiceProvider singletonServiceProvider,
  79                                   final EntityOwnershipService entityOwnershipService,
  80                                   final OwnershipChangeListener ownershipChangeListener) {
  81         this.singletonServiceProvider = singletonServiceProvider;
  82         this.executorService = executorService;
  83         this.ownershipChangeListener = ownershipChangeListener;
  84         this.ownershipChangeListener.setMasterChecker(this);
  85         this.eosListenerRegistration = Objects
  86                 .requireNonNull(entityOwnershipService.registerListener(ASYNC_SERVICE_ENTITY_TYPE, this));
  87     }
  88
  89     @Override
  90     public <T extends OFPManager> void addManager(final T manager) {
  91         if (deviceManager == null && manager instanceof DeviceManager) {
  92             LOG.trace("Context chain holder: Device manager OK.");
  93             deviceManager = (DeviceManager) manager;
  94         } else if (rpcManager == null && manager instanceof RpcManager) {
  95             LOG.trace("Context chain holder: RPC manager OK.");
  96             rpcManager = (RpcManager) manager;
  97         } else if (statisticsManager == null && manager instanceof StatisticsManager) {
  98             LOG.trace("Context chain holder: Statistics manager OK.");
  99             statisticsManager = (StatisticsManager) manager;
 100         } else if (roleManager == null && manager instanceof RoleManager) {
 101             LOG.trace("Context chain holder: Role manager OK.");
 102             roleManager = (RoleManager) manager;
 103         }
 104     }
 105
 106     @VisibleForTesting
 107     void createContextChain(final ConnectionContext connectionContext) {
 108         final DeviceInfo deviceInfo = connectionContext.getDeviceInfo();
 109
 110         final DeviceContext deviceContext = deviceManager.createContext(connectionContext);
 111         deviceContext.registerMastershipWatcher(this);
 112         LOG.debug("Device" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfo);
 113
 114         final RpcContext rpcContext = rpcManager.createContext(deviceContext);
 115         rpcContext.registerMastershipWatcher(this);
 116         LOG.debug("RPC" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfo);
 117
 118         final StatisticsContext statisticsContext = statisticsManager
 119                 .createContext(deviceContext, ownershipChangeListener.isReconciliationFrameworkRegistered());
 120         statisticsContext.registerMastershipWatcher(this);
 121         LOG.debug("Statistics" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfo);
 122
 123         final RoleContext roleContext = roleManager.createContext(deviceContext);
 124         roleContext.registerMastershipWatcher(this);
 125         LOG.debug("Role" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfo);
 126
 127         final ContextChain contextChain = new ContextChainImpl(this, connectionContext, 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) {
 147         final DeviceInfo deviceInfo = connectionContext.getDeviceInfo();
 148         final ContextChain contextChain = contextChainMap.get(deviceInfo);
 149         final FeaturesReply featuresReply = connectionContext.getFeatures();
 150         final Short auxiliaryId = featuresReply != null ? featuresReply.getAuxiliaryId() : null;
 151
 152         if (auxiliaryId != null && auxiliaryId != 0) {
 153             if (contextChain == null) {
 154                 LOG.warn("An auxiliary connection for device {}, but no primary connection. Refusing connection.",
 155                          deviceInfo);
 156                 return ConnectionStatus.REFUSING_AUXILIARY_CONNECTION;
 157             } else {
 158                 if (contextChain.addAuxiliaryConnection(connectionContext)) {
 159                     LOG.info("An auxiliary connection was added to device: {}", deviceInfo);
 160                     return ConnectionStatus.MAY_CONTINUE;
 161                 } else {
 162                     LOG.warn("Not able to add auxiliary connection to the device {}", deviceInfo);
 163                     return ConnectionStatus.REFUSING_AUXILIARY_CONNECTION;
 164                 }
 165             }
 166         } else {
 167             LOG.info("Device {} connected.", deviceInfo);
 168             final boolean contextExists = contextChain != null;
 169             final boolean isClosing = contextExists && contextChain.isClosing();
 170
 171             if (!isClosing && connectingDevices.putIfAbsent(deviceInfo, connectionContext) != null) {
 172                 LOG.warn("Device {} is already trying to connect, wait until succeeded or disconnected.", deviceInfo);
 173                 return ConnectionStatus.ALREADY_CONNECTED;
 174             }
 175
 176             if (contextExists) {
 177                 if (isClosing) {
 178                     LOG.warn("Device {} is already in termination state, closing all incoming connections.",
 179                              deviceInfo);
 180                     return ConnectionStatus.CLOSING;
 181                 }
 182
 183                 LOG.warn("Device {} already connected. Closing previous connection", deviceInfo);
 184                 destroyContextChain(deviceInfo);
 185                 LOG.info("Old connection dropped, creating new context chain for device {}", deviceInfo);
 186                 createContextChain(connectionContext);
 187             } else {
 188                 LOG.info("No context chain found for device: {}, creating new.", deviceInfo);
 189                 createContextChain(connectionContext);
 190             }
 191
 192             return ConnectionStatus.MAY_CONTINUE;
 193         }
 194
 195     }
 196
 197     @Override
 198     public void onNotAbleToStartMastership(@Nonnull final DeviceInfo deviceInfo, @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(ownershipChangeListener.becomeMasterBeforeSubmittedDS(deviceInfo),
 221                                         reconciliationFrameworkCallback(deviceInfo, contextChain),
 222                                         MoreExecutors.directExecutor());
 223                 }
 224             } else if (contextChain.isMastered(mastershipState, false)) {
 225                 LOG.info("Role MASTER was granted to device {}", deviceInfo);
 226                 ownershipChangeListener.becomeMaster(deviceInfo);
 227                 deviceManager.sendNodeAddedNotification(deviceInfo.getNodeInstanceIdentifier());
 228             }
 229         });
 230     }
 231
 232     @Override
 233     public void onSlaveRoleAcquired(final DeviceInfo deviceInfo) {
 234         ownershipChangeListener.becomeSlaveOrDisconnect(deviceInfo);
 235         LOG.info("Role SLAVE was granted to device {}", deviceInfo);
 236         Optional.ofNullable(contextChainMap.get(deviceInfo)).ifPresent(ContextChain::makeContextChainStateSlave);
 237     }
 238
 239     @Override
 240     public void onSlaveRoleNotAcquired(final DeviceInfo deviceInfo, final String reason) {
 241         LOG.warn("Not able to set SLAVE role on device {}, reason: {}", deviceInfo, reason);
 242         Optional.ofNullable(contextChainMap.get(deviceInfo)).ifPresent(contextChain -> destroyContextChain(deviceInfo));
 243     }
 244
 245     @Override
 246     public void onDeviceDisconnected(final ConnectionContext connectionContext) {
 247         final DeviceInfo deviceInfo = connectionContext.getDeviceInfo();
 248
 249         Optional.ofNullable(connectionContext.getDeviceInfo()).map(contextChainMap::get).ifPresent(contextChain -> {
 250             if (contextChain.auxiliaryConnectionDropped(connectionContext)) {
 251                 LOG.info("Auxiliary connection from device {} disconnected.", deviceInfo);
 252             } else {
 253                 LOG.info("Device {} disconnected.", deviceInfo);
 254                 destroyContextChain(deviceInfo);
 255             }
 256         });
 257     }
 258
 259     @VisibleForTesting
 260     boolean checkAllManagers() {
 261         return deviceManager != null && rpcManager != null && statisticsManager != null && roleManager != null;
 262     }
 263
 264     @Override
 265     public ContextChain getContextChain(final DeviceInfo deviceInfo) {
 266         return contextChainMap.get(deviceInfo);
 267     }
 268
 269     @Override
 270     public void close() {
 271         Map<DeviceInfo, ContextChain> copyOfChains = new HashMap<>(contextChainMap);
 272         copyOfChains.keySet().forEach(this::destroyContextChain);
 273         copyOfChains.clear();
 274         contextChainMap.clear();
 275         eosListenerRegistration.close();
 276     }
 277
 278     @Override
 279     @SuppressFBWarnings("BC_UNCONFIRMED_CAST_OF_RETURN_VALUE")
 280     public void ownershipChanged(EntityOwnershipChange entityOwnershipChange) {
 281         if (entityOwnershipChange.getState().hasOwner()) {
 282             return;
 283         }
 284
 285         // Findbugs flags a false violation for "Unchecked/unconfirmed cast" from GenericEntity to Entity hence the
 286         // suppression above. The suppression is temporary until EntityOwnershipChange is modified to eliminate the
 287         // violation.
 288         final String entityName = entityOwnershipChange
 289                 .getEntity()
 290                 .getIdentifier()
 291                 .firstKeyOf(Entity.class)
 292                 .getName();
 293
 294         if (entityName != null) {
 295             LOG.debug("Entity {} has no owner", entityName);
 296             try {
 297                 //TODO:Remove notifications
 298                 final KeyedInstanceIdentifier<Node, NodeKey> nodeInstanceIdentifier =
 299                         DeviceStateUtil.createNodeInstanceIdentifier(new NodeId(entityName));
 300                 deviceManager.sendNodeRemovedNotification(nodeInstanceIdentifier);
 301
 302                 LOG.info("Try to remove device {} from operational DS", entityName);
 303                 deviceManager.removeDeviceFromOperationalDS(nodeInstanceIdentifier)
 304                         .get(REMOVE_DEVICE_FROM_DS_TIMEOUT, TimeUnit.MILLISECONDS);
 305                 LOG.info("Removing device from operational DS {} was successful", entityName);
 306             } catch (TimeoutException | ExecutionException | NullPointerException | InterruptedException e) {
 307                 LOG.warn("Not able to remove device {} from operational DS. ", entityName, e);
 308             }
 309         }
 310     }
 311
 312     private void destroyContextChain(final DeviceInfo deviceInfo) {
 313         ownershipChangeListener.becomeSlaveOrDisconnect(deviceInfo);
 314         Optional.ofNullable(contextChainMap.get(deviceInfo)).ifPresent(contextChain -> {
 315             deviceManager.sendNodeRemovedNotification(deviceInfo.getNodeInstanceIdentifier());
 316             contextChain.close();
 317             connectingDevices.remove(deviceInfo);
 318         });
 319     }
 320
 321     @Override
 322     public List<DeviceInfo> listOfMasteredDevices() {
 323         return contextChainMap.entrySet().stream()
 324                 .filter(deviceInfoContextChainEntry -> deviceInfoContextChainEntry.getValue()
 325                         .isMastered(ContextChainMastershipState.CHECK, false)).map(Map.Entry::getKey)
 326                 .collect(Collectors.toList());
 327     }
 328
 329     @Override
 330     public boolean isAnyDeviceMastered() {
 331         return contextChainMap.entrySet().stream().findAny()
 332                 .filter(deviceInfoContextChainEntry -> deviceInfoContextChainEntry.getValue()
 333                         .isMastered(ContextChainMastershipState.CHECK, false)).isPresent();
 334     }
 335
 336     @Override
 337     public void onDeviceRemoved(final DeviceInfo deviceInfo) {
 338         contextChainMap.remove(deviceInfo);
 339         LOG.debug("Context chain removed for node {}", deviceInfo);
 340     }
 341
 342     private FutureCallback<ResultState> reconciliationFrameworkCallback(@Nonnull DeviceInfo deviceInfo,
 343                                                                         ContextChain contextChain) {
 344         return new FutureCallback<ResultState>() {
 345             @Override
 346             public void onSuccess(ResultState result) {
 347                 if (ResultState.DONOTHING == result) {
 348                     LOG.info("Device {} connection is enabled by reconciliation framework.", deviceInfo);
 349                     contextChain.continueInitializationAfterReconciliation();
 350                 } else {
 351                     LOG.warn("Reconciliation framework failure for device {}", deviceInfo);
 352                     destroyContextChain(deviceInfo);
 353                 }
 354             }
 355
 356             @Override
 357             public void onFailure(Throwable throwable) {
 358                 LOG.warn("Reconciliation framework failure.");
 359                 destroyContextChain(deviceInfo);
 360             }
 361         };
 362     }
 363 }