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