2 * Copyright (c) 2017 Cisco Systems, Inc. and others. All rights reserved.
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
8 package org.opendaylight.controller.cluster.sharding;
10 import static akka.actor.ActorRef.noSender;
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;
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.
46 * Sends {@link MakeLeaderLocal} message to local shards and translates its result
47 * on behalf users {@link #makeLeaderLocal()} calls.
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
54 final class CDSShardAccessImpl implements CDSShardAccess, LeaderLocationListener, AutoCloseable {
55 private static final Logger LOG = LoggerFactory.getLogger(CDSShardAccessImpl.class);
57 private final Collection<LeaderLocationListener> listeners = ConcurrentHashMap.newKeySet();
58 private final DOMDataTreeIdentifier prefix;
59 private final ActorContext actorContext;
60 private final Timeout makeLeaderLocalTimeout;
62 private ActorRef roleChangeListenerActor;
64 private volatile LeaderLocation currentLeader = LeaderLocation.UNKNOWN;
65 private volatile boolean closed = false;
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));
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));
83 private void checkNotClosed() {
84 Preconditions.checkState(!closed,
85 "CDSDataTreeProducer, that this CDSShardAccess is associated with, is no longer valid");
90 public DOMDataTreeIdentifier getShardIdentifier() {
97 public LeaderLocation getLeaderLocation() {
99 // TODO before getting first notification from roleChangeListenerActor
100 // we will always return UNKNOWN
101 return currentLeader;
106 public CompletionStage<Void> makeLeaderLocal() {
107 // TODO when we have running make leader local operation
108 // we should just return the same completion stage
111 // TODO can we cache local shard actorRef?
112 final Future<ActorRef> localShardReply =
113 actorContext.findLocalShardAsync(ClusterUtils.getCleanShardName(prefix.getRootIdentifier()));
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>() {
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);
131 .completeWith(actorContext
132 .executeOperationAsync(actorRef, MakeLeaderLocal.INSTANCE, makeLeaderLocalTimeout));
135 }, actorContext.getClientDispatcher());
137 // we have to transform make leader local request result
138 Future<Void> makeLeaderLocalFuture = makeLeaderLocalAsk.future()
139 .transform(new Mapper<Object, Void>() {
141 public Void apply(final Object parameter) {
144 }, new Mapper<Throwable, Throwable>() {
146 public Throwable apply(final Throwable parameter) {
147 if (parameter instanceof LeadershipTransferFailedException) {
148 // do nothing with exception and just pass it as it is
151 // wrap exception in LeadershipTransferFailedEx
152 return new LeadershipTransferFailedException("Leadership transfer failed", parameter);
154 }, actorContext.getClientDispatcher());
156 return FutureConverters.toJava(makeLeaderLocalFuture);
161 public <L extends LeaderLocationListener> LeaderLocationListenerRegistration<L>
162 registerLeaderLocationListener(@Nonnull final L listener) {
164 Preconditions.checkNotNull(listener);
165 Preconditions.checkArgument(!listeners.contains(listener),
166 "Listener {} is already registered with ShardAccess {}", listener, this);
168 LOG.debug("Registering LeaderLocationListener {}", listener);
170 listeners.add(listener);
172 return new LeaderLocationListenerRegistration<L>() {
174 public L getInstance() {
179 public void close() {
180 listeners.remove(listener);
186 @SuppressWarnings("checkstyle:IllegalCatch")
187 public void onLeaderLocationChanged(@Nonnull final LeaderLocation location) {
189 // we are closed already. Do not dispatch any new leader location
194 LOG.debug("Received leader location change notification. New leader location: {}", location);
195 currentLeader = location;
196 listeners.forEach(listener -> {
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);
207 public void close() {
208 // TODO should we also remove all listeners?
209 LOG.debug("Closing {} ShardAccess", prefix);
212 if (roleChangeListenerActor != null) {
213 // stop RoleChangeListenerActor
214 roleChangeListenerActor.tell(PoisonPill.getInstance(), noSender());
215 roleChangeListenerActor = null;