opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/sharding/CDSShardAccessImpl.java

   1 /*
   2  * Copyright (c) 2017 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.controller.cluster.sharding;
   9
  10 import static akka.actor.ActorRef.noSender;
  11
  12 import akka.actor.ActorRef;
  13 import akka.actor.PoisonPill;
  14 import akka.dispatch.Futures;
  15 import akka.dispatch.Mapper;
  16 import akka.dispatch.OnComplete;
  17 import akka.util.Timeout;
  18 import com.google.common.base.Optional;
  19 import com.google.common.base.Preconditions;
  20 import java.util.Collection;
  21 import java.util.concurrent.CompletionStage;
  22 import java.util.concurrent.ConcurrentHashMap;
  23 import javax.annotation.Nonnull;
  24 import org.opendaylight.controller.cluster.datastore.exceptions.LocalShardNotFoundException;
  25 import org.opendaylight.controller.cluster.datastore.messages.MakeLeaderLocal;
  26 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
  27 import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
  28 import org.opendaylight.controller.cluster.dom.api.CDSShardAccess;
  29 import org.opendaylight.controller.cluster.dom.api.LeaderLocation;
  30 import org.opendaylight.controller.cluster.dom.api.LeaderLocationListener;
  31 import org.opendaylight.controller.cluster.dom.api.LeaderLocationListenerRegistration;
  32 import org.opendaylight.controller.cluster.raft.LeadershipTransferFailedException;
  33 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
  34 import org.slf4j.Logger;
  35 import org.slf4j.LoggerFactory;
  36 import scala.compat.java8.FutureConverters;
  37 import scala.concurrent.Future;
  38
  39
  40 /**
  41  * Default {@link CDSShardAccess} implementation. Listens on leader location
  42  * change events and distributes them to registered listeners. Also updates
  43  * current information about leader location accordingly.
  44  *
  45  * <p>
  46  * Sends {@link MakeLeaderLocal} message to local shards and translates its result
  47  * on behalf users {@link #makeLeaderLocal()} calls.
  48  *
  49  * <p>
  50  * {@link org.opendaylight.controller.cluster.dom.api.CDSDataTreeProducer} that
  51  * creates instances of this class has to call {@link #close()} once it is no
  52  * longer valid.
  53  */
  54 final class CDSShardAccessImpl implements CDSShardAccess, LeaderLocationListener, AutoCloseable {
  55     private static final Logger LOG = LoggerFactory.getLogger(CDSShardAccessImpl.class);
  56
  57     private final Collection<LeaderLocationListener> listeners = ConcurrentHashMap.newKeySet();
  58     private final DOMDataTreeIdentifier prefix;
  59     private final ActorContext actorContext;
  60     private final Timeout makeLeaderLocalTimeout;
  61
  62     private ActorRef roleChangeListenerActor;
  63
  64     private volatile LeaderLocation currentLeader = LeaderLocation.UNKNOWN;
  65     private volatile boolean closed = false;
  66
  67     CDSShardAccessImpl(final DOMDataTreeIdentifier prefix, final ActorContext actorContext) {
  68         this.prefix = Preconditions.checkNotNull(prefix);
  69         this.actorContext = Preconditions.checkNotNull(actorContext);
  70         this.makeLeaderLocalTimeout =
  71                 new Timeout(actorContext.getDatastoreContext().getShardLeaderElectionTimeout().duration().$times(2));
  72
  73         // register RoleChangeListenerActor
  74         // TODO Maybe we should do this in async
  75         final Optional<ActorRef> localShardReply =
  76                 actorContext.findLocalShard(ClusterUtils.getCleanShardName(prefix.getRootIdentifier()));
  77         Preconditions.checkState(localShardReply.isPresent(),
  78                 "Local shard for {} not present. Cannot register RoleChangeListenerActor", prefix);
  79         roleChangeListenerActor =
  80                 actorContext.getActorSystem().actorOf(RoleChangeListenerActor.props(localShardReply.get(), this));
  81     }
  82
  83     private void checkNotClosed() {
  84         Preconditions.checkState(!closed,
  85                 "CDSDataTreeProducer, that this CDSShardAccess is associated with, is no longer valid");
  86     }
  87
  88     @Override
  89     @Nonnull
  90     public DOMDataTreeIdentifier getShardIdentifier() {
  91         checkNotClosed();
  92         return prefix;
  93     }
  94
  95     @Override
  96     @Nonnull
  97     public LeaderLocation getLeaderLocation() {
  98         checkNotClosed();
  99         // TODO before getting first notification from roleChangeListenerActor
 100         // we will always return UNKNOWN
 101         return currentLeader;
 102     }
 103
 104     @Override
 105     @Nonnull
 106     public CompletionStage<Void> makeLeaderLocal() {
 107         // TODO when we have running make leader local operation
 108         // we should just return the same completion stage
 109         checkNotClosed();
 110
 111         // TODO can we cache local shard actorRef?
 112         final Future<ActorRef> localShardReply =
 113                 actorContext.findLocalShardAsync(ClusterUtils.getCleanShardName(prefix.getRootIdentifier()));
 114
 115         // we have to tell local shard to make leader local
 116         final scala.concurrent.Promise<Object> makeLeaderLocalAsk = Futures.promise();
 117         localShardReply.onComplete(new OnComplete<ActorRef>() {
 118             @Override
 119             public void onComplete(final Throwable failure, final ActorRef actorRef) throws Throwable {
 120                 if (failure instanceof LocalShardNotFoundException) {
 121                     LOG.debug("No local shard found for {} - Cannot request leadership transfer to local shard.",
 122                             getShardIdentifier(), failure);
 123                     makeLeaderLocalAsk.failure(failure);
 124                 } else if (failure != null) {
 125                     // TODO should this be WARN?
 126                     LOG.debug("Failed to find local shard for {} - Cannot request leadership transfer to local shard.",
 127                             getShardIdentifier(), failure);
 128                     makeLeaderLocalAsk.failure(failure);
 129                 } else {
 130                     makeLeaderLocalAsk
 131                             .completeWith(actorContext
 132                                     .executeOperationAsync(actorRef, MakeLeaderLocal.INSTANCE, makeLeaderLocalTimeout));
 133                 }
 134             }
 135         }, actorContext.getClientDispatcher());
 136
 137         // we have to transform make leader local request result
 138         Future<Void> makeLeaderLocalFuture = makeLeaderLocalAsk.future()
 139                 .transform(new Mapper<Object, Void>() {
 140                     @Override
 141                     public Void apply(final Object parameter) {
 142                         return null;
 143                     }
 144                 }, new Mapper<Throwable, Throwable>() {
 145                     @Override
 146                     public Throwable apply(final Throwable parameter) {
 147                         if (parameter instanceof LeadershipTransferFailedException) {
 148                             // do nothing with exception and just pass it as it is
 149                             return parameter;
 150                         }
 151                         // wrap exception in LeadershipTransferFailedEx
 152                         return new LeadershipTransferFailedException("Leadership transfer failed", parameter);
 153                     }
 154                 }, actorContext.getClientDispatcher());
 155
 156         return FutureConverters.toJava(makeLeaderLocalFuture);
 157     }
 158
 159     @Override
 160     @Nonnull
 161     public <L extends LeaderLocationListener> LeaderLocationListenerRegistration<L>
 162             registerLeaderLocationListener(@Nonnull final L listener) {
 163         checkNotClosed();
 164         Preconditions.checkNotNull(listener);
 165         Preconditions.checkArgument(!listeners.contains(listener),
 166                 "Listener {} is already registered with ShardAccess {}", listener, this);
 167
 168         LOG.debug("Registering LeaderLocationListener {}", listener);
 169
 170         listeners.add(listener);
 171
 172         return new LeaderLocationListenerRegistration<L>() {
 173             @Override
 174             public L getInstance() {
 175                 return listener;
 176             }
 177
 178             @Override
 179             public void close() {
 180                 listeners.remove(listener);
 181             }
 182         };
 183     }
 184
 185     @Override
 186     @SuppressWarnings("checkstyle:IllegalCatch")
 187     public void onLeaderLocationChanged(@Nonnull final LeaderLocation location) {
 188         if (closed) {
 189             // we are closed already. Do not dispatch any new leader location
 190             // change events.
 191             return;
 192         }
 193
 194         LOG.debug("Received leader location change notification. New leader location: {}", location);
 195         currentLeader = location;
 196         listeners.forEach(listener -> {
 197             try {
 198                 listener.onLeaderLocationChanged(location);
 199             } catch (Exception e) {
 200                 LOG.warn("Ignoring uncaught exception thrown be LeaderLocationListener {} "
 201                         + "during processing leader location change {}", listener, location, e);
 202             }
 203         });
 204     }
 205
 206     @Override
 207     public void close() {
 208         // TODO should we also remove all listeners?
 209         LOG.debug("Closing {} ShardAccess", prefix);
 210         closed = true;
 211
 212         if (roleChangeListenerActor != null) {
 213             // stop RoleChangeListenerActor
 214             roleChangeListenerActor.tell(PoisonPill.getInstance(), noSender());
 215             roleChangeListenerActor = null;
 216         }
 217     }
 218 }