Convert sal-distributed-datastore to OSGi DS
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / sharding / ShardedDataTreeActor.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.sharding;
9
10 import static java.util.Objects.requireNonNull;
11
12 import akka.actor.ActorRef;
13 import akka.actor.ActorSelection;
14 import akka.actor.ActorSystem;
15 import akka.actor.PoisonPill;
16 import akka.actor.Props;
17 import akka.actor.Status;
18 import akka.actor.Status.Success;
19 import akka.cluster.ClusterEvent;
20 import akka.cluster.ClusterEvent.MemberExited;
21 import akka.cluster.ClusterEvent.MemberRemoved;
22 import akka.cluster.ClusterEvent.MemberUp;
23 import akka.cluster.ClusterEvent.MemberWeaklyUp;
24 import akka.cluster.ClusterEvent.ReachableMember;
25 import akka.cluster.ClusterEvent.UnreachableMember;
26 import akka.cluster.Member;
27 import akka.dispatch.OnComplete;
28 import akka.pattern.Patterns;
29 import akka.util.Timeout;
30 import java.util.ArrayList;
31 import java.util.Collection;
32 import java.util.Collections;
33 import java.util.HashMap;
34 import java.util.List;
35 import java.util.Map;
36 import java.util.Optional;
37 import java.util.concurrent.CompletableFuture;
38 import java.util.concurrent.TimeUnit;
39 import org.opendaylight.controller.cluster.access.concepts.MemberName;
40 import org.opendaylight.controller.cluster.common.actor.AbstractUntypedPersistentActor;
41 import org.opendaylight.controller.cluster.datastore.ClusterWrapper;
42 import org.opendaylight.controller.cluster.datastore.DistributedDataStoreInterface;
43 import org.opendaylight.controller.cluster.datastore.config.PrefixShardConfiguration;
44 import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
45 import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
46 import org.opendaylight.controller.cluster.raft.client.messages.FindLeader;
47 import org.opendaylight.controller.cluster.raft.client.messages.FindLeaderReply;
48 import org.opendaylight.controller.cluster.sharding.messages.LookupPrefixShard;
49 import org.opendaylight.controller.cluster.sharding.messages.NotifyProducerCreated;
50 import org.opendaylight.controller.cluster.sharding.messages.NotifyProducerRemoved;
51 import org.opendaylight.controller.cluster.sharding.messages.PrefixShardCreated;
52 import org.opendaylight.controller.cluster.sharding.messages.PrefixShardRemovalLookup;
53 import org.opendaylight.controller.cluster.sharding.messages.PrefixShardRemoved;
54 import org.opendaylight.controller.cluster.sharding.messages.ProducerCreated;
55 import org.opendaylight.controller.cluster.sharding.messages.ProducerRemoved;
56 import org.opendaylight.controller.cluster.sharding.messages.StartConfigShardLookup;
57 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
58 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
59 import org.opendaylight.mdsal.dom.api.DOMDataTreeProducer;
60 import org.opendaylight.mdsal.dom.api.DOMDataTreeProducerException;
61 import org.opendaylight.mdsal.dom.api.DOMDataTreeShard;
62 import org.opendaylight.mdsal.dom.broker.DOMDataTreeShardRegistration;
63 import org.opendaylight.mdsal.dom.spi.DOMDataTreePrefixTableEntry;
64 import org.opendaylight.yangtools.concepts.AbstractObjectRegistration;
65 import org.opendaylight.yangtools.concepts.ListenerRegistration;
66 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
67 import org.slf4j.Logger;
68 import org.slf4j.LoggerFactory;
69 import scala.compat.java8.FutureConverters;
70 import scala.concurrent.Future;
71 import scala.concurrent.duration.FiniteDuration;
72
73 /**
74  * Actor that tracks currently open producers/shards on remote nodes and handles notifications of remote
75  * nodes of newly open producers/shards on the local node.
76  */
77 public class ShardedDataTreeActor extends AbstractUntypedPersistentActor {
78
79     private static final Logger LOG = LoggerFactory.getLogger(ShardedDataTreeActor.class);
80
81     private static final String PERSISTENCE_ID = "sharding-service-actor";
82     private static final Timeout DEFAULT_ASK_TIMEOUT = new Timeout(15, TimeUnit.SECONDS);
83
84     static final FiniteDuration SHARD_LOOKUP_TASK_INTERVAL = new FiniteDuration(1L, TimeUnit.SECONDS);
85
86     private final DistributedShardedDOMDataTree shardingService;
87     private final ActorSystem actorSystem;
88     private final ClusterWrapper clusterWrapper;
89     // helper actorContext used only for static calls to executeAsync etc
90     // for calls that need specific actor context tied to a datastore use the one provided in the DistributedDataStore
91     private final ActorUtils actorUtils;
92     private final ShardingServiceAddressResolver resolver;
93     private final DistributedDataStoreInterface distributedConfigDatastore;
94     private final DistributedDataStoreInterface distributedOperDatastore;
95     private final int lookupTaskMaxRetries;
96
97     private final Map<DOMDataTreeIdentifier, ActorProducerRegistration> idToProducer = new HashMap<>();
98
99     ShardedDataTreeActor(final ShardedDataTreeActorCreator builder) {
100         LOG.debug("Creating ShardedDataTreeActor on {}", builder.getClusterWrapper().getCurrentMemberName());
101
102         shardingService = builder.getShardingService();
103         actorSystem = builder.getActorSystem();
104         clusterWrapper = builder.getClusterWrapper();
105         distributedConfigDatastore = builder.getDistributedConfigDatastore();
106         distributedOperDatastore = builder.getDistributedOperDatastore();
107         lookupTaskMaxRetries = builder.getLookupTaskMaxRetries();
108         actorUtils = distributedConfigDatastore.getActorUtils();
109         resolver = new ShardingServiceAddressResolver(
110                 DistributedShardedDOMDataTree.ACTOR_ID, clusterWrapper.getCurrentMemberName());
111
112         clusterWrapper.subscribeToMemberEvents(self());
113     }
114
115     @Override
116     public void preStart() {
117     }
118
119     @Override
120     protected void handleRecover(final Object message) {
121         LOG.debug("Received a recover message {}", message);
122     }
123
124     @Override
125     protected void handleCommand(final Object message) {
126         LOG.debug("{} : Received {}", clusterWrapper.getCurrentMemberName(), message);
127         if (message instanceof ClusterEvent.MemberUp) {
128             memberUp((ClusterEvent.MemberUp) message);
129         } else if (message instanceof ClusterEvent.MemberWeaklyUp) {
130             memberWeaklyUp((ClusterEvent.MemberWeaklyUp) message);
131         } else if (message instanceof ClusterEvent.MemberExited) {
132             memberExited((ClusterEvent.MemberExited) message);
133         } else if (message instanceof ClusterEvent.MemberRemoved) {
134             memberRemoved((ClusterEvent.MemberRemoved) message);
135         } else if (message instanceof ClusterEvent.UnreachableMember) {
136             memberUnreachable((ClusterEvent.UnreachableMember) message);
137         } else if (message instanceof ClusterEvent.ReachableMember) {
138             memberReachable((ClusterEvent.ReachableMember) message);
139         } else if (message instanceof ProducerCreated) {
140             onProducerCreated((ProducerCreated) message);
141         } else if (message instanceof NotifyProducerCreated) {
142             onNotifyProducerCreated((NotifyProducerCreated) message);
143         } else if (message instanceof ProducerRemoved) {
144             onProducerRemoved((ProducerRemoved) message);
145         } else if (message instanceof NotifyProducerRemoved) {
146             onNotifyProducerRemoved((NotifyProducerRemoved) message);
147         } else if (message instanceof PrefixShardCreated) {
148             onPrefixShardCreated((PrefixShardCreated) message);
149         } else if (message instanceof LookupPrefixShard) {
150             onLookupPrefixShard((LookupPrefixShard) message);
151         } else if (message instanceof PrefixShardRemovalLookup) {
152             onPrefixShardRemovalLookup((PrefixShardRemovalLookup) message);
153         } else if (message instanceof PrefixShardRemoved) {
154             onPrefixShardRemoved((PrefixShardRemoved) message);
155         } else if (message instanceof StartConfigShardLookup) {
156             onStartConfigShardLookup((StartConfigShardLookup) message);
157         }
158     }
159
160     @Override
161     public String persistenceId() {
162         return PERSISTENCE_ID;
163     }
164
165     private void memberUp(final MemberUp message) {
166         final MemberName memberName = memberToName(message.member());
167
168         LOG.info("{}: Received MemberUp: memberName: {}, address: {}", persistenceId(), memberName,
169                 message.member().address());
170
171         resolver.addPeerAddress(memberName, message.member().address());
172     }
173
174     private void memberWeaklyUp(final MemberWeaklyUp message) {
175         final MemberName memberName = memberToName(message.member());
176
177         LOG.info("{}: Received MemberWeaklyUp: memberName: {}, address: {}", persistenceId(), memberName,
178                 message.member().address());
179
180         resolver.addPeerAddress(memberName, message.member().address());
181     }
182
183     private void memberExited(final MemberExited message) {
184         final MemberName memberName = memberToName(message.member());
185
186         LOG.info("{}: Received MemberExited: memberName: {}, address: {}", persistenceId(), memberName,
187                 message.member().address());
188
189         resolver.removePeerAddress(memberName);
190     }
191
192     private void memberRemoved(final MemberRemoved message) {
193         final MemberName memberName = memberToName(message.member());
194
195         LOG.info("{}: Received MemberRemoved: memberName: {}, address: {}", persistenceId(), memberName,
196                 message.member().address());
197
198         resolver.removePeerAddress(memberName);
199     }
200
201     private void memberUnreachable(final UnreachableMember message) {
202         final MemberName memberName = memberToName(message.member());
203         LOG.debug("Received UnreachableMember: memberName {}, address: {}", memberName, message.member().address());
204
205         resolver.removePeerAddress(memberName);
206     }
207
208     private void memberReachable(final ReachableMember message) {
209         final MemberName memberName = memberToName(message.member());
210         LOG.debug("Received ReachableMember: memberName {}, address: {}", memberName, message.member().address());
211
212         resolver.addPeerAddress(memberName, message.member().address());
213     }
214
215     private void onProducerCreated(final ProducerCreated message) {
216         LOG.debug("Received ProducerCreated: {}", message);
217
218         // fastpath if we have no peers
219         if (resolver.getShardingServicePeerActorAddresses().isEmpty()) {
220             getSender().tell(new Status.Success(null), ActorRef.noSender());
221         }
222
223         final ActorRef sender = getSender();
224         final Collection<DOMDataTreeIdentifier> subtrees = message.getSubtrees();
225
226         final List<CompletableFuture<Object>> futures = new ArrayList<>();
227
228         for (final String address : resolver.getShardingServicePeerActorAddresses()) {
229             final ActorSelection actorSelection = actorSystem.actorSelection(address);
230             futures.add(
231                     FutureConverters.toJava(
232                             actorUtils.executeOperationAsync(
233                                     actorSelection, new NotifyProducerCreated(subtrees), DEFAULT_ASK_TIMEOUT))
234                     .toCompletableFuture());
235         }
236
237         final CompletableFuture<Void> combinedFuture = CompletableFuture.allOf(
238                 futures.toArray(new CompletableFuture[futures.size()]));
239
240         combinedFuture
241                 .thenRun(() -> sender.tell(new Success(null), ActorRef.noSender()))
242                 .exceptionally(throwable -> {
243                     sender.tell(new Status.Failure(throwable), self());
244                     return null;
245                 });
246     }
247
248     private void onNotifyProducerCreated(final NotifyProducerCreated message) {
249         LOG.debug("Received NotifyProducerCreated: {}", message);
250
251         final Collection<DOMDataTreeIdentifier> subtrees = message.getSubtrees();
252
253         try {
254             final ActorProducerRegistration registration =
255                     new ActorProducerRegistration(shardingService.localCreateProducer(subtrees), subtrees);
256             subtrees.forEach(id -> idToProducer.put(id, registration));
257             sender().tell(new Status.Success(null), self());
258         } catch (final IllegalArgumentException e) {
259             sender().tell(new Status.Failure(e), getSelf());
260         }
261     }
262
263     private void onProducerRemoved(final ProducerRemoved message) {
264         LOG.debug("Received ProducerRemoved: {}", message);
265
266         final List<CompletableFuture<Object>> futures = new ArrayList<>();
267
268         for (final String address : resolver.getShardingServicePeerActorAddresses()) {
269             final ActorSelection selection = actorSystem.actorSelection(address);
270
271             futures.add(FutureConverters.toJava(
272                     actorUtils.executeOperationAsync(selection, new NotifyProducerRemoved(message.getSubtrees())))
273                     .toCompletableFuture());
274         }
275
276         final CompletableFuture<Void> combinedFuture = CompletableFuture.allOf(
277                 futures.toArray(new CompletableFuture[futures.size()]));
278
279         final ActorRef respondTo = getSender();
280
281         combinedFuture
282                 .thenRun(() -> respondTo.tell(new Status.Success(null), self()))
283                 .exceptionally(e -> {
284                     respondTo.tell(new Status.Failure(null), self());
285                     return null;
286                 });
287
288     }
289
290     private void onNotifyProducerRemoved(final NotifyProducerRemoved message) {
291         LOG.debug("Received NotifyProducerRemoved: {}", message);
292
293         final ActorProducerRegistration registration = idToProducer.remove(message.getSubtrees().iterator().next());
294         if (registration == null) {
295             LOG.warn("The notification contained a path on which no producer is registered, throwing away");
296             getSender().tell(new Status.Success(null), ActorRef.noSender());
297             return;
298         }
299
300         try {
301             registration.close();
302             getSender().tell(new Status.Success(null), ActorRef.noSender());
303         } catch (final DOMDataTreeProducerException e) {
304             LOG.error("Unable to close producer", e);
305             getSender().tell(new Status.Failure(e), ActorRef.noSender());
306         }
307     }
308
309     @SuppressWarnings("checkstyle:IllegalCatch")
310     private void onLookupPrefixShard(final LookupPrefixShard message) {
311         LOG.debug("Member: {}, Received LookupPrefixShard: {}", clusterWrapper.getCurrentMemberName(), message);
312
313         final DOMDataTreeIdentifier prefix = message.getPrefix();
314
315         final ActorUtils utils = prefix.getDatastoreType() == LogicalDatastoreType.CONFIGURATION
316                         ? distributedConfigDatastore.getActorUtils() : distributedOperDatastore.getActorUtils();
317
318         // schedule a notification task for the reply
319         actorSystem.scheduler().scheduleOnce(SHARD_LOOKUP_TASK_INTERVAL,
320                 new ShardCreationLookupTask(actorSystem, getSender(), clusterWrapper,
321                         utils, shardingService, prefix, lookupTaskMaxRetries), actorSystem.dispatcher());
322     }
323
324     private void onPrefixShardCreated(final PrefixShardCreated message) {
325         LOG.debug("Member: {}, Received PrefixShardCreated: {}", clusterWrapper.getCurrentMemberName(), message);
326
327         final PrefixShardConfiguration config = message.getConfiguration();
328
329         shardingService.resolveShardAdditions(Collections.singleton(config.getPrefix()));
330     }
331
332     private void onPrefixShardRemovalLookup(final PrefixShardRemovalLookup message) {
333         LOG.debug("Member: {}, Received PrefixShardRemovalLookup: {}", clusterWrapper.getCurrentMemberName(), message);
334
335         final ShardRemovalLookupTask removalTask =
336                 new ShardRemovalLookupTask(actorSystem, getSender(),
337                         actorUtils, message.getPrefix(), lookupTaskMaxRetries);
338
339         actorSystem.scheduler().scheduleOnce(SHARD_LOOKUP_TASK_INTERVAL, removalTask, actorSystem.dispatcher());
340     }
341
342     private void onPrefixShardRemoved(final PrefixShardRemoved message) {
343         LOG.debug("Received PrefixShardRemoved: {}", message);
344
345         shardingService.resolveShardRemovals(Collections.singleton(message.getPrefix()));
346     }
347
348     private void onStartConfigShardLookup(final StartConfigShardLookup message) {
349         LOG.debug("Received StartConfigShardLookup: {}", message);
350
351         final ActorUtils context =
352                 message.getType().equals(LogicalDatastoreType.CONFIGURATION)
353                         ? distributedConfigDatastore.getActorUtils() : distributedOperDatastore.getActorUtils();
354
355         // schedule a notification task for the reply
356         actorSystem.scheduler().scheduleOnce(SHARD_LOOKUP_TASK_INTERVAL,
357                 new ConfigShardLookupTask(
358                         actorSystem, getSender(), context, message, lookupTaskMaxRetries),
359                 actorSystem.dispatcher());
360     }
361
362     private static MemberName memberToName(final Member member) {
363         return MemberName.forName(member.roles().iterator().next());
364     }
365
366     private class ActorProducerRegistration {
367
368         private final DOMDataTreeProducer producer;
369         private final Collection<DOMDataTreeIdentifier> subtrees;
370
371         ActorProducerRegistration(final DOMDataTreeProducer producer,
372                                   final Collection<DOMDataTreeIdentifier> subtrees) {
373             this.producer = producer;
374             this.subtrees = subtrees;
375         }
376
377         void close() throws DOMDataTreeProducerException {
378             producer.close();
379             subtrees.forEach(idToProducer::remove);
380         }
381     }
382
383     private static class ShardFrontendRegistration extends
384             AbstractObjectRegistration<ListenerRegistration<DistributedShardFrontend>> {
385
386         private final ActorRef clientActor;
387         private final ListenerRegistration<DistributedShardFrontend> shardRegistration;
388
389         ShardFrontendRegistration(final ActorRef clientActor,
390                                   final ListenerRegistration<DistributedShardFrontend> shardRegistration) {
391             super(shardRegistration);
392             this.clientActor = clientActor;
393             this.shardRegistration = shardRegistration;
394         }
395
396         @Override
397         protected void removeRegistration() {
398             shardRegistration.close();
399             clientActor.tell(PoisonPill.getInstance(), ActorRef.noSender());
400         }
401     }
402
403     /**
404      * Handles the lookup step of cds shard creation once the configuration is updated.
405      */
406     private static class ShardCreationLookupTask extends LookupTask {
407
408         private final ActorSystem system;
409         private final ActorRef replyTo;
410         private final ClusterWrapper clusterWrapper;
411         private final ActorUtils context;
412         private final DistributedShardedDOMDataTree shardingService;
413         private final DOMDataTreeIdentifier toLookup;
414         private final int lookupMaxRetries;
415
416         ShardCreationLookupTask(final ActorSystem system,
417                                 final ActorRef replyTo,
418                                 final ClusterWrapper clusterWrapper,
419                                 final ActorUtils context,
420                                 final DistributedShardedDOMDataTree shardingService,
421                                 final DOMDataTreeIdentifier toLookup,
422                                 final int lookupMaxRetries) {
423             super(replyTo, lookupMaxRetries);
424             this.system = system;
425             this.replyTo = replyTo;
426             this.clusterWrapper = clusterWrapper;
427             this.context = context;
428             this.shardingService = shardingService;
429             this.toLookup = toLookup;
430             this.lookupMaxRetries = lookupMaxRetries;
431         }
432
433         @Override
434         public void run() {
435             final Future<ActorRef> localShardFuture =
436                     context.findLocalShardAsync(ClusterUtils.getCleanShardName(toLookup.getRootIdentifier()));
437
438             localShardFuture.onComplete(new OnComplete<ActorRef>() {
439                 @Override
440                 public void onComplete(final Throwable throwable, final ActorRef actorRef) {
441                     if (throwable != null) {
442                         tryReschedule(throwable);
443                     } else {
444                         LOG.debug("Local backend for shard[{}] lookup successful, starting leader lookup..", toLookup);
445
446                         system.scheduler().scheduleOnce(
447                                 SHARD_LOOKUP_TASK_INTERVAL,
448                                 new ShardLeaderLookupTask(system, replyTo, context, clusterWrapper, actorRef,
449                                         shardingService, toLookup, lookupMaxRetries),
450                                 system.dispatcher());
451                     }
452                 }
453             }, system.dispatcher());
454         }
455
456         @Override
457         void reschedule(final int retries) {
458             LOG.debug("Local backend for shard[{}] not found, try: {}, rescheduling..", toLookup, retries);
459             system.scheduler().scheduleOnce(
460                     SHARD_LOOKUP_TASK_INTERVAL, ShardCreationLookupTask.this, system.dispatcher());
461         }
462     }
463
464     /**
465      * Handles the readiness step by waiting for a leader of the created shard.
466      */
467     private static class ShardLeaderLookupTask extends LookupTask {
468
469         private final ActorSystem system;
470         private final ActorRef replyTo;
471         private final ActorUtils context;
472         private final ClusterWrapper clusterWrapper;
473         private final ActorRef shard;
474         private final DistributedShardedDOMDataTree shardingService;
475         private final DOMDataTreeIdentifier toLookup;
476         private final int lookupMaxRetries;
477
478         ShardLeaderLookupTask(final ActorSystem system,
479                               final ActorRef replyTo,
480                               final ActorUtils context,
481                               final ClusterWrapper clusterWrapper,
482                               final ActorRef shard,
483                               final DistributedShardedDOMDataTree shardingService,
484                               final DOMDataTreeIdentifier toLookup,
485                               final int lookupMaxRetries) {
486             super(replyTo, lookupMaxRetries);
487             this.system = system;
488             this.replyTo = replyTo;
489             this.context = context;
490             this.clusterWrapper = clusterWrapper;
491             this.shard = shard;
492             this.shardingService = shardingService;
493             this.toLookup = toLookup;
494             this.lookupMaxRetries = lookupMaxRetries;
495         }
496
497         @Override
498         public void run() {
499
500             final Future<Object> ask = Patterns.ask(shard, FindLeader.INSTANCE, context.getOperationTimeout());
501
502             ask.onComplete(new OnComplete<>() {
503                 @Override
504                 public void onComplete(final Throwable throwable, final Object findLeaderReply) {
505                     if (throwable != null) {
506                         tryReschedule(throwable);
507                     } else {
508                         final FindLeaderReply findLeader = (FindLeaderReply) findLeaderReply;
509                         final Optional<String> leaderActor = findLeader.getLeaderActor();
510                         if (leaderActor.isPresent()) {
511                             // leader is found, backend seems ready, check if the frontend is ready
512                             LOG.debug("{} - Leader for shard[{}] backend ready, starting frontend lookup..",
513                                     clusterWrapper.getCurrentMemberName(), toLookup);
514                             system.scheduler().scheduleOnce(
515                                     SHARD_LOOKUP_TASK_INTERVAL,
516                                     new FrontendLookupTask(
517                                             system, replyTo, shardingService, toLookup, lookupMaxRetries),
518                                     system.dispatcher());
519                         } else {
520                             tryReschedule(null);
521                         }
522                     }
523                 }
524             }, system.dispatcher());
525
526         }
527
528         @Override
529         void reschedule(final int retries) {
530             LOG.debug("{} - Leader for shard[{}] backend not found on try: {}, retrying..",
531                     clusterWrapper.getCurrentMemberName(), toLookup, retries);
532             system.scheduler().scheduleOnce(
533                     SHARD_LOOKUP_TASK_INTERVAL, ShardLeaderLookupTask.this, system.dispatcher());
534         }
535     }
536
537     /**
538      * After backend is ready this handles the last step - checking if we have a frontend shard for the backend,
539      * once this completes(which should be ready by the time the backend is created, this is just a sanity check in
540      * case they race), the future for the cds shard creation is completed and the shard is ready for use.
541      */
542     private static final class FrontendLookupTask extends LookupTask {
543
544         private final ActorSystem system;
545         private final ActorRef replyTo;
546         private final DistributedShardedDOMDataTree shardingService;
547         private final DOMDataTreeIdentifier toLookup;
548
549         FrontendLookupTask(final ActorSystem system,
550                            final ActorRef replyTo,
551                            final DistributedShardedDOMDataTree shardingService,
552                            final DOMDataTreeIdentifier toLookup,
553                            final int lookupMaxRetries) {
554             super(replyTo, lookupMaxRetries);
555             this.system = system;
556             this.replyTo = replyTo;
557             this.shardingService = shardingService;
558             this.toLookup = toLookup;
559         }
560
561         @Override
562         public void run() {
563             final DOMDataTreePrefixTableEntry<DOMDataTreeShardRegistration<DOMDataTreeShard>> entry =
564                     shardingService.lookupShardFrontend(toLookup);
565
566             if (entry != null && tableEntryIdCheck(entry, toLookup) && entry.getValue() != null) {
567                 replyTo.tell(new Success(null), ActorRef.noSender());
568             } else {
569                 tryReschedule(null);
570             }
571         }
572
573         private boolean tableEntryIdCheck(final DOMDataTreePrefixTableEntry<?> entry,
574                                           final DOMDataTreeIdentifier prefix) {
575             if (entry == null) {
576                 return false;
577             }
578
579             if (YangInstanceIdentifier.empty().equals(prefix.getRootIdentifier())) {
580                 return true;
581             }
582
583             if (entry.getIdentifier().equals(toLookup.getRootIdentifier().getLastPathArgument())) {
584                 return true;
585             }
586
587             return false;
588         }
589
590         @Override
591         void reschedule(final int retries) {
592             LOG.debug("Frontend for shard[{}] not found on try: {}, retrying..", toLookup, retries);
593             system.scheduler().scheduleOnce(
594                     SHARD_LOOKUP_TASK_INTERVAL, FrontendLookupTask.this, system.dispatcher());
595         }
596     }
597
598     /**
599      * Task that is run once a cds shard registration is closed and completes once the backend shard is removed from the
600      * configuration.
601      */
602     private static class ShardRemovalLookupTask extends LookupTask {
603
604         private final ActorSystem system;
605         private final ActorRef replyTo;
606         private final ActorUtils context;
607         private final DOMDataTreeIdentifier toLookup;
608
609         ShardRemovalLookupTask(final ActorSystem system,
610                                final ActorRef replyTo,
611                                final ActorUtils context,
612                                final DOMDataTreeIdentifier toLookup,
613                                final int lookupMaxRetries) {
614             super(replyTo, lookupMaxRetries);
615             this.system = system;
616             this.replyTo = replyTo;
617             this.context = context;
618             this.toLookup = toLookup;
619         }
620
621         @Override
622         public void run() {
623             final Future<ActorRef> localShardFuture =
624                     context.findLocalShardAsync(ClusterUtils.getCleanShardName(toLookup.getRootIdentifier()));
625
626             localShardFuture.onComplete(new OnComplete<ActorRef>() {
627                 @Override
628                 public void onComplete(final Throwable throwable, final ActorRef actorRef) {
629                     if (throwable != null) {
630                         //TODO Shouldn't we check why findLocalShard failed?
631                         LOG.debug("Backend shard[{}] removal lookup successful notifying the registration future",
632                                 toLookup);
633                         replyTo.tell(new Success(null), ActorRef.noSender());
634                     } else {
635                         tryReschedule(null);
636                     }
637                 }
638             }, system.dispatcher());
639         }
640
641         @Override
642         void reschedule(final int retries) {
643             LOG.debug("Backend shard[{}] removal lookup failed, shard is still present, try: {}, rescheduling..",
644                     toLookup, retries);
645             system.scheduler().scheduleOnce(
646                     SHARD_LOOKUP_TASK_INTERVAL, ShardRemovalLookupTask.this, system.dispatcher());
647         }
648     }
649
650     /**
651      * Task for handling the lookup of the backend for the configuration shard.
652      */
653     private static class ConfigShardLookupTask extends LookupTask {
654
655         private final ActorSystem system;
656         private final ActorRef replyTo;
657         private final ActorUtils context;
658
659         ConfigShardLookupTask(final ActorSystem system,
660                               final ActorRef replyTo,
661                               final ActorUtils context,
662                               final StartConfigShardLookup message,
663                               final int lookupMaxRetries) {
664             super(replyTo, lookupMaxRetries);
665             this.system = system;
666             this.replyTo = replyTo;
667             this.context = context;
668         }
669
670         @Override
671         void reschedule(final int retries) {
672             LOG.debug("Local backend for prefix configuration shard not found, try: {}, rescheduling..", retries);
673             system.scheduler().scheduleOnce(
674                     SHARD_LOOKUP_TASK_INTERVAL, ConfigShardLookupTask.this, system.dispatcher());
675         }
676
677         @Override
678         public void run() {
679             final Optional<ActorRef> localShard =
680                     context.findLocalShard(ClusterUtils.PREFIX_CONFIG_SHARD_ID);
681
682             if (!localShard.isPresent()) {
683                 tryReschedule(null);
684             } else {
685                 LOG.debug("Local backend for prefix configuration shard lookup successful");
686                 replyTo.tell(new Status.Success(null), ActorRef.noSender());
687             }
688         }
689     }
690
691     /**
692      * Task for handling the readiness state of the config shard. Reports success once the leader is elected.
693      */
694     private static class ConfigShardReadinessTask extends LookupTask {
695
696         private final ActorSystem system;
697         private final ActorRef replyTo;
698         private final ActorUtils context;
699         private final ClusterWrapper clusterWrapper;
700         private final ActorRef shard;
701
702         ConfigShardReadinessTask(final ActorSystem system,
703                                  final ActorRef replyTo,
704                                  final ActorUtils context,
705                                  final ClusterWrapper clusterWrapper,
706                                  final ActorRef shard,
707                                  final int lookupMaxRetries) {
708             super(replyTo, lookupMaxRetries);
709             this.system = system;
710             this.replyTo = replyTo;
711             this.context = context;
712             this.clusterWrapper = clusterWrapper;
713             this.shard = shard;
714         }
715
716         @Override
717         void reschedule(final int retries) {
718             LOG.debug("{} - Leader for config shard not found on try: {}, retrying..",
719                     clusterWrapper.getCurrentMemberName(), retries);
720             system.scheduler().scheduleOnce(
721                     SHARD_LOOKUP_TASK_INTERVAL, ConfigShardReadinessTask.this, system.dispatcher());
722         }
723
724         @Override
725         public void run() {
726             final Future<Object> ask = Patterns.ask(shard, FindLeader.INSTANCE, context.getOperationTimeout());
727
728             ask.onComplete(new OnComplete<>() {
729                 @Override
730                 public void onComplete(final Throwable throwable, final Object findLeaderReply) {
731                     if (throwable != null) {
732                         tryReschedule(throwable);
733                     } else {
734                         final FindLeaderReply findLeader = (FindLeaderReply) findLeaderReply;
735                         final Optional<String> leaderActor = findLeader.getLeaderActor();
736                         if (leaderActor.isPresent()) {
737                             // leader is found, backend seems ready, check if the frontend is ready
738                             LOG.debug("{} - Leader for config shard is ready. Ending lookup.",
739                                     clusterWrapper.getCurrentMemberName());
740                             replyTo.tell(new Status.Success(null), ActorRef.noSender());
741                         } else {
742                             tryReschedule(null);
743                         }
744                     }
745                 }
746             }, system.dispatcher());
747         }
748     }
749
750     public static class ShardedDataTreeActorCreator {
751
752         private DistributedShardedDOMDataTree shardingService;
753         private DistributedDataStoreInterface distributedConfigDatastore;
754         private DistributedDataStoreInterface distributedOperDatastore;
755         private ActorSystem actorSystem;
756         private ClusterWrapper cluster;
757         private int maxRetries;
758
759         public DistributedShardedDOMDataTree getShardingService() {
760             return shardingService;
761         }
762
763         public ShardedDataTreeActorCreator setShardingService(final DistributedShardedDOMDataTree shardingService) {
764             this.shardingService = shardingService;
765             return this;
766         }
767
768         public ActorSystem getActorSystem() {
769             return actorSystem;
770         }
771
772         public ShardedDataTreeActorCreator setActorSystem(final ActorSystem actorSystem) {
773             this.actorSystem = actorSystem;
774             return this;
775         }
776
777         public ShardedDataTreeActorCreator setClusterWrapper(final ClusterWrapper clusterWrapper) {
778             this.cluster = clusterWrapper;
779             return this;
780         }
781
782         public ClusterWrapper getClusterWrapper() {
783             return cluster;
784         }
785
786         public DistributedDataStoreInterface getDistributedConfigDatastore() {
787             return distributedConfigDatastore;
788         }
789
790         public ShardedDataTreeActorCreator setDistributedConfigDatastore(
791                 final DistributedDataStoreInterface distributedConfigDatastore) {
792             this.distributedConfigDatastore = distributedConfigDatastore;
793             return this;
794         }
795
796         public DistributedDataStoreInterface getDistributedOperDatastore() {
797             return distributedOperDatastore;
798         }
799
800         public ShardedDataTreeActorCreator setDistributedOperDatastore(
801                 final DistributedDataStoreInterface distributedOperDatastore) {
802             this.distributedOperDatastore = distributedOperDatastore;
803             return this;
804         }
805
806         public ShardedDataTreeActorCreator setLookupTaskMaxRetries(final int newMaxRetries) {
807             this.maxRetries = newMaxRetries;
808             return this;
809         }
810
811         public int getLookupTaskMaxRetries() {
812             return maxRetries;
813         }
814
815         private void verify() {
816             requireNonNull(shardingService);
817             requireNonNull(actorSystem);
818             requireNonNull(cluster);
819             requireNonNull(distributedConfigDatastore);
820             requireNonNull(distributedOperDatastore);
821         }
822
823         public Props props() {
824             verify();
825             return Props.create(ShardedDataTreeActor.class, this);
826         }
827     }
828 }

©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.