Allow shard settle timeout to be tuned
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / datastore / AbstractDataStore.java
1 /*
2  * Copyright (c) 2014 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.datastore;
9
10 import static java.util.Objects.requireNonNull;
11
12 import akka.actor.ActorRef;
13 import akka.actor.ActorSystem;
14 import akka.actor.PoisonPill;
15 import akka.actor.Props;
16 import com.google.common.annotations.Beta;
17 import com.google.common.annotations.VisibleForTesting;
18 import com.google.common.base.Throwables;
19 import com.google.common.util.concurrent.Uninterruptibles;
20 import java.util.concurrent.CountDownLatch;
21 import java.util.concurrent.TimeUnit;
22 import java.util.concurrent.TimeoutException;
23 import org.opendaylight.controller.cluster.access.concepts.ClientIdentifier;
24 import org.opendaylight.controller.cluster.common.actor.Dispatchers;
25 import org.opendaylight.controller.cluster.databroker.actors.dds.DataStoreClient;
26 import org.opendaylight.controller.cluster.databroker.actors.dds.DistributedDataStoreClientActor;
27 import org.opendaylight.controller.cluster.datastore.config.Configuration;
28 import org.opendaylight.controller.cluster.datastore.identifiers.ShardManagerIdentifier;
29 import org.opendaylight.controller.cluster.datastore.jmx.mbeans.DatastoreConfigurationMXBeanImpl;
30 import org.opendaylight.controller.cluster.datastore.jmx.mbeans.DatastoreInfoMXBeanImpl;
31 import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot;
32 import org.opendaylight.controller.cluster.datastore.shardmanager.AbstractShardManagerCreator;
33 import org.opendaylight.controller.cluster.datastore.shardmanager.ShardManagerCreator;
34 import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
35 import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
36 import org.opendaylight.controller.cluster.datastore.utils.PrimaryShardInfoFutureCache;
37 import org.opendaylight.mdsal.dom.api.ClusteredDOMDataTreeChangeListener;
38 import org.opendaylight.mdsal.dom.api.DOMDataTreeChangeListener;
39 import org.opendaylight.mdsal.dom.api.DOMDataTreeCommitCohort;
40 import org.opendaylight.mdsal.dom.api.DOMDataTreeCommitCohortRegistration;
41 import org.opendaylight.mdsal.dom.api.DOMDataTreeCommitCohortRegistry;
42 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
43 import org.opendaylight.mdsal.dom.spi.store.DOMStoreTreeChangePublisher;
44 import org.opendaylight.yangtools.concepts.ListenerRegistration;
45 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
46 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
47 import org.opendaylight.yangtools.yang.model.api.SchemaContextListener;
48 import org.slf4j.Logger;
49 import org.slf4j.LoggerFactory;
50 import scala.concurrent.duration.Duration;
51
52 /**
53  * Base implementation of a distributed DOMStore.
54  */
55 public abstract class AbstractDataStore implements DistributedDataStoreInterface, SchemaContextListener,
56         DatastoreContextPropertiesUpdater.Listener, DOMStoreTreeChangePublisher,
57         DOMDataTreeCommitCohortRegistry, AutoCloseable {
58
59     private static final Logger LOG = LoggerFactory.getLogger(AbstractDataStore.class);
60
61     private final CountDownLatch waitTillReadyCountDownLatch = new CountDownLatch(1);
62     private final ClientIdentifier identifier;
63     private final DataStoreClient client;
64     private final ActorUtils actorUtils;
65
66     private AutoCloseable closeable;
67     private DatastoreConfigurationMXBeanImpl datastoreConfigMXBean;
68     private DatastoreInfoMXBeanImpl datastoreInfoMXBean;
69
70     @SuppressWarnings("checkstyle:IllegalCatch")
71     protected AbstractDataStore(final ActorSystem actorSystem, final ClusterWrapper cluster,
72             final Configuration configuration, final DatastoreContextFactory datastoreContextFactory,
73             final DatastoreSnapshot restoreFromSnapshot) {
74         requireNonNull(actorSystem, "actorSystem should not be null");
75         requireNonNull(cluster, "cluster should not be null");
76         requireNonNull(configuration, "configuration should not be null");
77         requireNonNull(datastoreContextFactory, "datastoreContextFactory should not be null");
78
79         String shardManagerId = ShardManagerIdentifier.builder()
80                 .type(datastoreContextFactory.getBaseDatastoreContext().getDataStoreName()).build().toString();
81
82         LOG.info("Creating ShardManager : {}", shardManagerId);
83
84         String shardDispatcher =
85                 new Dispatchers(actorSystem.dispatchers()).getDispatcherPath(Dispatchers.DispatcherType.Shard);
86
87         PrimaryShardInfoFutureCache primaryShardInfoCache = new PrimaryShardInfoFutureCache();
88
89         AbstractShardManagerCreator<?> creator = getShardManagerCreator().cluster(cluster).configuration(configuration)
90                 .datastoreContextFactory(datastoreContextFactory)
91                 .waitTillReadyCountDownLatch(waitTillReadyCountDownLatch)
92                 .primaryShardInfoCache(primaryShardInfoCache)
93                 .restoreFromSnapshot(restoreFromSnapshot)
94                 .distributedDataStore(this);
95
96         actorUtils = new ActorUtils(actorSystem, createShardManager(actorSystem, creator, shardDispatcher,
97                 shardManagerId), cluster, configuration, datastoreContextFactory.getBaseDatastoreContext(),
98                 primaryShardInfoCache);
99
100         final Props clientProps = DistributedDataStoreClientActor.props(cluster.getCurrentMemberName(),
101             datastoreContextFactory.getBaseDatastoreContext().getDataStoreName(), actorUtils);
102         final ActorRef clientActor = actorSystem.actorOf(clientProps);
103         try {
104             client = DistributedDataStoreClientActor.getDistributedDataStoreClient(clientActor, 30, TimeUnit.SECONDS);
105         } catch (Exception e) {
106             LOG.error("Failed to get actor for {}", clientProps, e);
107             clientActor.tell(PoisonPill.getInstance(), ActorRef.noSender());
108             Throwables.throwIfUnchecked(e);
109             throw new RuntimeException(e);
110         }
111
112         identifier = client.getIdentifier();
113         LOG.debug("Distributed data store client {} started", identifier);
114
115         datastoreConfigMXBean = new DatastoreConfigurationMXBeanImpl(
116                 datastoreContextFactory.getBaseDatastoreContext().getDataStoreMXBeanType());
117         datastoreConfigMXBean.setContext(datastoreContextFactory.getBaseDatastoreContext());
118         datastoreConfigMXBean.registerMBean();
119
120         datastoreInfoMXBean = new DatastoreInfoMXBeanImpl(datastoreContextFactory.getBaseDatastoreContext()
121                 .getDataStoreMXBeanType(), actorUtils);
122         datastoreInfoMXBean.registerMBean();
123     }
124
125     @VisibleForTesting
126     protected AbstractDataStore(final ActorUtils actorUtils, final ClientIdentifier identifier) {
127         this.actorUtils = requireNonNull(actorUtils, "actorContext should not be null");
128         this.client = null;
129         this.identifier = requireNonNull(identifier);
130     }
131
132     @VisibleForTesting
133     protected AbstractDataStore(final ActorUtils actorUtils, final ClientIdentifier identifier,
134                                 final DataStoreClient clientActor) {
135         this.actorUtils = requireNonNull(actorUtils, "actorContext should not be null");
136         this.client = clientActor;
137         this.identifier = requireNonNull(identifier);
138     }
139
140     protected AbstractShardManagerCreator<?> getShardManagerCreator() {
141         return new ShardManagerCreator();
142     }
143
144     protected final DataStoreClient getClient() {
145         return client;
146     }
147
148     final ClientIdentifier getIdentifier() {
149         return identifier;
150     }
151
152     public void setCloseable(final AutoCloseable closeable) {
153         this.closeable = closeable;
154     }
155
156     @Override
157     public <L extends DOMDataTreeChangeListener> ListenerRegistration<L> registerTreeChangeListener(
158             final YangInstanceIdentifier treeId, final L listener) {
159         requireNonNull(treeId, "treeId should not be null");
160         requireNonNull(listener, "listener should not be null");
161
162         final String shardName = actorUtils.getShardStrategyFactory().getStrategy(treeId).findShard(treeId);
163         LOG.debug("Registering tree listener: {} for tree: {} shard: {}", listener, treeId, shardName);
164
165         final DataTreeChangeListenerProxy<L> listenerRegistrationProxy =
166                 new DataTreeChangeListenerProxy<>(actorUtils, listener, treeId);
167         listenerRegistrationProxy.init(shardName);
168
169         return listenerRegistrationProxy;
170     }
171
172
173     @Override
174     public <C extends DOMDataTreeCommitCohort> DOMDataTreeCommitCohortRegistration<C> registerCommitCohort(
175             final DOMDataTreeIdentifier subtree, final C cohort) {
176         YangInstanceIdentifier treeId = requireNonNull(subtree, "subtree should not be null").getRootIdentifier();
177         requireNonNull(cohort, "listener should not be null");
178
179
180         final String shardName = actorUtils.getShardStrategyFactory().getStrategy(treeId).findShard(treeId);
181         LOG.debug("Registering cohort: {} for tree: {} shard: {}", cohort, treeId, shardName);
182
183         DataTreeCohortRegistrationProxy<C> cohortProxy =
184                 new DataTreeCohortRegistrationProxy<>(actorUtils, subtree, cohort);
185         cohortProxy.init(shardName);
186         return cohortProxy;
187     }
188
189     @Override
190     public void onGlobalContextUpdated(final SchemaContext schemaContext) {
191         actorUtils.setSchemaContext(schemaContext);
192     }
193
194     @Override
195     public void onDatastoreContextUpdated(final DatastoreContextFactory contextFactory) {
196         LOG.info("DatastoreContext updated for data store {}", actorUtils.getDataStoreName());
197
198         actorUtils.setDatastoreContext(contextFactory);
199         datastoreConfigMXBean.setContext(contextFactory.getBaseDatastoreContext());
200     }
201
202     @Override
203     @SuppressWarnings("checkstyle:IllegalCatch")
204     public void close() {
205         LOG.info("Closing data store {}", identifier);
206
207         if (datastoreConfigMXBean != null) {
208             datastoreConfigMXBean.unregisterMBean();
209         }
210         if (datastoreInfoMXBean != null) {
211             datastoreInfoMXBean.unregisterMBean();
212         }
213
214         if (closeable != null) {
215             try {
216                 closeable.close();
217             } catch (Exception e) {
218                 LOG.debug("Error closing instance", e);
219             }
220         }
221
222         actorUtils.shutdown();
223
224         if (client != null) {
225             client.close();
226         }
227     }
228
229     @Override
230     public ActorUtils getActorUtils() {
231         return actorUtils;
232     }
233
234     // TODO: consider removing this in favor of awaitReadiness()
235     public void waitTillReady() {
236         LOG.info("Beginning to wait for data store to become ready : {}", identifier);
237
238         final Duration toWait = initialSettleTime();
239         try {
240             if (toWait.isFinite()) {
241                 if (!waitTillReadyCountDownLatch.await(toWait.toNanos(), TimeUnit.NANOSECONDS)) {
242                     LOG.error("Shard leaders failed to settle in {}, giving up", toWait);
243                     return;
244                 }
245             } else {
246                 waitTillReadyCountDownLatch.await();
247             }
248         } catch (InterruptedException e) {
249             LOG.error("Interrupted while waiting for shards to settle", e);
250             return;
251         }
252
253         LOG.debug("Data store {} is now ready", identifier);
254     }
255
256     @Beta
257     public void awaitReadiness(final long timeout, final TimeUnit unit) throws InterruptedException, TimeoutException {
258         if (!waitTillReadyCountDownLatch.await(timeout, unit)) {
259             throw new TimeoutException("Shard leaders failed to settle");
260         }
261     }
262
263     @SuppressWarnings("checkstyle:IllegalCatch")
264     private static ActorRef createShardManager(final ActorSystem actorSystem,
265             final AbstractShardManagerCreator<?> creator, final String shardDispatcher,
266             final String shardManagerId) {
267         Exception lastException = null;
268
269         for (int i = 0; i < 100; i++) {
270             try {
271                 return actorSystem.actorOf(creator.props().withDispatcher(shardDispatcher), shardManagerId);
272             } catch (Exception e) {
273                 lastException = e;
274                 Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
275                 LOG.debug("Could not create actor {} because of {} - waiting for sometime before retrying "
276                         + "(retry count = {})", shardManagerId, e.getMessage(), i);
277             }
278         }
279
280         throw new IllegalStateException("Failed to create Shard Manager", lastException);
281     }
282
283     @VisibleForTesting
284     public CountDownLatch getWaitTillReadyCountDownLatch() {
285         return waitTillReadyCountDownLatch;
286     }
287
288     @SuppressWarnings("unchecked")
289     public <L extends DOMDataTreeChangeListener> ListenerRegistration<L> registerProxyListener(
290             final YangInstanceIdentifier shardLookup,
291             final YangInstanceIdentifier insideShard,
292             final org.opendaylight.mdsal.dom.api.DOMDataTreeChangeListener delegate) {
293
294         requireNonNull(shardLookup, "shardLookup should not be null");
295         requireNonNull(insideShard, "insideShard should not be null");
296         requireNonNull(delegate, "delegate should not be null");
297
298         final String shardName = actorUtils.getShardStrategyFactory().getStrategy(shardLookup).findShard(shardLookup);
299         LOG.debug("Registering tree listener: {} for tree: {} shard: {}, path inside shard: {}",
300                 delegate,shardLookup, shardName, insideShard);
301
302         final DataTreeChangeListenerProxy<DOMDataTreeChangeListener> listenerRegistrationProxy =
303                 new DataTreeChangeListenerProxy<>(actorUtils,
304                         // wrap this in the ClusteredDOMDataTreeChangeLister interface
305                         // since we always want clustered registration
306                         (ClusteredDOMDataTreeChangeListener) delegate::onDataTreeChanged, insideShard);
307         listenerRegistrationProxy.init(shardName);
308
309         return (ListenerRegistration<L>) listenerRegistrationProxy;
310     }
311
312     @SuppressWarnings("unchecked")
313     public <L extends DOMDataTreeChangeListener> ListenerRegistration<L> registerShardConfigListener(
314             final YangInstanceIdentifier internalPath,
315             final DOMDataTreeChangeListener delegate) {
316         requireNonNull(delegate, "delegate should not be null");
317
318         LOG.debug("Registering a listener for the configuration shard: {}", internalPath);
319
320         final DataTreeChangeListenerProxy<DOMDataTreeChangeListener> proxy =
321                 new DataTreeChangeListenerProxy<>(actorUtils, delegate, internalPath);
322         proxy.init(ClusterUtils.PREFIX_CONFIG_SHARD_ID);
323
324         return (ListenerRegistration<L>) proxy;
325     }
326
327     private Duration initialSettleTime() {
328         final DatastoreContext context = actorUtils.getDatastoreContext();
329         final int multiplier = context.getInitialSettleTimeoutMultiplier();
330         return multiplier == 0 ? Duration.Inf() : context.getShardLeaderElectionTimeout().duration().$times(multiplier);
331     }
332 }

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