2 * Copyright (c) 2016 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;
13 import akka.actor.ActorRef;
14 import akka.actor.ActorSystem;
15 import akka.actor.PoisonPill;
16 import akka.actor.Props;
17 import akka.util.Timeout;
18 import com.google.common.base.Preconditions;
19 import com.google.common.base.Throwables;
20 import com.google.common.collect.ForwardingObject;
21 import com.google.common.util.concurrent.Uninterruptibles;
22 import java.util.Collection;
23 import java.util.Collections;
24 import java.util.concurrent.CompletionException;
25 import java.util.concurrent.TimeUnit;
26 import javax.annotation.Nonnull;
27 import org.opendaylight.controller.cluster.access.concepts.MemberName;
28 import org.opendaylight.controller.cluster.databroker.actors.dds.DataStoreClient;
29 import org.opendaylight.controller.cluster.databroker.actors.dds.SimpleDataStoreClientActor;
30 import org.opendaylight.controller.cluster.datastore.DistributedDataStore;
31 import org.opendaylight.controller.cluster.datastore.Shard;
32 import org.opendaylight.controller.cluster.datastore.config.PrefixShardConfiguration;
33 import org.opendaylight.controller.cluster.datastore.messages.CreatePrefixedShard;
34 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
35 import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
36 import org.opendaylight.controller.cluster.sharding.ShardedDataTreeActor.ShardedDataTreeActorCreator;
37 import org.opendaylight.controller.cluster.sharding.messages.PrefixShardCreated;
38 import org.opendaylight.controller.cluster.sharding.messages.PrefixShardRemoved;
39 import org.opendaylight.controller.cluster.sharding.messages.ProducerCreated;
40 import org.opendaylight.controller.cluster.sharding.messages.ProducerRemoved;
41 import org.opendaylight.mdsal.dom.api.DOMDataTreeCursorAwareTransaction;
42 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
43 import org.opendaylight.mdsal.dom.api.DOMDataTreeListener;
44 import org.opendaylight.mdsal.dom.api.DOMDataTreeLoopException;
45 import org.opendaylight.mdsal.dom.api.DOMDataTreeProducer;
46 import org.opendaylight.mdsal.dom.api.DOMDataTreeProducerException;
47 import org.opendaylight.mdsal.dom.api.DOMDataTreeService;
48 import org.opendaylight.mdsal.dom.api.DOMDataTreeShard;
49 import org.opendaylight.mdsal.dom.api.DOMDataTreeShardingConflictException;
50 import org.opendaylight.mdsal.dom.api.DOMDataTreeShardingService;
51 import org.opendaylight.mdsal.dom.broker.ShardedDOMDataTree;
52 import org.opendaylight.yangtools.concepts.ListenerRegistration;
53 import org.slf4j.Logger;
54 import org.slf4j.LoggerFactory;
55 import scala.concurrent.Await;
56 import scala.concurrent.Future;
59 * A layer on top of DOMDataTreeService that distributes producer/shard registrations to remote nodes via
60 * {@link ShardedDataTreeActor}. Also provides QoL method for addition of prefix based clustered shard into the system.
62 public class DistributedShardedDOMDataTree implements DOMDataTreeService, DOMDataTreeShardingService,
63 DistributedShardFactory {
65 private static final Logger LOG = LoggerFactory.getLogger(DistributedShardedDOMDataTree.class);
67 private static final Timeout DEFAULT_ASK_TIMEOUT = new Timeout(15, TimeUnit.SECONDS);
68 private static final int MAX_ACTOR_CREATION_RETRIES = 100;
69 private static final int ACTOR_RETRY_DELAY = 100;
70 private static final TimeUnit ACTOR_RETRY_TIME_UNIT = TimeUnit.MILLISECONDS;
72 static final String ACTOR_ID = "ShardedDOMDataTreeFrontend";
74 private final ShardedDOMDataTree shardedDOMDataTree;
75 private final ActorSystem actorSystem;
76 private final DistributedDataStore distributedOperDatastore;
77 private final DistributedDataStore distributedConfigDatastore;
79 private final ActorRef shardedDataTreeActor;
80 private final MemberName memberName;
82 public DistributedShardedDOMDataTree(final ActorSystem actorSystem,
83 final DistributedDataStore distributedOperDatastore,
84 final DistributedDataStore distributedConfigDatastore) {
85 this.actorSystem = Preconditions.checkNotNull(actorSystem);
86 this.distributedOperDatastore = Preconditions.checkNotNull(distributedOperDatastore);
87 this.distributedConfigDatastore = Preconditions.checkNotNull(distributedConfigDatastore);
88 shardedDOMDataTree = new ShardedDOMDataTree();
90 shardedDataTreeActor = createShardedDataTreeActor(actorSystem,
91 new ShardedDataTreeActorCreator()
92 .setDataTreeService(shardedDOMDataTree)
93 .setShardingService(shardedDOMDataTree)
94 .setActorSystem(actorSystem)
95 .setClusterWrapper(distributedConfigDatastore.getActorContext().getClusterWrapper())
96 .setDistributedConfigDatastore(distributedConfigDatastore)
97 .setDistributedOperDatastore(distributedOperDatastore),
100 this.memberName = distributedConfigDatastore.getActorContext().getCurrentMemberName();
105 public <T extends DOMDataTreeListener> ListenerRegistration<T> registerListener(
106 final T listener, final Collection<DOMDataTreeIdentifier> subtrees,
107 final boolean allowRxMerges, final Collection<DOMDataTreeProducer> producers)
108 throws DOMDataTreeLoopException {
110 throw new UnsupportedOperationException("Not implemented");
115 public DOMDataTreeProducer createProducer(@Nonnull final Collection<DOMDataTreeIdentifier> subtrees) {
116 LOG.debug("Creating producer for {}", subtrees);
117 final DOMDataTreeProducer producer = shardedDOMDataTree.createProducer(subtrees);
119 final Object response = distributedConfigDatastore.getActorContext()
120 .executeOperation(shardedDataTreeActor, new ProducerCreated(subtrees));
121 if (response == null) {
122 LOG.debug("Received success from remote nodes, creating producer:{}", subtrees);
123 return new ProxyProducer(producer, subtrees, shardedDataTreeActor,
124 distributedConfigDatastore.getActorContext());
125 } else if (response instanceof Exception) {
126 closeProducer(producer);
127 throw Throwables.propagate((Exception) response);
129 closeProducer(producer);
130 throw new RuntimeException("Unexpected response to create producer received." + response);
135 @SuppressWarnings("checkstyle:IllegalCatch")
136 public DistributedShardRegistration createDistributedShard(
137 final DOMDataTreeIdentifier prefix, final Collection<MemberName> replicaMembers)
138 throws DOMDataTreeShardingConflictException, DOMDataTreeProducerException,
139 DOMDataTreeShardCreationFailedException {
141 final String shardName = ClusterUtils.getCleanShardName(prefix.getRootIdentifier());
142 final DistributedDataStore distributedDataStore =
143 prefix.getDatastoreType().equals(org.opendaylight.mdsal.common.api.LogicalDatastoreType.CONFIGURATION)
144 ? distributedConfigDatastore : distributedOperDatastore;
146 final PrefixShardConfiguration config = new PrefixShardConfiguration(prefix, "prefix", replicaMembers);
147 if (replicaMembers.contains(memberName)) {
148 // spawn the backend shard and have the shard Manager create all replicas
149 final ActorRef shardManager = distributedDataStore.getActorContext().getShardManager();
151 shardManager.tell(new CreatePrefixedShard(config, null, Shard.builder()), noSender());
154 LOG.debug("Creating distributed datastore client for shard {}", shardName);
155 final Props distributedDataStoreClientProps =
156 SimpleDataStoreClientActor
157 .props(memberName, "Shard-" + shardName, distributedDataStore.getActorContext(), shardName);
159 final ActorRef clientActor = actorSystem.actorOf(distributedDataStoreClientProps);
160 final DataStoreClient client;
162 client = SimpleDataStoreClientActor.getDistributedDataStoreClient(clientActor, 30, TimeUnit.SECONDS);
163 } catch (final Exception e) {
164 LOG.error("Failed to get actor for {}", distributedDataStoreClientProps, e);
165 clientActor.tell(PoisonPill.getInstance(), noSender());
166 throw new DOMDataTreeProducerException("Unable to create producer", e);
169 // register the frontend into the sharding service and let the actor distribute this onto the other nodes
170 final ListenerRegistration<ShardFrontend> shardFrontendRegistration;
171 try (DOMDataTreeProducer producer = createProducer(Collections.singletonList(prefix))) {
172 shardFrontendRegistration = shardedDOMDataTree
173 .registerDataTreeShard(prefix,
174 new ShardFrontend(client, prefix),
175 ((ProxyProducer) producer).delegate());
178 final Future<Object> future = distributedDataStore.getActorContext()
179 .executeOperationAsync(shardedDataTreeActor, new PrefixShardCreated(config), DEFAULT_ASK_TIMEOUT);
181 final Object result = Await.result(future, DEFAULT_ASK_TIMEOUT.duration());
182 if (result != null) {
183 throw new DOMDataTreeShardCreationFailedException("Received unexpected response to PrefixShardCreated"
187 return new DistributedShardRegistrationImpl(shardFrontendRegistration, prefix, shardedDataTreeActor);
188 } catch (final CompletionException e) {
189 shardedDataTreeActor.tell(new PrefixShardRemoved(prefix), noSender());
190 clientActor.tell(PoisonPill.getInstance(), noSender());
192 final Throwable cause = e.getCause();
193 if (cause instanceof DOMDataTreeShardingConflictException) {
194 throw (DOMDataTreeShardingConflictException) cause;
197 throw new DOMDataTreeShardCreationFailedException("Shard creation failed.", e.getCause());
198 } catch (final Exception e) {
199 shardedDataTreeActor.tell(new PrefixShardRemoved(prefix), noSender());
200 clientActor.tell(PoisonPill.getInstance(), noSender());
202 throw new DOMDataTreeShardCreationFailedException("Shard creation failed.", e);
208 public <T extends DOMDataTreeShard> ListenerRegistration<T> registerDataTreeShard(
209 @Nonnull final DOMDataTreeIdentifier prefix,
210 @Nonnull final T shard,
211 @Nonnull final DOMDataTreeProducer producer)
212 throws DOMDataTreeShardingConflictException {
214 LOG.debug("Registering shard[{}] at prefix: {}", shard, prefix);
216 return shardedDOMDataTree.registerDataTreeShard(prefix, shard, producer);
219 private static void closeProducer(final DOMDataTreeProducer producer) {
222 } catch (final DOMDataTreeProducerException e) {
223 LOG.error("Unable to close producer", e);
227 @SuppressWarnings("checkstyle:IllegalCatch")
228 private static ActorRef createShardedDataTreeActor(final ActorSystem actorSystem,
229 final ShardedDataTreeActorCreator creator,
230 final String shardDataTreeActorId) {
231 Exception lastException = null;
233 for (int i = 0; i < MAX_ACTOR_CREATION_RETRIES; i++) {
235 return actorSystem.actorOf(creator.props(), shardDataTreeActorId);
236 } catch (final Exception e) {
238 Uninterruptibles.sleepUninterruptibly(ACTOR_RETRY_DELAY, ACTOR_RETRY_TIME_UNIT);
239 LOG.debug("Could not create actor {} because of {} -"
240 + " waiting for sometime before retrying (retry count = {})",
241 shardDataTreeActorId, e.getMessage(), i);
245 throw new IllegalStateException("Failed to create actor for ShardedDOMDataTree", lastException);
248 private static class DistributedShardRegistrationImpl implements DistributedShardRegistration {
249 private final ListenerRegistration<ShardFrontend> registration;
250 private final DOMDataTreeIdentifier prefix;
251 private final ActorRef shardedDataTreeActor;
253 DistributedShardRegistrationImpl(final ListenerRegistration<ShardFrontend> registration,
254 final DOMDataTreeIdentifier prefix,
255 final ActorRef shardedDataTreeActor) {
256 this.registration = registration;
257 this.prefix = prefix;
258 this.shardedDataTreeActor = shardedDataTreeActor;
262 public void close() {
263 // TODO send the correct messages to ShardManager to destroy the shard
264 // maybe we could provide replica removal mechanisms also?
265 shardedDataTreeActor.tell(new PrefixShardRemoved(prefix), noSender());
266 registration.close();
270 private static final class ProxyProducer extends ForwardingObject implements DOMDataTreeProducer {
272 private final DOMDataTreeProducer delegate;
273 private final Collection<DOMDataTreeIdentifier> subtrees;
274 private final ActorRef shardDataTreeActor;
275 private final ActorContext actorContext;
277 ProxyProducer(final DOMDataTreeProducer delegate,
278 final Collection<DOMDataTreeIdentifier> subtrees,
279 final ActorRef shardDataTreeActor,
280 final ActorContext actorContext) {
281 this.delegate = Preconditions.checkNotNull(delegate);
282 this.subtrees = Preconditions.checkNotNull(subtrees);
283 this.shardDataTreeActor = Preconditions.checkNotNull(shardDataTreeActor);
284 this.actorContext = Preconditions.checkNotNull(actorContext);
289 public DOMDataTreeCursorAwareTransaction createTransaction(final boolean isolated) {
290 return delegate.createTransaction(isolated);
295 public DOMDataTreeProducer createProducer(@Nonnull final Collection<DOMDataTreeIdentifier> subtrees) {
296 // TODO we probably don't need to distribute this on the remote nodes since once we have this producer
297 // open we surely have the rights to all the subtrees.
298 return delegate.createProducer(subtrees);
302 public void close() throws DOMDataTreeProducerException {
305 final Object o = actorContext.executeOperation(shardDataTreeActor, new ProducerRemoved(subtrees));
306 if (o instanceof DOMDataTreeProducerException) {
307 throw ((DOMDataTreeProducerException) o);
308 } else if (o instanceof Throwable) {
309 throw new DOMDataTreeProducerException("Unable to close producer", (Throwable) o);
314 protected DOMDataTreeProducer delegate() {