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.util.Timeout;
  12 import com.google.common.base.Preconditions;
  13 import com.google.common.base.Throwables;
  14 import com.google.common.collect.BiMap;
  15 import com.google.common.collect.ImmutableBiMap;
  16 import com.google.common.collect.ImmutableBiMap.Builder;
  17 import com.google.common.primitives.UnsignedLong;
  18 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
  19 import java.util.concurrent.CompletableFuture;
  20 import java.util.concurrent.CompletionStage;
  21 import java.util.concurrent.ConcurrentHashMap;
  22 import java.util.concurrent.ConcurrentMap;
  23 import java.util.concurrent.TimeUnit;
  24 import java.util.concurrent.atomic.AtomicLong;
  25 import javax.annotation.Nonnull;
  26 import javax.annotation.Nullable;
  27 import javax.annotation.concurrent.GuardedBy;
  28 import javax.annotation.concurrent.ThreadSafe;
  29 import org.opendaylight.controller.cluster.access.ABIVersion;
  30 import org.opendaylight.controller.cluster.access.client.BackendInfoResolver;
  31 import org.opendaylight.controller.cluster.access.commands.ConnectClientRequest;
  32 import org.opendaylight.controller.cluster.access.commands.ConnectClientSuccess;
  33 import org.opendaylight.controller.cluster.access.concepts.ClientIdentifier;
  34 import org.opendaylight.controller.cluster.access.concepts.RequestFailure;
  35 import org.opendaylight.controller.cluster.common.actor.ExplicitAsk;
  36 import org.opendaylight.controller.cluster.datastore.shardstrategy.DefaultShardStrategy;
  37 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
  38 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  39 import org.slf4j.Logger;
  40 import org.slf4j.LoggerFactory;
  41 import scala.Function1;
  42 import scala.compat.java8.FutureConverters;
  43
  44 /**
  45  * {@link BackendInfoResolver} implementation for static shard configuration based on ShardManager. Each string-named
  46  * shard is assigned a single cookie and this mapping is stored in a bidirectional map. Information about corresponding
  47  * shard leader is resolved via {@link ActorContext}. The product of resolution is {@link ShardBackendInfo}.
  48  *
  49  * @author Robert Varga
  50  */
  51 @SuppressFBWarnings(value = "NP_NONNULL_PARAM_VIOLATION",
  52                     justification = "Pertains to the NULL_FUTURE field below. Null is allowed and is intended")
  53 @ThreadSafe
  54 final class ModuleShardBackendResolver extends BackendInfoResolver<ShardBackendInfo> {
  55     private static final class Entry {
  56         private final CompletionStage<ShardBackendInfo> stage;
  57         @GuardedBy("this")
  58         private ShardBackendInfo result;
  59
  60         Entry(final CompletionStage<ShardBackendInfo> stage) {
  61             this.stage = Preconditions.checkNotNull(stage);
  62             stage.whenComplete(this::onStageResolved);
  63         }
  64
  65         @Nonnull CompletionStage<ShardBackendInfo> getStage() {
  66             return stage;
  67         }
  68
  69         synchronized @Nullable ShardBackendInfo getResult() {
  70             return result;
  71         }
  72
  73         private synchronized void onStageResolved(final ShardBackendInfo result, final Throwable failure) {
  74             if (failure == null) {
  75                 this.result = Preconditions.checkNotNull(result);
  76             } else {
  77                 LOG.warn("Failed to resolve shard", failure);
  78             }
  79         }
  80     }
  81
  82     private static final CompletableFuture<ShardBackendInfo> NULL_FUTURE = CompletableFuture.completedFuture(null);
  83     private static final Logger LOG = LoggerFactory.getLogger(ModuleShardBackendResolver.class);
  84
  85     /**
  86      * Fall-over-dead timeout. If we do not make progress in this long, just fall over and propagate the failure.
  87      * All users are expected to fail, possibly attempting to recover by restarting. It is fair to remain
  88      * non-operational.
  89      */
  90     // TODO: maybe make this configurable somehow?
  91     private static final Timeout DEAD_TIMEOUT = Timeout.apply(15, TimeUnit.MINUTES);
  92
  93     private final ConcurrentMap<Long, Entry> backends = new ConcurrentHashMap<>();
  94     private final AtomicLong nextSessionId = new AtomicLong();
  95     private final Function1<ActorRef, ?> connectFunction;
  96     private final ActorContext actorContext;
  97
  98     @GuardedBy("this")
  99     private long nextShard = 1;
 100
 101     private volatile BiMap<String, Long> shards = ImmutableBiMap.of(DefaultShardStrategy.DEFAULT_SHARD, 0L);
 102
 103     // FIXME: we really need just ActorContext.findPrimaryShardAsync()
 104     ModuleShardBackendResolver(final ClientIdentifier clientId, final ActorContext actorContext) {
 105         this.actorContext = Preconditions.checkNotNull(actorContext);
 106         this.connectFunction = ExplicitAsk.toScala(t -> new ConnectClientRequest(clientId, t, ABIVersion.BORON,
 107             ABIVersion.current()));
 108     }
 109
 110     Long resolveShardForPath(final YangInstanceIdentifier path) {
 111         final String shardName = actorContext.getShardStrategyFactory().getStrategy(path).findShard(path);
 112         Long cookie = shards.get(shardName);
 113         if (cookie == null) {
 114             synchronized (this) {
 115                 cookie = shards.get(shardName);
 116                 if (cookie == null) {
 117                     cookie = nextShard++;
 118
 119                     Builder<String, Long> builder = ImmutableBiMap.builder();
 120                     builder.putAll(shards);
 121                     builder.put(shardName, cookie);
 122                     shards = builder.build();
 123                 }
 124             }
 125         }
 126
 127         return cookie;
 128     }
 129
 130     private CompletionStage<ShardBackendInfo> resolveBackendInfo(final Long cookie) {
 131         final String shardName = shards.inverse().get(cookie);
 132         if (shardName == null) {
 133             LOG.warn("Failing request for non-existent cookie {}", cookie);
 134             return NULL_FUTURE;
 135         }
 136
 137         LOG.debug("Resolving cookie {} to shard {}", cookie, shardName);
 138
 139         return FutureConverters.toJava(actorContext.findPrimaryShardAsync(shardName)).thenCompose(info -> {
 140             LOG.debug("Looking up primary info for {} from {}", shardName, info);
 141             return FutureConverters.toJava(ExplicitAsk.ask(info.getPrimaryShardActor(), connectFunction, DEAD_TIMEOUT));
 142         }).thenApply(response -> {
 143             if (response instanceof RequestFailure) {
 144                 final RequestFailure<?, ?> failure = (RequestFailure<?, ?>) response;
 145                 LOG.debug("Connect request failed {}", failure, failure.getCause());
 146                 throw Throwables.propagate(failure.getCause());
 147             }
 148
 149             LOG.debug("Resolved backend information to {}", response);
 150
 151             Preconditions.checkArgument(response instanceof ConnectClientSuccess, "Unhandled response {}", response);
 152             final ConnectClientSuccess success = (ConnectClientSuccess) response;
 153
 154             return new ShardBackendInfo(success.getBackend(),
 155                 nextSessionId.getAndIncrement(), success.getVersion(), shardName, UnsignedLong.fromLongBits(cookie),
 156                 success.getDataTree(), success.getMaxMessages());
 157         });
 158     }
 159
 160     @Override
 161     public CompletionStage<? extends ShardBackendInfo> getBackendInfo(final Long cookie) {
 162         return backends.computeIfAbsent(cookie, key -> new Entry(resolveBackendInfo(key))).getStage();
 163     }
 164
 165     @Override
 166     public CompletionStage<? extends ShardBackendInfo> refreshBackendInfo(final Long cookie,
 167             final ShardBackendInfo staleInfo) {
 168         final Entry existing = backends.get(cookie);
 169         if (existing != null) {
 170             if (!staleInfo.equals(existing.getResult())) {
 171                 return existing.getStage();
 172             }
 173
 174             LOG.debug("Invalidating backend information {}", staleInfo);
 175             actorContext.getPrimaryShardInfoCache().remove(staleInfo.getShardName());
 176
 177             LOG.trace("Invalidated cache %s -> %s", Long.toUnsignedString(cookie), staleInfo);
 178             backends.remove(cookie, existing);
 179         }
 180
 181         return getBackendInfo(cookie);
 182     }
 183 }