Bug 2187: Implement add-shard-replica RPC

[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / datastore / DatastoreContext.java

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 7f8a4e779d0446c769f5b096b1cbb0aace8b1370..4f7d68ea52861ae923212d4c8aa13668ae00b473 100644 (file)
--- 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,12 +9,16 @@
 package org.opendaylight.controller.cluster.datastore;
 
 import akka.util.Timeout;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Sets;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import org.apache.commons.lang3.text.WordUtils;
-import org.opendaylight.controller.cluster.datastore.config.ConfigurationReader;
-import org.opendaylight.controller.cluster.datastore.config.FileConfigurationReader;
+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.dom.store.impl.InMemoryDOMDataStoreConfigProperties;
 import scala.concurrent.duration.Duration;
 import scala.concurrent.duration.FiniteDuration;
@@ -25,39 +29,52 @@ import scala.concurrent.duration.FiniteDuration;
  * @author Thomas Pantelis
  */
 public class DatastoreContext {
+    public static final String METRICS_DOMAIN = "org.opendaylight.controller.cluster.datastore";
 
     public static final Duration DEFAULT_SHARD_TRANSACTION_IDLE_TIMEOUT = Duration.create(10, TimeUnit.MINUTES);
-    public static final int DEFAULT_OPERATION_TIMEOUT_IN_SECONDS = 5;
+    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_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;
-    public static final int DEFAULT_SHARD_TX_COMMIT_QUEUE_CAPACITY = 20000;
+    public static final int DEFAULT_SHARD_TX_COMMIT_QUEUE_CAPACITY = 50000;
     public static final Timeout DEFAULT_SHARD_INITIALIZATION_TIMEOUT = new Timeout(5, TimeUnit.MINUTES);
     public static final Timeout DEFAULT_SHARD_LEADER_ELECTION_TIMEOUT = new Timeout(30, TimeUnit.SECONDS);
     public static final boolean DEFAULT_PERSISTENT = true;
-    public static final FileConfigurationReader DEFAULT_CONFIGURATION_READER = new FileConfigurationReader();
+    public static final FileAkkaConfigurationReader DEFAULT_CONFIGURATION_READER = new FileAkkaConfigurationReader();
     public static final int DEFAULT_SHARD_SNAPSHOT_DATA_THRESHOLD_PERCENTAGE = 12;
     public static final int DEFAULT_SHARD_ELECTION_TIMEOUT_FACTOR = 2;
     public static final int DEFAULT_TX_CREATION_INITIAL_RATE_LIMIT = 100;
     public static final String UNKNOWN_DATA_STORE_TYPE = "unknown";
-    public static final int DEFAULT_SHARD_BATCHED_MODIFICATION_COUNT= 100;
+    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<String> globalDatastoreTypes = Sets.newConcurrentHashSet();
 
     private InMemoryDOMDataStoreConfigProperties dataStoreProperties;
     private Duration shardTransactionIdleTimeout = DatastoreContext.DEFAULT_SHARD_TRANSACTION_IDLE_TIMEOUT;
-    private int operationTimeoutInSeconds = DEFAULT_OPERATION_TIMEOUT_IN_SECONDS;
+    private long operationTimeoutInMillis = DEFAULT_OPERATION_TIMEOUT_IN_MS;
     private String dataStoreMXBeanType;
     private int shardTransactionCommitTimeoutInSeconds = DEFAULT_SHARD_TX_COMMIT_TIMEOUT_IN_SECONDS;
     private int shardTransactionCommitQueueCapacity = DEFAULT_SHARD_TX_COMMIT_QUEUE_CAPACITY;
     private Timeout shardInitializationTimeout = DEFAULT_SHARD_INITIALIZATION_TIMEOUT;
     private Timeout shardLeaderElectionTimeout = DEFAULT_SHARD_LEADER_ELECTION_TIMEOUT;
     private boolean persistent = DEFAULT_PERSISTENT;
-    private ConfigurationReader configurationReader = DEFAULT_CONFIGURATION_READER;
+    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 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 shardManagerPersistenceId;
+
+    public static Set<String> getGlobalDatastoreTypes() {
+        return globalDatastoreTypes;
+    }
 
     private DatastoreContext() {
         setShardJournalRecoveryLogBatchSize(DEFAULT_JOURNAL_RECOVERY_BATCH_SIZE);
@@ -66,12 +83,13 @@ 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) {
         this.dataStoreProperties = other.dataStoreProperties;
         this.shardTransactionIdleTimeout = other.shardTransactionIdleTimeout;
-        this.operationTimeoutInSeconds = other.operationTimeoutInSeconds;
+        this.operationTimeoutInMillis = other.operationTimeoutInMillis;
         this.dataStoreMXBeanType = other.dataStoreMXBeanType;
         this.shardTransactionCommitTimeoutInSeconds = other.shardTransactionCommitTimeoutInSeconds;
         this.shardTransactionCommitQueueCapacity = other.shardTransactionCommitQueueCapacity;
@@ -82,6 +100,10 @@ public class DatastoreContext {
         this.transactionCreationInitialRateLimit = other.transactionCreationInitialRateLimit;
         this.dataStoreType = other.dataStoreType;
         this.shardBatchedModificationCount = other.shardBatchedModificationCount;
+        this.writeOnlyTransactionOptimizationsEnabled = other.writeOnlyTransactionOptimizationsEnabled;
+        this.shardCommitQueueExpiryTimeoutInMillis = other.shardCommitQueueExpiryTimeoutInMillis;
+        this.transactionDebugContextEnabled = other.transactionDebugContextEnabled;
+        this.shardManagerPersistenceId = shardManagerPersistenceId;
 
         setShardJournalRecoveryLogBatchSize(other.raftConfig.getJournalRecoveryLogBatchSize());
         setSnapshotBatchCount(other.raftConfig.getSnapshotBatchCount());
@@ -89,6 +111,9 @@ public class DatastoreContext {
         setIsolatedLeaderCheckInterval(other.raftConfig.getIsolatedCheckIntervalInMillis());
         setSnapshotDataThresholdPercentage(other.raftConfig.getSnapshotDataThresholdPercentage());
         setElectionTimeoutFactor(other.raftConfig.getElectionTimeoutFactor());
+        setCustomRaftPolicyImplementation(other.raftConfig.getCustomRaftPolicyImplementationClass());
+        setShardSnapshotChunkSize(other.raftConfig.getSnapshotChunkSize());
+        setPeerAddressResolver(other.raftConfig.getPeerAddressResolver());
     }
 
     public static Builder newBuilder() {
@@ -111,8 +136,8 @@ public class DatastoreContext {
         return dataStoreMXBeanType;
     }
 
-    public int getOperationTimeoutInSeconds() {
-        return operationTimeoutInSeconds;
+    public long getOperationTimeoutInMillis() {
+        return operationTimeoutInMillis;
     }
 
     public ConfigParams getShardRaftConfig() {
@@ -139,7 +164,7 @@ public class DatastoreContext {
         return persistent;
     }
 
-    public ConfigurationReader getConfigurationReader() {
+    public AkkaConfigurationReader getConfigurationReader() {
         return configurationReader;
     }
 
@@ -155,6 +180,14 @@ public class DatastoreContext {
         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));
@@ -174,6 +207,10 @@ public class DatastoreContext {
         raftConfig.setElectionTimeoutFactor(shardElectionTimeoutFactor);
     }
 
+    private void setCustomRaftPolicyImplementation(String customRaftPolicyImplementation) {
+        raftConfig.setCustomRaftPolicyImplementationClass(customRaftPolicyImplementation);
+    }
+
     private void setSnapshotDataThresholdPercentage(int shardSnapshotDataThresholdPercentage) {
         raftConfig.setSnapshotDataThresholdPercentage(shardSnapshotDataThresholdPercentage);
     }
@@ -182,10 +219,30 @@ public class DatastoreContext {
         raftConfig.setSnapshotBatchCount(shardSnapshotBatchCount);
     }
 
+    private void setShardSnapshotChunkSize(int shardSnapshotChunkSize) {
+        raftConfig.setSnapshotChunkSize(shardSnapshotChunkSize);
+    }
+
     public int getShardBatchedModificationCount() {
         return shardBatchedModificationCount;
     }
 
+    public boolean isWriteOnlyTransactionOptimizationsEnabled() {
+        return writeOnlyTransactionOptimizationsEnabled;
+    }
+
+    public long getShardCommitQueueExpiryTimeoutInMillis() {
+        return shardCommitQueueExpiryTimeoutInMillis;
+    }
+
+    public boolean isTransactionDebugContextEnabled() {
+        return transactionDebugContextEnabled;
+    }
+
+    public int getShardSnapshotChunkSize() {
+        return raftConfig.getSnapshotChunkSize();
+    }
+
     public static class Builder {
         private final DatastoreContext datastoreContext;
         private int maxShardDataChangeExecutorPoolSize =
@@ -233,7 +290,12 @@ public class DatastoreContext {
         }
 
         public Builder operationTimeoutInSeconds(int operationTimeoutInSeconds) {
-            datastoreContext.operationTimeoutInSeconds = operationTimeoutInSeconds;
+            datastoreContext.operationTimeoutInMillis = TimeUnit.SECONDS.toMillis(operationTimeoutInSeconds);
+            return this;
+        }
+
+        public Builder operationTimeoutInMillis(long operationTimeoutInMillis) {
+            datastoreContext.operationTimeoutInMillis = operationTimeoutInMillis;
             return this;
         }
 
@@ -290,7 +352,7 @@ public class DatastoreContext {
             return shardLeaderElectionTimeout(timeout, TimeUnit.SECONDS);
         }
 
-        public Builder configurationReader(ConfigurationReader configurationReader){
+        public Builder configurationReader(AkkaConfigurationReader configurationReader){
             datastoreContext.configurationReader = configurationReader;
             return this;
         }
@@ -326,6 +388,27 @@ public class DatastoreContext {
             return this;
         }
 
+        public Builder writeOnlyTransactionOptimizationsEnabled(boolean value) {
+            datastoreContext.writeOnlyTransactionOptimizationsEnabled = value;
+            return this;
+        }
+
+        public Builder shardCommitQueueExpiryTimeoutInMillis(long value) {
+            datastoreContext.shardCommitQueueExpiryTimeoutInMillis = value;
+            return this;
+        }
+
+        public Builder shardCommitQueueExpiryTimeoutInSeconds(long value) {
+            datastoreContext.shardCommitQueueExpiryTimeoutInMillis = TimeUnit.MILLISECONDS.convert(
+                    value, TimeUnit.SECONDS);
+            return this;
+        }
+
+        public Builder transactionDebugContextEnabled(boolean value) {
+            datastoreContext.transactionDebugContextEnabled = value;
+            return this;
+        }
+
         public Builder maxShardDataChangeExecutorPoolSize(int maxShardDataChangeExecutorPoolSize) {
             this.maxShardDataChangeExecutorPoolSize = maxShardDataChangeExecutorPoolSize;
             return this;
@@ -346,11 +429,40 @@ 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);
+            }
+
             return datastoreContext;
         }
+
+        public Builder customRaftPolicyImplementation(String customRaftPolicyImplementation) {
+            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;
+        }
     }
 }