X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FDatastoreContext.java;h=cf14ebe28d11b0c39e105ab6396f8aff4c37903a;hp=26d1a6eb97a82abdcee1bd8ddcf73ecc606ab2fc;hb=8e6e0aeb5ca474cb292aab6ca581e921b0a56489;hpb=e08568ddef3a5455b6b477e6672b8629f6935c20 diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/DatastoreContext.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/DatastoreContext.java index 26d1a6eb97..cf14ebe28d 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/DatastoreContext.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/DatastoreContext.java @@ -9,6 +9,8 @@ package org.opendaylight.controller.cluster.datastore; import akka.util.Timeout; +import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Preconditions; import com.google.common.collect.Sets; import java.util.Set; import java.util.concurrent.TimeUnit; @@ -17,6 +19,8 @@ import org.opendaylight.controller.cluster.common.actor.AkkaConfigurationReader; import org.opendaylight.controller.cluster.common.actor.FileAkkaConfigurationReader; import org.opendaylight.controller.cluster.raft.ConfigParams; import org.opendaylight.controller.cluster.raft.DefaultConfigParamsImpl; +import org.opendaylight.controller.cluster.raft.PeerAddressResolver; +import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType; import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStoreConfigProperties; import scala.concurrent.duration.Duration; import scala.concurrent.duration.FiniteDuration; @@ -32,7 +36,7 @@ public class DatastoreContext { public static final Duration DEFAULT_SHARD_TRANSACTION_IDLE_TIMEOUT = Duration.create(10, TimeUnit.MINUTES); public static final int DEFAULT_OPERATION_TIMEOUT_IN_MS = 5000; public static final int DEFAULT_SHARD_TX_COMMIT_TIMEOUT_IN_SECONDS = 30; - public static final int DEFAULT_JOURNAL_RECOVERY_BATCH_SIZE = 1000; + public static final int DEFAULT_JOURNAL_RECOVERY_BATCH_SIZE = 1; public static final int DEFAULT_SNAPSHOT_BATCH_COUNT = 20000; public static final int DEFAULT_HEARTBEAT_INTERVAL_IN_MILLIS = 500; public static final int DEFAULT_ISOLATED_LEADER_CHECK_INTERVAL_IN_MILLIS = DEFAULT_HEARTBEAT_INTERVAL_IN_MILLIS * 10; @@ -47,8 +51,9 @@ public class DatastoreContext { public static final String UNKNOWN_DATA_STORE_TYPE = "unknown"; public static final int DEFAULT_SHARD_BATCHED_MODIFICATION_COUNT = 1000; public static final long DEFAULT_SHARD_COMMIT_QUEUE_EXPIRY_TIMEOUT_IN_MS = TimeUnit.MILLISECONDS.convert(2, TimeUnit.MINUTES); + public static final int DEFAULT_SHARD_SNAPSHOT_CHUNK_SIZE = 2048000; - private static Set globalDatastoreTypes = Sets.newConcurrentHashSet(); + private static final Set globalDatastoreNames = Sets.newConcurrentHashSet(); private InMemoryDOMDataStoreConfigProperties dataStoreProperties; private Duration shardTransactionIdleTimeout = DatastoreContext.DEFAULT_SHARD_TRANSACTION_IDLE_TIMEOUT; @@ -62,15 +67,16 @@ public class DatastoreContext { private AkkaConfigurationReader configurationReader = DEFAULT_CONFIGURATION_READER; private long transactionCreationInitialRateLimit = DEFAULT_TX_CREATION_INITIAL_RATE_LIMIT; private final DefaultConfigParamsImpl raftConfig = new DefaultConfigParamsImpl(); - private String dataStoreType = UNKNOWN_DATA_STORE_TYPE; + private String dataStoreName = UNKNOWN_DATA_STORE_TYPE; + private LogicalDatastoreType logicalStoreType = LogicalDatastoreType.OPERATIONAL; private int shardBatchedModificationCount = DEFAULT_SHARD_BATCHED_MODIFICATION_COUNT; private boolean writeOnlyTransactionOptimizationsEnabled = true; private long shardCommitQueueExpiryTimeoutInMillis = DEFAULT_SHARD_COMMIT_QUEUE_EXPIRY_TIMEOUT_IN_MS; private boolean transactionDebugContextEnabled = false; - private String customRaftPolicyImplementation = ""; + private String shardManagerPersistenceId; - public static Set getGlobalDatastoreTypes() { - return globalDatastoreTypes; + public static Set getGlobalDatastoreNames() { + return globalDatastoreNames; } private DatastoreContext() { @@ -80,6 +86,7 @@ public class DatastoreContext { setIsolatedLeaderCheckInterval(DEFAULT_ISOLATED_LEADER_CHECK_INTERVAL_IN_MILLIS); setSnapshotDataThresholdPercentage(DEFAULT_SHARD_SNAPSHOT_DATA_THRESHOLD_PERCENTAGE); setElectionTimeoutFactor(DEFAULT_SHARD_ELECTION_TIMEOUT_FACTOR); + setShardSnapshotChunkSize(DEFAULT_SHARD_SNAPSHOT_CHUNK_SIZE); } private DatastoreContext(DatastoreContext other) { @@ -94,12 +101,13 @@ public class DatastoreContext { this.persistent = other.persistent; this.configurationReader = other.configurationReader; this.transactionCreationInitialRateLimit = other.transactionCreationInitialRateLimit; - this.dataStoreType = other.dataStoreType; + this.dataStoreName = other.dataStoreName; + this.logicalStoreType = other.logicalStoreType; this.shardBatchedModificationCount = other.shardBatchedModificationCount; this.writeOnlyTransactionOptimizationsEnabled = other.writeOnlyTransactionOptimizationsEnabled; this.shardCommitQueueExpiryTimeoutInMillis = other.shardCommitQueueExpiryTimeoutInMillis; this.transactionDebugContextEnabled = other.transactionDebugContextEnabled; - this.customRaftPolicyImplementation = other.customRaftPolicyImplementation; + this.shardManagerPersistenceId = other.shardManagerPersistenceId; setShardJournalRecoveryLogBatchSize(other.raftConfig.getJournalRecoveryLogBatchSize()); setSnapshotBatchCount(other.raftConfig.getSnapshotBatchCount()); @@ -107,8 +115,9 @@ public class DatastoreContext { setIsolatedLeaderCheckInterval(other.raftConfig.getIsolatedCheckIntervalInMillis()); setSnapshotDataThresholdPercentage(other.raftConfig.getSnapshotDataThresholdPercentage()); setElectionTimeoutFactor(other.raftConfig.getElectionTimeoutFactor()); - setCustomRaftPolicyImplementation(other.customRaftPolicyImplementation); - + setCustomRaftPolicyImplementation(other.raftConfig.getCustomRaftPolicyImplementationClass()); + setShardSnapshotChunkSize(other.raftConfig.getSnapshotChunkSize()); + setPeerAddressResolver(other.raftConfig.getPeerAddressResolver()); } public static Builder newBuilder() { @@ -167,14 +176,26 @@ public class DatastoreContext { return raftConfig.getElectionTimeoutFactor(); } - public String getDataStoreType(){ - return dataStoreType; + public String getDataStoreName(){ + return dataStoreName; + } + + public LogicalDatastoreType getLogicalStoreType() { + return logicalStoreType; } public long getTransactionCreationInitialRateLimit() { return transactionCreationInitialRateLimit; } + public String getShardManagerPersistenceId() { + return shardManagerPersistenceId; + } + + private void setPeerAddressResolver(PeerAddressResolver resolver) { + raftConfig.setPeerAddressResolver(resolver); + } + private void setHeartbeatInterval(long shardHeartbeatIntervalInMillis){ raftConfig.setHeartBeatInterval(new FiniteDuration(shardHeartbeatIntervalInMillis, TimeUnit.MILLISECONDS)); @@ -198,8 +219,9 @@ public class DatastoreContext { raftConfig.setCustomRaftPolicyImplementationClass(customRaftPolicyImplementation); } - private void setSnapshotDataThresholdPercentage(int shardSnapshotDataThresholdPercentage) { + Preconditions.checkArgument(shardSnapshotDataThresholdPercentage >= 0 + && shardSnapshotDataThresholdPercentage <= 100); raftConfig.setSnapshotDataThresholdPercentage(shardSnapshotDataThresholdPercentage); } @@ -207,6 +229,10 @@ public class DatastoreContext { raftConfig.setSnapshotBatchCount(shardSnapshotBatchCount); } + private void setShardSnapshotChunkSize(int shardSnapshotChunkSize) { + raftConfig.setSnapshotChunkSize(shardSnapshotChunkSize); + } + public int getShardBatchedModificationCount() { return shardBatchedModificationCount; } @@ -223,6 +249,10 @@ public class DatastoreContext { return transactionDebugContextEnabled; } + public int getShardSnapshotChunkSize() { + return raftConfig.getSnapshotChunkSize(); + } + public static class Builder { private final DatastoreContext datastoreContext; private int maxShardDataChangeExecutorPoolSize = @@ -357,9 +387,27 @@ public class DatastoreContext { return this; } - public Builder dataStoreType(String dataStoreType){ - datastoreContext.dataStoreType = dataStoreType; - datastoreContext.dataStoreMXBeanType = "Distributed" + WordUtils.capitalize(dataStoreType) + "Datastore"; + public Builder logicalStoreType(LogicalDatastoreType logicalStoreType){ + datastoreContext.logicalStoreType = Preconditions.checkNotNull(logicalStoreType); + + // Retain compatible naming + switch (logicalStoreType) { + case CONFIGURATION: + dataStoreName("config"); + break; + case OPERATIONAL: + dataStoreName("operational"); + break; + default: + dataStoreName(logicalStoreType.name()); + } + + return this; + } + + public Builder dataStoreName(String dataStoreName){ + datastoreContext.dataStoreName = Preconditions.checkNotNull(dataStoreName); + datastoreContext.dataStoreMXBeanType = "Distributed" + WordUtils.capitalize(dataStoreName) + "Datastore"; return this; } @@ -409,13 +457,22 @@ public class DatastoreContext { return this; } + /** + * For unit tests only. + */ + @VisibleForTesting + public Builder shardManagerPersistenceId(String id) { + datastoreContext.shardManagerPersistenceId = id; + return this; + } + public DatastoreContext build() { datastoreContext.dataStoreProperties = InMemoryDOMDataStoreConfigProperties.create( maxShardDataChangeExecutorPoolSize, maxShardDataChangeExecutorQueueSize, maxShardDataChangeListenerQueueSize, maxShardDataStoreExecutorQueueSize); - if(datastoreContext.dataStoreType != null) { - globalDatastoreTypes.add(datastoreContext.dataStoreType); + if(datastoreContext.dataStoreName != null) { + globalDatastoreNames.add(datastoreContext.dataStoreName); } return datastoreContext; @@ -425,5 +482,15 @@ public class DatastoreContext { datastoreContext.setCustomRaftPolicyImplementation(customRaftPolicyImplementation); return this; } + + public Builder shardSnapshotChunkSize(int shardSnapshotChunkSize) { + datastoreContext.setShardSnapshotChunkSize(shardSnapshotChunkSize); + return this; + } + + public Builder shardPeerAddressResolver(PeerAddressResolver resolver) { + datastoreContext.setPeerAddressResolver(resolver); + return this; + } } }