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

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.