2 * Copyright (c) 2016 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.databroker.actors.dds;
10 import akka.actor.ActorRef;
11 import akka.japi.Function;
12 import akka.pattern.Patterns;
13 import akka.util.Timeout;
14 import com.google.common.base.Preconditions;
15 import com.google.common.base.Throwables;
16 import com.google.common.collect.BiMap;
17 import com.google.common.collect.ImmutableBiMap;
18 import com.google.common.collect.ImmutableBiMap.Builder;
19 import com.google.common.primitives.UnsignedLong;
20 import java.util.concurrent.CompletableFuture;
21 import java.util.concurrent.CompletionStage;
22 import java.util.concurrent.TimeUnit;
23 import java.util.concurrent.atomic.AtomicLong;
24 import javax.annotation.concurrent.GuardedBy;
25 import org.opendaylight.controller.cluster.access.ABIVersion;
26 import org.opendaylight.controller.cluster.access.client.BackendInfo;
27 import org.opendaylight.controller.cluster.access.client.BackendInfoResolver;
28 import org.opendaylight.controller.cluster.access.commands.ConnectClientRequest;
29 import org.opendaylight.controller.cluster.access.commands.ConnectClientSuccess;
30 import org.opendaylight.controller.cluster.access.concepts.RequestFailure;
31 import org.opendaylight.controller.cluster.datastore.shardstrategy.DefaultShardStrategy;
32 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
33 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
34 import org.slf4j.Logger;
35 import org.slf4j.LoggerFactory;
36 import scala.compat.java8.FutureConverters;
39 * {@link BackendInfoResolver} implementation for static shard configuration based on ShardManager. Each string-named
40 * shard is assigned a single cookie and this mapping is stored in a bidirectional map. Information about corresponding
41 * shard leader is resolved via {@link ActorContext}. The product of resolution is {@link ShardBackendInfo}.
43 * @author Robert Varga
45 final class ModuleShardBackendResolver extends BackendInfoResolver<ShardBackendInfo> {
46 private static final CompletableFuture<ShardBackendInfo> NULL_FUTURE = CompletableFuture.completedFuture(null);
47 private static final Logger LOG = LoggerFactory.getLogger(ModuleShardBackendResolver.class);
50 * Fall-over-dead timeout. If we do not make progress in this long, just fall over and propagate the failure.
51 * All users are expected to fail, possibly attempting to recover by restarting. It is fair to remain
54 // TODO: maybe make this configurable somehow?
55 private static final Timeout DEAD_TIMEOUT = Timeout.apply(15, TimeUnit.MINUTES);
57 private final ActorContext actorContext;
58 // FIXME: this counter should be in superclass somewhere
59 private final AtomicLong nextSessionId = new AtomicLong();
62 private long nextShard = 1;
64 private volatile BiMap<String, Long> shards = ImmutableBiMap.of(DefaultShardStrategy.DEFAULT_SHARD, 0L);
66 // FIXME: we really need just ActorContext.findPrimaryShardAsync()
67 ModuleShardBackendResolver(final ActorContext actorContext) {
68 this.actorContext = Preconditions.checkNotNull(actorContext);
72 protected void invalidateBackendInfo(final CompletionStage<? extends BackendInfo> info) {
73 LOG.trace("Initiated invalidation of backend information {}", info);
74 info.thenAccept(this::invalidate);
77 private void invalidate(final BackendInfo result) {
78 Preconditions.checkArgument(result instanceof ShardBackendInfo);
79 LOG.debug("Invalidating backend information {}", result);
80 actorContext.getPrimaryShardInfoCache().remove(((ShardBackendInfo)result).getShardName());
83 Long resolveShardForPath(final YangInstanceIdentifier path) {
84 final String shardName = actorContext.getShardStrategyFactory().getStrategy(path).findShard(path);
85 Long cookie = shards.get(shardName);
88 cookie = shards.get(shardName);
92 Builder<String, Long> b = ImmutableBiMap.builder();
94 b.put(shardName, cookie);
104 protected CompletableFuture<ShardBackendInfo> resolveBackendInfo(final Long cookie) {
105 final String shardName = shards.inverse().get(cookie);
106 if (shardName == null) {
107 LOG.warn("Failing request for non-existent cookie {}", cookie);
111 final CompletableFuture<ShardBackendInfo> ret = new CompletableFuture<ShardBackendInfo>();
113 FutureConverters.toJava(actorContext.findPrimaryShardAsync(shardName)).thenCompose(info -> {
114 LOG.debug("Looking up primary info for {} from {}", shardName, info);
115 return FutureConverters.toJava(Patterns.ask(info.getPrimaryShardActor(),
116 (Function<ActorRef, Object>) replyTo -> new ConnectClientRequest(null, replyTo,
117 ABIVersion.BORON, ABIVersion.current()), DEAD_TIMEOUT));
118 }).thenApply(response -> {
119 if (response instanceof RequestFailure) {
120 final RequestFailure<?, ?> failure = (RequestFailure<?, ?>) response;
121 LOG.debug("Connect request failed {}", failure, failure.getCause());
122 throw Throwables.propagate(failure.getCause());
125 LOG.debug("Resolved backend information to {}", response);
127 Preconditions.checkArgument(response instanceof ConnectClientSuccess, "Unhandled response {}", response);
128 final ConnectClientSuccess success = (ConnectClientSuccess) response;
130 return new ShardBackendInfo(success.getBackend(),
131 nextSessionId.getAndIncrement(), success.getVersion(), shardName, UnsignedLong.fromLongBits(cookie),
132 success.getDataTree(), success.getMaxMessages());
133 }).whenComplete((info, t) -> {
135 ret.completeExceptionally(t);
141 LOG.debug("Resolving cookie {} to shard {}", cookie, shardName);