359b428c99d5060f59baa45ca2414a34eba46f98
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / databroker / actors / dds / ModuleShardBackendResolver.java
1 /*
2  * Copyright (c) 2016 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.databroker.actors.dds;
9
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;
37
38 /**
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}.
42  *
43  * @author Robert Varga
44  */
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);
48
49     /**
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
52      * non-operational.
53      */
54     // TODO: maybe make this configurable somehow?
55     private static final Timeout DEAD_TIMEOUT = Timeout.apply(15, TimeUnit.MINUTES);
56
57     private final ActorContext actorContext;
58     // FIXME: this counter should be in superclass somewhere
59     private final AtomicLong nextSessionId = new AtomicLong();
60
61     @GuardedBy("this")
62     private long nextShard = 1;
63
64     private volatile BiMap<String, Long> shards = ImmutableBiMap.of(DefaultShardStrategy.DEFAULT_SHARD, 0L);
65
66     // FIXME: we really need just ActorContext.findPrimaryShardAsync()
67     ModuleShardBackendResolver(final ActorContext actorContext) {
68         this.actorContext = Preconditions.checkNotNull(actorContext);
69     }
70
71     @Override
72     protected void invalidateBackendInfo(final CompletionStage<? extends BackendInfo> info) {
73         LOG.trace("Initiated invalidation of backend information {}", info);
74         info.thenAccept(this::invalidate);
75     }
76
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());
81     }
82
83     Long resolveShardForPath(final YangInstanceIdentifier path) {
84         final String shardName = actorContext.getShardStrategyFactory().getStrategy(path).findShard(path);
85         Long cookie = shards.get(shardName);
86         if (cookie == null) {
87             synchronized (this) {
88                 cookie = shards.get(shardName);
89                 if (cookie == null) {
90                     cookie = nextShard++;
91
92                     Builder<String, Long> b = ImmutableBiMap.builder();
93                     b.putAll(shards);
94                     b.put(shardName, cookie);
95                     shards = b.build();
96                 }
97             }
98         }
99
100         return cookie;
101     }
102
103     @Override
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);
108             return NULL_FUTURE;
109         }
110
111         final CompletableFuture<ShardBackendInfo> ret = new CompletableFuture<ShardBackendInfo>();
112
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());
123             }
124
125             LOG.debug("Resolved backend information to {}", response);
126
127             Preconditions.checkArgument(response instanceof ConnectClientSuccess, "Unhandled response {}", response);
128             final ConnectClientSuccess success = (ConnectClientSuccess) response;
129
130             return new ShardBackendInfo(success.getBackend(),
131                 nextSessionId.getAndIncrement(), success.getVersion(), shardName, UnsignedLong.fromLongBits(cookie),
132                 success.getDataTree(), success.getMaxMessages());
133         }).whenComplete((info, t) -> {
134             if (t != null) {
135                 ret.completeExceptionally(t);
136             } else {
137                 ret.complete(info);
138             }
139         });
140
141         LOG.debug("Resolving cookie {} to shard {}", cookie, shardName);
142         return ret;
143     }
144 }