openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/role/RoleManagerImpl.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.role;
   9
  10 import com.google.common.base.Preconditions;
  11 import com.google.common.base.Verify;
  12 import com.google.common.collect.Iterators;
  13 import com.google.common.util.concurrent.CheckedFuture;
  14 import com.google.common.util.concurrent.FutureCallback;
  15 import com.google.common.util.concurrent.Futures;
  16 import io.netty.util.Timeout;
  17 import io.netty.util.TimerTask;
  18 import java.util.Iterator;
  19 import java.util.concurrent.ConcurrentHashMap;
  20 import java.util.concurrent.ConcurrentMap;
  21 import java.util.concurrent.TimeUnit;
  22 import javax.annotation.CheckForNull;
  23 import javax.annotation.Nonnull;
  24 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  25 import org.opendaylight.controller.md.sal.binding.api.WriteTransaction;
  26 import org.opendaylight.controller.md.sal.common.api.clustering.Entity;
  27 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipChange;
  28 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipListener;
  29 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipListenerRegistration;
  30 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipService;
  31 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  32 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
  33 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
  34 import org.opendaylight.openflowplugin.api.openflow.device.DeviceState;
  35 import org.opendaylight.openflowplugin.api.openflow.device.handlers.DeviceInitializationPhaseHandler;
  36 import org.opendaylight.openflowplugin.api.openflow.device.handlers.DeviceTerminationPhaseHandler;
  37 import org.opendaylight.openflowplugin.api.openflow.role.RoleChangeListener;
  38 import org.opendaylight.openflowplugin.api.openflow.role.RoleContext;
  39 import org.opendaylight.openflowplugin.api.openflow.role.RoleManager;
  40 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
  41 import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.OfpRole;
  42 import org.slf4j.Logger;
  43 import org.slf4j.LoggerFactory;
  44
  45 /**
  46  * Gets invoked from RpcManagerInitial, registers a candidate with EntityOwnershipService.
  47  * On receipt of the ownership notification, makes an rpc call to SalRoleSevice.
  48  *
  49  * Hands over to StatisticsManager at the end.
  50  */
  51 public class RoleManagerImpl implements RoleManager, EntityOwnershipListener {
  52     private static final Logger LOG = LoggerFactory.getLogger(RoleManagerImpl.class);
  53
  54     private DeviceInitializationPhaseHandler deviceInitializationPhaseHandler;
  55     private DeviceTerminationPhaseHandler deviceTerminationPhaseHandler;
  56     private final DataBroker dataBroker;
  57     private final EntityOwnershipService entityOwnershipService;
  58     private final ConcurrentMap<Entity, RoleContext> contexts = new ConcurrentHashMap<>();
  59     private final ConcurrentMap<Entity, RoleContext> txContexts = new ConcurrentHashMap<>();
  60     private final EntityOwnershipListenerRegistration entityOwnershipListenerRegistration;
  61     private final EntityOwnershipListenerRegistration txEntityOwnershipListenerRegistration;
  62
  63     public RoleManagerImpl(final EntityOwnershipService entityOwnershipService, final DataBroker dataBroker) {
  64         this.entityOwnershipService = Preconditions.checkNotNull(entityOwnershipService);
  65         this.dataBroker = Preconditions.checkNotNull(dataBroker);
  66         this.entityOwnershipListenerRegistration = Preconditions.checkNotNull(entityOwnershipService.registerListener(RoleManager.ENTITY_TYPE, this));
  67         this.txEntityOwnershipListenerRegistration = Preconditions.checkNotNull(entityOwnershipService.registerListener(TX_ENTITY_TYPE, this));
  68         LOG.debug("Register OpenflowOwnershipListener to all entity ownership changes");
  69     }
  70
  71     @Override
  72     public void setDeviceInitializationPhaseHandler(final DeviceInitializationPhaseHandler handler) {
  73         deviceInitializationPhaseHandler = handler;
  74     }
  75
  76     @Override
  77     public void onDeviceContextLevelUp(@CheckForNull final DeviceContext deviceContext) throws Exception {
  78         LOG.trace("Role manager called for device:{}", deviceContext.getPrimaryConnectionContext().getNodeId());
  79         final RoleContext roleContext = new RoleContextImpl(deviceContext, entityOwnershipService,
  80                 makeEntity(deviceContext.getDeviceState().getNodeId()),
  81                 makeTxEntity(deviceContext.getDeviceState().getNodeId()));
  82
  83         Verify.verify(contexts.putIfAbsent(roleContext.getEntity(), roleContext) == null, "Role context for master Node {} is still not closed.", deviceContext.getDeviceState().getNodeId());
  84         Verify.verify(!txContexts.containsKey(roleContext.getTxEntity()),
  85                 "Role context for master Node {} is still not closed. TxEntity was not unregistered yet.", deviceContext.getDeviceState().getNodeId());
  86
  87         // if the device context gets closed (mostly on connection close), we would need to cleanup
  88         deviceContext.addDeviceContextClosedHandler(this);
  89         roleContext.initializationRoleContext();
  90         deviceInitializationPhaseHandler.onDeviceContextLevelUp(deviceContext);
  91     }
  92
  93     @Override
  94     public void close() {
  95         entityOwnershipListenerRegistration.close();
  96         txEntityOwnershipListenerRegistration.close();
  97         for (final Iterator<RoleContext> iterator = Iterators.consumingIterator(contexts.values().iterator()); iterator.hasNext();) {
  98             // got here because last known role is LEADER and DS might need clearing up
  99             final RoleContext roleCtx = iterator.next();
 100             final NodeId nodeId = roleCtx.getDeviceState().getNodeId();
 101             if (OfpRole.BECOMEMASTER.equals(roleCtx.getClusterRole())) {
 102                 LOG.debug("Last role is LEADER and ownershipService returned hasOwner=false for node: {}; "
 103                         + "cleaning DS as being probably the last owner", nodeId);
 104                 removeDeviceFromOperDS(roleCtx);
 105             } else {
 106                 // NOOP - there is another owner
 107                 LOG.debug("Last role is LEADER and ownershipService returned hasOwner=true for node: {}; "
 108                         + "leaving DS untouched", nodeId);
 109             }
 110             txContexts.remove(roleCtx.getTxEntity(), roleCtx);
 111             roleCtx.close();
 112         }
 113     }
 114
 115     @Override
 116     public void onDeviceContextLevelDown(final DeviceContext deviceContext) {
 117         final NodeId nodeId = deviceContext.getDeviceState().getNodeId();
 118         LOG.trace("onDeviceContextLevelDown for node {}", nodeId);
 119         final Entity entity = makeEntity(nodeId);
 120         final RoleContext roleContext = contexts.get(entity);
 121         if (roleContext != null) {
 122             LOG.debug("Found roleContext associated to deviceContext: {}, now closing the roleContext", nodeId);
 123             roleContext.terminationRoleContext();
 124             final TimerTask timerTask = new TimerTask() {
 125
 126                 @Override
 127                 public void run(final Timeout timeout) throws Exception {
 128                     final RoleContext foundMainRoleCtx = contexts.get(roleContext.getEntity());
 129                     final RoleContext foundTxRoleCtx = txContexts.get(roleContext.getTxEntity());
 130
 131                     if (roleContext.equals(foundMainRoleCtx)) {
 132                         LOG.info("OldRoleCtx was not remove for entity {} from contexts", roleContext.getEntity());
 133                         contexts.remove(roleContext.getEntity(), roleContext);
 134                         foundMainRoleCtx.close();
 135                     }
 136
 137                     if (roleContext.equals(foundTxRoleCtx)) {
 138                         LOG.info("OldRoleCtx was not remove for txEntity {} from contexts", roleContext.getTxEntity());
 139                         txContexts.remove(roleContext.getTxEntity(), roleContext);
 140                         foundTxRoleCtx.close();
 141                     }
 142                 }
 143             };
 144             deviceContext.getTimer().newTimeout(timerTask, 10, TimeUnit.SECONDS);
 145         }
 146         deviceTerminationPhaseHandler.onDeviceContextLevelDown(deviceContext);
 147     }
 148
 149     private static Entity makeEntity(final NodeId nodeId) {
 150         return new Entity(RoleManager.ENTITY_TYPE, nodeId.getValue());
 151     }
 152
 153     private static Entity makeTxEntity(final NodeId nodeId) {
 154         return new Entity(RoleManager.TX_ENTITY_TYPE, nodeId.getValue());
 155     }
 156
 157     @Override
 158     public void ownershipChanged(final EntityOwnershipChange ownershipChange) {
 159         Preconditions.checkArgument(ownershipChange != null);
 160         RoleContext roleContext = null;
 161         try {
 162             roleContext = contexts.get(ownershipChange.getEntity());
 163             if (roleContext != null) {
 164                 changeOwnershipForMainEntity(ownershipChange, roleContext);
 165                 return;
 166             }
 167
 168             roleContext = txContexts.get(ownershipChange.getEntity());
 169             if (roleContext != null) {
 170                 changeOwnershipForTxEntity(ownershipChange, roleContext);
 171                 return;
 172             }
 173         } catch (final Exception e) {
 174             LOG.warn("fail to acquire semaphore: {}", ownershipChange.getEntity(), e);
 175             if (roleContext != null) {
 176                 roleContext.getDeviceContext().shutdownConnection();
 177             }
 178         }
 179
 180         LOG.debug("We are not able to find Entity {} ownershipChange {} - disregarding ownership notification",
 181                 ownershipChange.getEntity(), ownershipChange);
 182     }
 183
 184     private void changeOwnershipForMainEntity(final EntityOwnershipChange ownershipChange,
 185             @CheckForNull final RoleContext roleContext) {
 186
 187         LOG.debug("Received Main-EntityOwnershipChange:{}", ownershipChange);
 188         Preconditions.checkArgument(roleContext != null);
 189         if (roleContext.isMainCandidateRegistered()) {
 190             LOG.debug("Main-EntityOwnershipRegistration is active for entity {}", ownershipChange.getEntity());
 191             if (!ownershipChange.wasOwner() && ownershipChange.isOwner()) {
 192                 // SLAVE -> MASTER
 193                 txContexts.put(roleContext.getTxEntity(), roleContext);
 194                 roleContext.onDeviceTryToTakeClusterLeadership();
 195             } else if (ownershipChange.wasOwner() && !ownershipChange.isOwner()) {
 196                 // MASTER -> SLAVE
 197                 roleContext.onDeviceLostClusterLeadership();
 198             } else if (LOG.isDebugEnabled()) {
 199                 LOG.debug("Not processed Ownership Main Entity {} Event {}", ownershipChange.getEntity(), ownershipChange);
 200             }
 201         } else {
 202             LOG.debug("Main-EntityOwnershipRegistration is not active for entity {}", ownershipChange.getEntity());
 203             contexts.remove(ownershipChange.getEntity(), roleContext);
 204             if (!ownershipChange.hasOwner() && !ownershipChange.isOwner() && ownershipChange.wasOwner()) {
 205                 /* Method has to clean all context and registrations */
 206                 unregistrationHelper(ownershipChange, roleContext);
 207             } else {
 208                 txContexts.remove(roleContext.getTxEntity(), roleContext);
 209                 roleContext.close();
 210             }
 211         }
 212     }
 213
 214     private void changeOwnershipForTxEntity(final EntityOwnershipChange ownershipChange,
 215             @Nonnull final RoleContext roleContext) {
 216
 217         LOG.debug("Received TX-EntityOwnershipChange:{}", ownershipChange);
 218         Preconditions.checkArgument(roleContext != null);
 219         if (roleContext.isTxCandidateRegistered()) {
 220             LOG.debug("Tx-EntityOwnershipRegistration is active for entity {}", ownershipChange.getEntity());
 221             if (!ownershipChange.wasOwner() && ownershipChange.isOwner()) {
 222                 // SLAVE -> MASTER
 223                 roleContext.onDeviceTakeClusterLeadership();
 224             } else if (ownershipChange.wasOwner() && !ownershipChange.isOwner()) {
 225                 // MASTER -> SLAVE
 226                 LOG.warn("Tx-EntityOwnershipRegistration unexpected lost Leadership entity {}", ownershipChange.getEntity());
 227                 roleContext.getDeviceContext().shutdownConnection();
 228             } else {
 229                 LOG.debug("NOOP state transition for TxEntity {} ", roleContext.getTxEntity());
 230             }
 231         } else {
 232             LOG.debug("Tx-EntityOwnershipRegistration is not active for entity {}", ownershipChange.getEntity());
 233             txContexts.remove(ownershipChange.getEntity(), roleContext);
 234         }
 235     }
 236
 237     private CheckedFuture<Void, TransactionCommitFailedException> removeDeviceFromOperDS(
 238             final RoleChangeListener roleChangeListener) {
 239         Preconditions.checkArgument(roleChangeListener != null);
 240         final DeviceState deviceState = roleChangeListener.getDeviceState();
 241         final WriteTransaction delWtx = dataBroker.newWriteOnlyTransaction();
 242         delWtx.delete(LogicalDatastoreType.OPERATIONAL, deviceState.getNodeInstanceIdentifier());
 243         final CheckedFuture<Void, TransactionCommitFailedException> delFuture = delWtx.submit();
 244         Futures.addCallback(delFuture, new FutureCallback<Void>() {
 245
 246             @Override
 247             public void onSuccess(final Void result) {
 248                 LOG.debug("Delete Node {} was successful", deviceState.getNodeId());
 249             }
 250
 251             @Override
 252             public void onFailure(final Throwable t) {
 253                 LOG.warn("Delete Node {} failed.", deviceState.getNodeId(), t);
 254             }
 255         });
 256         return delFuture;
 257     }
 258
 259     private void unregistrationHelper(final EntityOwnershipChange ownershipChange, final RoleContext roleContext) {
 260         LOG.info("Initiate removal from operational. Possibly the last node to be disconnected for :{}. ",
 261                 ownershipChange);
 262         Futures.addCallback(removeDeviceFromOperDS(roleContext), new FutureCallback<Void>() {
 263             @Override
 264             public void onSuccess(final Void aVoid) {
 265                 LOG.debug("Removing context for device: {}", roleContext.getDeviceState().getNodeId());
 266                 txContexts.remove(roleContext.getTxEntity(), roleContext);
 267                 roleContext.close();
 268             }
 269
 270             @Override
 271             public void onFailure(final Throwable throwable) {
 272                 LOG.warn("Removing role context for device: {}, but {}", roleContext.getDeviceState().getNodeId(),
 273                         throwable.getMessage());
 274                 txContexts.remove(roleContext.getTxEntity(), roleContext);
 275                 roleContext.close();
 276             }
 277         });
 278     }
 279
 280     @Override
 281     public void setDeviceTerminationPhaseHandler(final DeviceTerminationPhaseHandler handler) {
 282         deviceTerminationPhaseHandler = handler;
 283     }
 284 }