2 * Copyright (c) 2016, 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
9 package org.opendaylight.controller.cluster.sharding;
11 import static akka.actor.ActorRef.noSender;
12 import static com.google.common.base.Preconditions.checkArgument;
13 import static com.google.common.base.Preconditions.checkState;
14 import static java.util.Objects.requireNonNull;
16 import akka.actor.ActorRef;
17 import akka.actor.ActorSystem;
18 import akka.actor.PoisonPill;
19 import akka.actor.Props;
20 import akka.dispatch.Mapper;
21 import akka.dispatch.OnComplete;
22 import akka.pattern.Patterns;
23 import akka.util.Timeout;
24 import com.google.common.base.Optional;
25 import com.google.common.base.Preconditions;
26 import com.google.common.base.Throwables;
27 import com.google.common.collect.ClassToInstanceMap;
28 import com.google.common.collect.ForwardingObject;
29 import com.google.common.collect.ImmutableClassToInstanceMap;
30 import com.google.common.util.concurrent.FutureCallback;
31 import com.google.common.util.concurrent.Futures;
32 import com.google.common.util.concurrent.ListenableFuture;
33 import com.google.common.util.concurrent.MoreExecutors;
34 import com.google.common.util.concurrent.SettableFuture;
35 import com.google.common.util.concurrent.Uninterruptibles;
36 import java.util.AbstractMap.SimpleEntry;
37 import java.util.Collection;
38 import java.util.Collections;
39 import java.util.Comparator;
40 import java.util.EnumMap;
41 import java.util.HashMap;
42 import java.util.List;
44 import java.util.Map.Entry;
46 import java.util.concurrent.CompletionStage;
47 import java.util.concurrent.ExecutionException;
48 import java.util.concurrent.TimeUnit;
49 import java.util.concurrent.TimeoutException;
50 import javax.annotation.Nonnull;
51 import javax.annotation.Nullable;
52 import javax.annotation.concurrent.GuardedBy;
53 import org.opendaylight.controller.cluster.ActorSystemProvider;
54 import org.opendaylight.controller.cluster.access.concepts.MemberName;
55 import org.opendaylight.controller.cluster.databroker.actors.dds.DataStoreClient;
56 import org.opendaylight.controller.cluster.databroker.actors.dds.SimpleDataStoreClientActor;
57 import org.opendaylight.controller.cluster.datastore.AbstractDataStore;
58 import org.opendaylight.controller.cluster.datastore.Shard;
59 import org.opendaylight.controller.cluster.datastore.config.Configuration;
60 import org.opendaylight.controller.cluster.datastore.config.ModuleShardConfiguration;
61 import org.opendaylight.controller.cluster.datastore.messages.CreateShard;
62 import org.opendaylight.controller.cluster.datastore.shardstrategy.ModuleShardStrategy;
63 import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
64 import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
65 import org.opendaylight.controller.cluster.dom.api.CDSDataTreeProducer;
66 import org.opendaylight.controller.cluster.dom.api.CDSShardAccess;
67 import org.opendaylight.controller.cluster.sharding.ShardedDataTreeActor.ShardedDataTreeActorCreator;
68 import org.opendaylight.controller.cluster.sharding.messages.InitConfigListener;
69 import org.opendaylight.controller.cluster.sharding.messages.LookupPrefixShard;
70 import org.opendaylight.controller.cluster.sharding.messages.PrefixShardRemovalLookup;
71 import org.opendaylight.controller.cluster.sharding.messages.ProducerCreated;
72 import org.opendaylight.controller.cluster.sharding.messages.ProducerRemoved;
73 import org.opendaylight.controller.cluster.sharding.messages.StartConfigShardLookup;
74 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
75 import org.opendaylight.mdsal.dom.api.DOMDataTreeCursorAwareTransaction;
76 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
77 import org.opendaylight.mdsal.dom.api.DOMDataTreeListener;
78 import org.opendaylight.mdsal.dom.api.DOMDataTreeLoopException;
79 import org.opendaylight.mdsal.dom.api.DOMDataTreeProducer;
80 import org.opendaylight.mdsal.dom.api.DOMDataTreeProducerException;
81 import org.opendaylight.mdsal.dom.api.DOMDataTreeService;
82 import org.opendaylight.mdsal.dom.api.DOMDataTreeServiceExtension;
83 import org.opendaylight.mdsal.dom.api.DOMDataTreeShard;
84 import org.opendaylight.mdsal.dom.api.DOMDataTreeShardingConflictException;
85 import org.opendaylight.mdsal.dom.api.DOMDataTreeShardingService;
86 import org.opendaylight.mdsal.dom.broker.DOMDataTreeShardRegistration;
87 import org.opendaylight.mdsal.dom.broker.ShardedDOMDataTree;
88 import org.opendaylight.mdsal.dom.spi.DOMDataTreePrefixTable;
89 import org.opendaylight.mdsal.dom.spi.DOMDataTreePrefixTableEntry;
90 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.clustering.prefix.shard.configuration.rev170110.PrefixShards;
91 import org.opendaylight.yangtools.concepts.ListenerRegistration;
92 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
93 import org.slf4j.Logger;
94 import org.slf4j.LoggerFactory;
95 import scala.compat.java8.FutureConverters;
96 import scala.concurrent.Future;
97 import scala.concurrent.Promise;
98 import scala.concurrent.duration.FiniteDuration;
101 * A layer on top of DOMDataTreeService that distributes producer/shard registrations to remote nodes via
102 * {@link ShardedDataTreeActor}. Also provides QoL method for addition of prefix based clustered shard into the system.
104 public class DistributedShardedDOMDataTree implements DOMDataTreeService, DOMDataTreeShardingService,
105 DistributedShardFactory {
107 private static final Logger LOG = LoggerFactory.getLogger(DistributedShardedDOMDataTree.class);
109 private static final int MAX_ACTOR_CREATION_RETRIES = 100;
110 private static final int ACTOR_RETRY_DELAY = 100;
111 private static final TimeUnit ACTOR_RETRY_TIME_UNIT = TimeUnit.MILLISECONDS;
112 private static final int LOOKUP_TASK_MAX_RETRIES = 100;
113 static final FiniteDuration SHARD_FUTURE_TIMEOUT_DURATION =
114 new FiniteDuration(LOOKUP_TASK_MAX_RETRIES * LOOKUP_TASK_MAX_RETRIES * 3, TimeUnit.SECONDS);
115 static final Timeout SHARD_FUTURE_TIMEOUT = new Timeout(SHARD_FUTURE_TIMEOUT_DURATION);
117 static final String ACTOR_ID = "ShardedDOMDataTreeFrontend";
119 private final ShardedDOMDataTree shardedDOMDataTree;
120 private final ActorSystem actorSystem;
121 private final AbstractDataStore distributedOperDatastore;
122 private final AbstractDataStore distributedConfigDatastore;
124 private final ActorRef shardedDataTreeActor;
125 private final MemberName memberName;
128 private final DOMDataTreePrefixTable<DOMDataTreeShardRegistration<DOMDataTreeShard>> shards =
129 DOMDataTreePrefixTable.create();
131 private final EnumMap<LogicalDatastoreType, Entry<DataStoreClient, ActorRef>> configurationShardMap =
132 new EnumMap<>(LogicalDatastoreType.class);
134 private final EnumMap<LogicalDatastoreType, PrefixedShardConfigWriter> writerMap =
135 new EnumMap<>(LogicalDatastoreType.class);
137 private final PrefixedShardConfigUpdateHandler updateHandler;
139 public DistributedShardedDOMDataTree(final ActorSystemProvider actorSystemProvider,
140 final AbstractDataStore distributedOperDatastore,
141 final AbstractDataStore distributedConfigDatastore) {
142 this.actorSystem = Preconditions.checkNotNull(actorSystemProvider).getActorSystem();
143 this.distributedOperDatastore = Preconditions.checkNotNull(distributedOperDatastore);
144 this.distributedConfigDatastore = Preconditions.checkNotNull(distributedConfigDatastore);
145 shardedDOMDataTree = new ShardedDOMDataTree();
147 shardedDataTreeActor = createShardedDataTreeActor(actorSystem,
148 new ShardedDataTreeActorCreator()
149 .setShardingService(this)
150 .setActorSystem(actorSystem)
151 .setClusterWrapper(distributedConfigDatastore.getActorUtils().getClusterWrapper())
152 .setDistributedConfigDatastore(distributedConfigDatastore)
153 .setDistributedOperDatastore(distributedOperDatastore)
154 .setLookupTaskMaxRetries(LOOKUP_TASK_MAX_RETRIES),
157 this.memberName = distributedConfigDatastore.getActorUtils().getCurrentMemberName();
159 updateHandler = new PrefixedShardConfigUpdateHandler(shardedDataTreeActor,
160 distributedConfigDatastore.getActorUtils().getCurrentMemberName());
162 LOG.debug("{} - Starting prefix configuration shards", memberName);
163 createPrefixConfigShard(distributedConfigDatastore);
164 createPrefixConfigShard(distributedOperDatastore);
167 private static void createPrefixConfigShard(final AbstractDataStore dataStore) {
168 Configuration configuration = dataStore.getActorUtils().getConfiguration();
169 Collection<MemberName> memberNames = configuration.getUniqueMemberNamesForAllShards();
170 CreateShard createShardMessage =
171 new CreateShard(new ModuleShardConfiguration(PrefixShards.QNAME.getNamespace(),
172 "prefix-shard-configuration", ClusterUtils.PREFIX_CONFIG_SHARD_ID, ModuleShardStrategy.NAME,
174 Shard.builder(), dataStore.getActorUtils().getDatastoreContext());
176 dataStore.getActorUtils().getShardManager().tell(createShardMessage, noSender());
180 * This will try to initialize prefix configuration shards upon their
181 * successful start. We need to create writers to these shards, so we can
182 * satisfy future {@link #createDistributedShard} and
183 * {@link #resolveShardAdditions} requests and update prefix configuration
184 * shards accordingly.
187 * We also need to initialize listeners on these shards, so we can react
188 * on changes made on them by other cluster members or even by ourselves.
191 * Finally, we need to be sure that default shards for both operational and
192 * configuration data stores are up and running and we have distributed
193 * shards frontend created for them.
196 * This is intended to be invoked by blueprint as initialization method.
199 // create our writers to the configuration
201 LOG.debug("{} - starting config shard lookup.", memberName);
203 // We have to wait for prefix config shards to be up and running
204 // so we can create datastore clients for them
205 handleConfigShardLookup().get(SHARD_FUTURE_TIMEOUT_DURATION.length(), SHARD_FUTURE_TIMEOUT_DURATION.unit());
206 } catch (InterruptedException | ExecutionException | TimeoutException e) {
207 throw new IllegalStateException("Prefix config shards not found", e);
211 LOG.debug("{}: Prefix configuration shards ready - creating clients", memberName);
212 configurationShardMap.put(LogicalDatastoreType.CONFIGURATION,
213 createDatastoreClient(ClusterUtils.PREFIX_CONFIG_SHARD_ID,
214 distributedConfigDatastore.getActorUtils()));
215 } catch (final DOMDataTreeShardCreationFailedException e) {
216 throw new IllegalStateException(
217 "Unable to create datastoreClient for config DS prefix configuration shard.", e);
221 configurationShardMap.put(LogicalDatastoreType.OPERATIONAL,
222 createDatastoreClient(ClusterUtils.PREFIX_CONFIG_SHARD_ID,
223 distributedOperDatastore.getActorUtils()));
225 } catch (final DOMDataTreeShardCreationFailedException e) {
226 throw new IllegalStateException(
227 "Unable to create datastoreClient for oper DS prefix configuration shard.", e);
230 writerMap.put(LogicalDatastoreType.CONFIGURATION, new PrefixedShardConfigWriter(
231 configurationShardMap.get(LogicalDatastoreType.CONFIGURATION).getKey()));
233 writerMap.put(LogicalDatastoreType.OPERATIONAL, new PrefixedShardConfigWriter(
234 configurationShardMap.get(LogicalDatastoreType.OPERATIONAL).getKey()));
236 updateHandler.initListener(distributedConfigDatastore, LogicalDatastoreType.CONFIGURATION);
237 updateHandler.initListener(distributedOperDatastore, LogicalDatastoreType.OPERATIONAL);
239 distributedConfigDatastore.getActorUtils().getShardManager().tell(InitConfigListener.INSTANCE, noSender());
240 distributedOperDatastore.getActorUtils().getShardManager().tell(InitConfigListener.INSTANCE, noSender());
243 //create shard registration for DEFAULT_SHARD
244 initDefaultShard(LogicalDatastoreType.CONFIGURATION);
245 initDefaultShard(LogicalDatastoreType.OPERATIONAL);
248 private ListenableFuture<List<Void>> handleConfigShardLookup() {
250 final ListenableFuture<Void> configFuture = lookupConfigShard(LogicalDatastoreType.CONFIGURATION);
251 final ListenableFuture<Void> operFuture = lookupConfigShard(LogicalDatastoreType.OPERATIONAL);
253 return Futures.allAsList(configFuture, operFuture);
256 private ListenableFuture<Void> lookupConfigShard(final LogicalDatastoreType type) {
257 final SettableFuture<Void> future = SettableFuture.create();
259 final Future<Object> ask =
260 Patterns.ask(shardedDataTreeActor, new StartConfigShardLookup(type), SHARD_FUTURE_TIMEOUT);
262 ask.onComplete(new OnComplete<Object>() {
264 public void onComplete(final Throwable throwable, final Object result) {
265 if (throwable != null) {
266 future.setException(throwable);
271 }, actorSystem.dispatcher());
278 public <T extends DOMDataTreeListener> ListenerRegistration<T> registerListener(
279 final T listener, final Collection<DOMDataTreeIdentifier> subtrees,
280 final boolean allowRxMerges, final Collection<DOMDataTreeProducer> producers)
281 throws DOMDataTreeLoopException {
282 return shardedDOMDataTree.registerListener(listener, subtrees, allowRxMerges, producers);
286 public ClassToInstanceMap<DOMDataTreeServiceExtension> getExtensions() {
287 return ImmutableClassToInstanceMap.of();
292 public DOMDataTreeProducer createProducer(@Nonnull final Collection<DOMDataTreeIdentifier> subtrees) {
293 LOG.debug("{} - Creating producer for {}", memberName, subtrees);
294 final DOMDataTreeProducer producer = shardedDOMDataTree.createProducer(subtrees);
296 final Object response = distributedConfigDatastore.getActorUtils()
297 .executeOperation(shardedDataTreeActor, new ProducerCreated(subtrees));
298 if (response == null) {
299 LOG.debug("{} - Received success from remote nodes, creating producer:{}", memberName, subtrees);
300 return new ProxyProducer(producer, subtrees, shardedDataTreeActor,
301 distributedConfigDatastore.getActorUtils(), shards);
304 closeProducer(producer);
306 if (response instanceof Throwable) {
307 Throwables.throwIfUnchecked((Throwable) response);
308 throw new RuntimeException((Throwable) response);
310 throw new RuntimeException("Unexpected response to create producer received." + response);
314 public CompletionStage<DistributedShardRegistration> createDistributedShard(
315 final DOMDataTreeIdentifier prefix, final Collection<MemberName> replicaMembers)
316 throws DOMDataTreeShardingConflictException {
318 synchronized (shards) {
319 final DOMDataTreePrefixTableEntry<DOMDataTreeShardRegistration<DOMDataTreeShard>> lookup =
320 shards.lookup(prefix);
321 if (lookup != null && lookup.getValue().getPrefix().equals(prefix)) {
322 throw new DOMDataTreeShardingConflictException(
323 "Prefix " + prefix + " is already occupied by another shard.");
327 final PrefixedShardConfigWriter writer = writerMap.get(prefix.getDatastoreType());
329 final ListenableFuture<Void> writeFuture =
330 writer.writeConfig(prefix.getRootIdentifier(), replicaMembers);
332 final Promise<DistributedShardRegistration> shardRegistrationPromise = akka.dispatch.Futures.promise();
333 Futures.addCallback(writeFuture, new FutureCallback<Void>() {
335 public void onSuccess(@Nullable final Void result) {
337 final Future<Object> ask =
338 Patterns.ask(shardedDataTreeActor, new LookupPrefixShard(prefix), SHARD_FUTURE_TIMEOUT);
340 shardRegistrationPromise.completeWith(ask.transform(
341 new Mapper<Object, DistributedShardRegistration>() {
343 public DistributedShardRegistration apply(final Object parameter) {
344 return new DistributedShardRegistrationImpl(
345 prefix, shardedDataTreeActor, DistributedShardedDOMDataTree.this);
348 new Mapper<Throwable, Throwable>() {
350 public Throwable apply(final Throwable throwable) {
351 return new DOMDataTreeShardCreationFailedException(
352 "Unable to create a cds shard.", throwable);
354 }, actorSystem.dispatcher()));
358 public void onFailure(final Throwable throwable) {
359 shardRegistrationPromise.failure(
360 new DOMDataTreeShardCreationFailedException("Unable to create a cds shard.", throwable));
362 }, MoreExecutors.directExecutor());
364 return FutureConverters.toJava(shardRegistrationPromise.future());
367 void resolveShardAdditions(final Set<DOMDataTreeIdentifier> additions) {
368 LOG.debug("{}: Resolving additions : {}", memberName, additions);
369 // we need to register the shards from top to bottom, so we need to atleast make sure the ordering reflects that
372 .sorted(Comparator.comparingInt(o -> o.getRootIdentifier().getPathArguments().size()))
373 .forEachOrdered(this::createShardFrontend);
376 void resolveShardRemovals(final Set<DOMDataTreeIdentifier> removals) {
377 LOG.debug("{}: Resolving removals : {}", memberName, removals);
379 // do we need to go from bottom to top?
380 removals.forEach(this::despawnShardFrontend);
383 private void createShardFrontend(final DOMDataTreeIdentifier prefix) {
384 LOG.debug("{}: Creating CDS shard for prefix: {}", memberName, prefix);
385 final String shardName = ClusterUtils.getCleanShardName(prefix.getRootIdentifier());
386 final AbstractDataStore distributedDataStore =
387 prefix.getDatastoreType().equals(org.opendaylight.mdsal.common.api.LogicalDatastoreType.CONFIGURATION)
388 ? distributedConfigDatastore : distributedOperDatastore;
390 try (DOMDataTreeProducer producer = localCreateProducer(Collections.singletonList(prefix))) {
391 final Entry<DataStoreClient, ActorRef> entry =
392 createDatastoreClient(shardName, distributedDataStore.getActorUtils());
394 final DistributedShardFrontend shard =
395 new DistributedShardFrontend(distributedDataStore, entry.getKey(), prefix);
397 final DOMDataTreeShardRegistration<DOMDataTreeShard> reg =
398 shardedDOMDataTree.registerDataTreeShard(prefix, shard, producer);
400 synchronized (shards) {
401 shards.store(prefix, reg);
404 } catch (final DOMDataTreeShardingConflictException e) {
405 LOG.error("{}: Prefix {} is already occupied by another shard",
406 distributedConfigDatastore.getActorUtils().getClusterWrapper().getCurrentMemberName(), prefix, e);
407 } catch (DOMDataTreeProducerException e) {
408 LOG.error("Unable to close producer", e);
409 } catch (DOMDataTreeShardCreationFailedException e) {
410 LOG.error("Unable to create datastore client for shard {}", prefix, e);
414 private void despawnShardFrontend(final DOMDataTreeIdentifier prefix) {
415 LOG.debug("{}: Removing CDS shard for prefix: {}", memberName, prefix);
416 final DOMDataTreePrefixTableEntry<DOMDataTreeShardRegistration<DOMDataTreeShard>> lookup;
417 synchronized (shards) {
418 lookup = shards.lookup(prefix);
421 if (lookup == null || !lookup.getValue().getPrefix().equals(prefix)) {
422 LOG.debug("{}: Received despawn for non-existing CDS shard frontend, prefix: {}, ignoring..",
427 lookup.getValue().close();
428 // need to remove from our local table thats used for tracking
429 synchronized (shards) {
430 shards.remove(prefix);
433 final PrefixedShardConfigWriter writer = writerMap.get(prefix.getDatastoreType());
434 final ListenableFuture<Void> future = writer.removeConfig(prefix.getRootIdentifier());
436 Futures.addCallback(future, new FutureCallback<Void>() {
438 public void onSuccess(@Nullable final Void result) {
439 LOG.debug("{} - Succesfuly removed shard for {}", memberName, prefix);
443 public void onFailure(final Throwable throwable) {
444 LOG.error("Removal of shard {} from configuration failed.", prefix, throwable);
446 }, MoreExecutors.directExecutor());
449 DOMDataTreePrefixTableEntry<DOMDataTreeShardRegistration<DOMDataTreeShard>> lookupShardFrontend(
450 final DOMDataTreeIdentifier prefix) {
451 synchronized (shards) {
452 return shards.lookup(prefix);
456 DOMDataTreeProducer localCreateProducer(final Collection<DOMDataTreeIdentifier> prefix) {
457 return shardedDOMDataTree.createProducer(prefix);
462 public <T extends DOMDataTreeShard> ListenerRegistration<T> registerDataTreeShard(
463 @Nonnull final DOMDataTreeIdentifier prefix,
464 @Nonnull final T shard,
465 @Nonnull final DOMDataTreeProducer producer)
466 throws DOMDataTreeShardingConflictException {
468 LOG.debug("Registering shard[{}] at prefix: {}", shard, prefix);
470 if (producer instanceof ProxyProducer) {
471 return shardedDOMDataTree.registerDataTreeShard(prefix, shard, ((ProxyProducer) producer).delegate());
474 return shardedDOMDataTree.registerDataTreeShard(prefix, shard, producer);
477 @SuppressWarnings("checkstyle:IllegalCatch")
478 private Entry<DataStoreClient, ActorRef> createDatastoreClient(final String shardName, final ActorUtils actorUtils)
479 throws DOMDataTreeShardCreationFailedException {
481 LOG.debug("{}: Creating distributed datastore client for shard {}", memberName, shardName);
482 final Props distributedDataStoreClientProps =
483 SimpleDataStoreClientActor.props(memberName, "Shard-" + shardName, actorUtils, shardName);
485 final ActorRef clientActor = actorSystem.actorOf(distributedDataStoreClientProps);
487 return new SimpleEntry<>(SimpleDataStoreClientActor
488 .getDistributedDataStoreClient(clientActor, 30, TimeUnit.SECONDS), clientActor);
489 } catch (final Exception e) {
490 LOG.error("{}: Failed to get actor for {}", distributedDataStoreClientProps, memberName, e);
491 clientActor.tell(PoisonPill.getInstance(), noSender());
492 throw new DOMDataTreeShardCreationFailedException(
493 "Unable to create datastore client for shard{" + shardName + "}", e);
497 @SuppressWarnings("checkstyle:IllegalCatch")
498 private void initDefaultShard(final LogicalDatastoreType logicalDatastoreType) {
500 final PrefixedShardConfigWriter writer = writerMap.get(logicalDatastoreType);
502 if (writer.checkDefaultIsPresent()) {
503 LOG.debug("{}: Default shard for {} is already present in the config. Possibly saved in snapshot.",
504 memberName, logicalDatastoreType);
507 // Currently the default shard configuration is present in the out-of-box modules.conf and is
508 // expected to be present. So look up the local default shard here and create the frontend.
510 // TODO we don't have to do it for config and operational default shard separately. Just one of them
512 final ActorUtils actorUtils = logicalDatastoreType == LogicalDatastoreType.CONFIGURATION
513 ? distributedConfigDatastore.getActorUtils() : distributedOperDatastore.getActorUtils();
515 final Optional<ActorRef> defaultLocalShardOptional =
516 actorUtils.findLocalShard(ClusterUtils.getCleanShardName(YangInstanceIdentifier.EMPTY));
518 if (defaultLocalShardOptional.isPresent()) {
519 LOG.debug("{}: Default shard for {} is already started, creating just frontend", memberName,
520 logicalDatastoreType);
521 createShardFrontend(new DOMDataTreeIdentifier(logicalDatastoreType, YangInstanceIdentifier.EMPTY));
524 // The local shard isn't present - we assume that means the local member isn't in the replica list
525 // and will be dynamically created later via an explicit add-shard-replica request. This is the
526 // bootstrapping mechanism to add a new node into an existing cluster. The following code to create
527 // the default shard as a prefix shard is problematic in this scenario so it is commented out. Since
528 // the default shard is a module-based shard by default, it makes sense to always treat it as such,
529 // ie bootstrap it in the same manner as the special prefix-configuration and EOS shards.
530 // final Collection<MemberName> names = distributedConfigDatastore.getActorUtils().getConfiguration()
531 // .getUniqueMemberNamesForAllShards();
532 // Await.result(FutureConverters.toScala(createDistributedShard(
533 // new DOMDataTreeIdentifier(logicalDatastoreType, YangInstanceIdentifier.EMPTY), names)),
534 // SHARD_FUTURE_TIMEOUT_DURATION);
535 // } catch (DOMDataTreeShardingConflictException e) {
536 // LOG.debug("{}: Default shard for {} already registered, possibly due to other node doing it faster",
537 // memberName, logicalDatastoreType);
538 } catch (Exception e) {
539 LOG.error("{}: Default shard initialization for {} failed", memberName, logicalDatastoreType, e);
540 throw new RuntimeException(e);
545 private static void closeProducer(final DOMDataTreeProducer producer) {
548 } catch (final DOMDataTreeProducerException e) {
549 LOG.error("Unable to close producer", e);
553 @SuppressWarnings("checkstyle:IllegalCatch")
554 private static ActorRef createShardedDataTreeActor(final ActorSystem actorSystem,
555 final ShardedDataTreeActorCreator creator,
556 final String shardDataTreeActorId) {
557 Exception lastException = null;
559 for (int i = 0; i < MAX_ACTOR_CREATION_RETRIES; i++) {
561 return actorSystem.actorOf(creator.props(), shardDataTreeActorId);
562 } catch (final Exception e) {
564 Uninterruptibles.sleepUninterruptibly(ACTOR_RETRY_DELAY, ACTOR_RETRY_TIME_UNIT);
565 LOG.debug("Could not create actor {} because of {} -"
566 + " waiting for sometime before retrying (retry count = {})",
567 shardDataTreeActorId, e.getMessage(), i);
571 throw new IllegalStateException("Failed to create actor for ShardedDOMDataTree", lastException);
574 private class DistributedShardRegistrationImpl implements DistributedShardRegistration {
576 private final DOMDataTreeIdentifier prefix;
577 private final ActorRef shardedDataTreeActor;
578 private final DistributedShardedDOMDataTree distributedShardedDOMDataTree;
580 DistributedShardRegistrationImpl(final DOMDataTreeIdentifier prefix,
581 final ActorRef shardedDataTreeActor,
582 final DistributedShardedDOMDataTree distributedShardedDOMDataTree) {
583 this.prefix = prefix;
584 this.shardedDataTreeActor = shardedDataTreeActor;
585 this.distributedShardedDOMDataTree = distributedShardedDOMDataTree;
589 public CompletionStage<Void> close() {
590 // first despawn on the local node
591 distributedShardedDOMDataTree.despawnShardFrontend(prefix);
592 // update the config so the remote nodes are updated
593 final Future<Object> ask =
594 Patterns.ask(shardedDataTreeActor, new PrefixShardRemovalLookup(prefix), SHARD_FUTURE_TIMEOUT);
596 final Future<Void> closeFuture = ask.transform(
597 new Mapper<Object, Void>() {
599 public Void apply(final Object parameter) {
603 new Mapper<Throwable, Throwable>() {
605 public Throwable apply(final Throwable throwable) {
608 }, actorSystem.dispatcher());
610 return FutureConverters.toJava(closeFuture);
614 // TODO what about producers created by this producer?
615 // They should also be CDSProducers
616 private static final class ProxyProducer extends ForwardingObject implements CDSDataTreeProducer {
618 private final DOMDataTreeProducer delegate;
619 private final Collection<DOMDataTreeIdentifier> subtrees;
620 private final ActorRef shardDataTreeActor;
621 private final ActorUtils actorUtils;
622 @GuardedBy("shardAccessMap")
623 private final Map<DOMDataTreeIdentifier, CDSShardAccessImpl> shardAccessMap = new HashMap<>();
625 // We don't have to guard access to shardTable in ProxyProducer.
626 // ShardTable's entries relevant to this ProxyProducer shouldn't
627 // change during producer's lifetime.
628 private final DOMDataTreePrefixTable<DOMDataTreeShardRegistration<DOMDataTreeShard>> shardTable;
630 ProxyProducer(final DOMDataTreeProducer delegate,
631 final Collection<DOMDataTreeIdentifier> subtrees,
632 final ActorRef shardDataTreeActor,
633 final ActorUtils actorUtils,
634 final DOMDataTreePrefixTable<DOMDataTreeShardRegistration<DOMDataTreeShard>> shardLayout) {
635 this.delegate = requireNonNull(delegate);
636 this.subtrees = requireNonNull(subtrees);
637 this.shardDataTreeActor = requireNonNull(shardDataTreeActor);
638 this.actorUtils = requireNonNull(actorUtils);
639 this.shardTable = requireNonNull(shardLayout);
644 public DOMDataTreeCursorAwareTransaction createTransaction(final boolean isolated) {
645 return delegate.createTransaction(isolated);
650 @SuppressWarnings("checkstyle:hiddenField")
651 public DOMDataTreeProducer createProducer(@Nonnull final Collection<DOMDataTreeIdentifier> subtrees) {
652 // TODO we probably don't need to distribute this on the remote nodes since once we have this producer
653 // open we surely have the rights to all the subtrees.
654 return delegate.createProducer(subtrees);
658 @SuppressWarnings("checkstyle:IllegalCatch")
659 public void close() throws DOMDataTreeProducerException {
662 synchronized (shardAccessMap) {
663 shardAccessMap.values().forEach(CDSShardAccessImpl::close);
666 final Object o = actorUtils.executeOperation(shardDataTreeActor, new ProducerRemoved(subtrees));
667 if (o instanceof DOMDataTreeProducerException) {
668 throw (DOMDataTreeProducerException) o;
669 } else if (o instanceof Throwable) {
670 throw new DOMDataTreeProducerException("Unable to close producer", (Throwable) o);
675 protected DOMDataTreeProducer delegate() {
681 public CDSShardAccess getShardAccess(@Nonnull final DOMDataTreeIdentifier subtree) {
682 checkArgument(subtrees.stream().anyMatch(dataTreeIdentifier -> dataTreeIdentifier.contains(subtree)),
683 "Subtree %s is not controlled by this producer %s", subtree, this);
685 final DOMDataTreePrefixTableEntry<DOMDataTreeShardRegistration<DOMDataTreeShard>> lookup =
686 shardTable.lookup(subtree);
687 checkState(lookup != null, "Subtree %s is not contained in any registered shard.", subtree);
689 final DOMDataTreeIdentifier lookupId = lookup.getValue().getPrefix();
691 synchronized (shardAccessMap) {
692 if (shardAccessMap.get(lookupId) != null) {
693 return shardAccessMap.get(lookupId);
696 // TODO Maybe we can have static factory method and return the same instance
697 // for same subtrees. But maybe it is not needed since there can be only one
698 // producer attached to some subtree at a time. And also how we can close ShardAccess
700 final CDSShardAccessImpl shardAccess = new CDSShardAccessImpl(lookupId, actorUtils);
701 shardAccessMap.put(lookupId, shardAccess);