package org.opendaylight.controller.cluster.datastore;
import akka.actor.ActorSystem;
-import akka.actor.Props;
-import akka.osgi.BundleDelegatingClassLoader;
-import com.google.common.base.Preconditions;
-import com.typesafe.config.ConfigFactory;
-import java.util.HashSet;
-import java.util.Set;
-import java.util.concurrent.TimeUnit;
-import org.opendaylight.controller.cluster.common.actor.AkkaConfigurationReader;
+import org.opendaylight.controller.cluster.ActorSystemProvider;
+import org.opendaylight.controller.cluster.databroker.ClientBackedDataStore;
import org.opendaylight.controller.cluster.datastore.config.Configuration;
import org.opendaylight.controller.cluster.datastore.config.ConfigurationImpl;
-import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategyFactory;
-import org.opendaylight.controller.sal.core.api.model.SchemaService;
-import org.osgi.framework.BundleContext;
+import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot;
+import org.opendaylight.mdsal.dom.api.DOMSchemaService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import scala.concurrent.duration.Duration;
-public class DistributedDataStoreFactory {
- private static final String ACTOR_SYSTEM_NAME = "opendaylight-cluster-data";
- private static final String CONFIGURATION_NAME = "odl-cluster-data";
- private static ActorSystem actorSystem = null;
- private static final Set<DistributedDataStore> createdInstances = new HashSet<>(2);
+public final class DistributedDataStoreFactory {
+
private static final Logger LOG = LoggerFactory.getLogger(DistributedDataStoreFactory.class);
+ private static final String DEFAULT_MODULE_SHARDS_PATH = "./configuration/initial/module-shards.conf";
+ private static final String DEFAULT_MODULES_PATH = "./configuration/initial/modules.conf";
- public static synchronized DistributedDataStore createInstance(SchemaService schemaService,
- DatastoreContext datastoreContext, BundleContext bundleContext) {
+ private DistributedDataStoreFactory() {
+ }
- LOG.info("Create data store instance of type : {}", datastoreContext.getDataStoreType());
+ public static AbstractDataStore createInstance(final DOMSchemaService schemaService,
+ final DatastoreContext initialDatastoreContext, final DatastoreSnapshotRestore datastoreSnapshotRestore,
+ final ActorSystemProvider actorSystemProvider, final DatastoreContextIntrospector introspector,
+ final DatastoreContextPropertiesUpdater updater) {
+ return createInstance(schemaService, initialDatastoreContext, datastoreSnapshotRestore, actorSystemProvider,
+ introspector, updater, null);
+ }
- DatastoreContextIntrospector introspector = new DatastoreContextIntrospector(datastoreContext);
- DatastoreContextConfigAdminOverlay overlay = new DatastoreContextConfigAdminOverlay(
- introspector, bundleContext);
+ // TODO: separate out settle wait so it is better controlled
+ public static AbstractDataStore createInstance(final DOMSchemaService schemaService,
+ final DatastoreContext initialDatastoreContext, final DatastoreSnapshotRestore datastoreSnapshotRestore,
+ final ActorSystemProvider actorSystemProvider, final DatastoreContextIntrospector introspector,
+ final DatastoreContextPropertiesUpdater updater, final Configuration orgConfig) {
- ActorSystem actorSystem = getActorSystem(bundleContext, datastoreContext.getConfigurationReader());
- Configuration config = new ConfigurationImpl("module-shards.conf", "modules.conf");
- final DistributedDataStore dataStore = new DistributedDataStore(actorSystem,
- new ClusterWrapperImpl(actorSystem), config, introspector.getContext());
+ final AbstractDataStore dataStore = createInstance(actorSystemProvider, initialDatastoreContext,
+ introspector, datastoreSnapshotRestore, orgConfig);
- overlay.setListener(dataStore);
+ updater.setListener(dataStore);
- ShardStrategyFactory.setConfiguration(config);
schemaService.registerSchemaContextListener(dataStore);
- dataStore.setCloseable(overlay);
+ dataStore.setCloseable(updater);
dataStore.waitTillReady();
- createdInstances.add(dataStore);
return dataStore;
}
- private static synchronized final ActorSystem getActorSystem(final BundleContext bundleContext,
- AkkaConfigurationReader configurationReader) {
- if (actorSystem == null) {
- // Create an OSGi bundle classloader for actor system
- BundleDelegatingClassLoader classLoader = new BundleDelegatingClassLoader(bundleContext.getBundle(),
- Thread.currentThread().getContextClassLoader());
+ public static AbstractDataStore createInstance(final ActorSystemProvider actorSystemProvider,
+ final DatastoreContext initialDatastoreContext, final DatastoreContextIntrospector introspector,
+ final DatastoreSnapshotRestore datastoreSnapshotRestore, final Configuration orgConfig) {
- actorSystem = ActorSystem.create(ACTOR_SYSTEM_NAME,
- ConfigFactory.load(configurationReader.read()).getConfig(CONFIGURATION_NAME), classLoader);
- actorSystem.actorOf(Props.create(TerminationMonitor.class), "termination-monitor");
- }
+ final String datastoreName = initialDatastoreContext.getDataStoreName();
+ LOG.info("Create data store instance of type : {}", datastoreName);
- return actorSystem;
- }
+ final ActorSystem actorSystem = actorSystemProvider.getActorSystem();
+ final DatastoreSnapshot restoreFromSnapshot = datastoreSnapshotRestore.getAndRemove(datastoreName).orElse(null);
- public static synchronized void destroyInstance(DistributedDataStore dataStore){
- Preconditions.checkNotNull(dataStore, "dataStore should not be null");
-
- LOG.info("Destroy data store instance of type : {}", dataStore.getActorContext().getDataStoreType());
-
- if(createdInstances.remove(dataStore)){
- if(createdInstances.size() == 0){
- if(actorSystem != null) {
- actorSystem.shutdown();
- try {
- actorSystem.awaitTermination(Duration.create(10, TimeUnit.SECONDS));
- } catch (Exception e) {
- LOG.warn("Error awaiting actor termination", e);
- }
- actorSystem = null;
- }
- }
+ final Configuration config;
+ if (orgConfig == null) {
+ config = new ConfigurationImpl(DEFAULT_MODULE_SHARDS_PATH, DEFAULT_MODULES_PATH);
+ } else {
+ config = orgConfig;
+ }
+ final ClusterWrapper clusterWrapper = new ClusterWrapperImpl(actorSystem);
+ final DatastoreContextFactory contextFactory = introspector.newContextFactory();
+
+ // This is the potentially-updated datastore context, distinct from the initial one
+ final DatastoreContext datastoreContext = contextFactory.getBaseDatastoreContext();
+
+ final AbstractDataStore dataStore;
+ if (datastoreContext.isUseTellBasedProtocol()) {
+ dataStore = new ClientBackedDataStore(actorSystem, clusterWrapper, config, contextFactory,
+ restoreFromSnapshot);
+ LOG.info("Data store {} is using tell-based protocol", datastoreName);
+ } else {
+ dataStore = new DistributedDataStore(actorSystem, clusterWrapper, config, contextFactory,
+ restoreFromSnapshot);
+ LOG.info("Data store {} is using ask-based protocol", datastoreName);
}
- }
+ return dataStore;
+ }
}
Code Review / controller.git / blobdiff