cf14ebe28d11b0c39e105ab6396f8aff4c37903a
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / datastore / DatastoreContext.java
1 /*
2  * Copyright (c) 2014 Brocade Communications 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
9 package org.opendaylight.controller.cluster.datastore;
10
11 import akka.util.Timeout;
12 import com.google.common.annotations.VisibleForTesting;
13 import com.google.common.base.Preconditions;
14 import com.google.common.collect.Sets;
15 import java.util.Set;
16 import java.util.concurrent.TimeUnit;
17 import org.apache.commons.lang3.text.WordUtils;
18 import org.opendaylight.controller.cluster.common.actor.AkkaConfigurationReader;
19 import org.opendaylight.controller.cluster.common.actor.FileAkkaConfigurationReader;
20 import org.opendaylight.controller.cluster.raft.ConfigParams;
21 import org.opendaylight.controller.cluster.raft.DefaultConfigParamsImpl;
22 import org.opendaylight.controller.cluster.raft.PeerAddressResolver;
23 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
24 import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStoreConfigProperties;
25 import scala.concurrent.duration.Duration;
26 import scala.concurrent.duration.FiniteDuration;
27
28 /**
29  * Contains contextual data for a data store.
30  *
31  * @author Thomas Pantelis
32  */
33 public class DatastoreContext {
34     public static final String METRICS_DOMAIN = "org.opendaylight.controller.cluster.datastore";
35
36     public static final Duration DEFAULT_SHARD_TRANSACTION_IDLE_TIMEOUT = Duration.create(10, TimeUnit.MINUTES);
37     public static final int DEFAULT_OPERATION_TIMEOUT_IN_MS = 5000;
38     public static final int DEFAULT_SHARD_TX_COMMIT_TIMEOUT_IN_SECONDS = 30;
39     public static final int DEFAULT_JOURNAL_RECOVERY_BATCH_SIZE = 1;
40     public static final int DEFAULT_SNAPSHOT_BATCH_COUNT = 20000;
41     public static final int DEFAULT_HEARTBEAT_INTERVAL_IN_MILLIS = 500;
42     public static final int DEFAULT_ISOLATED_LEADER_CHECK_INTERVAL_IN_MILLIS = DEFAULT_HEARTBEAT_INTERVAL_IN_MILLIS * 10;
43     public static final int DEFAULT_SHARD_TX_COMMIT_QUEUE_CAPACITY = 50000;
44     public static final Timeout DEFAULT_SHARD_INITIALIZATION_TIMEOUT = new Timeout(5, TimeUnit.MINUTES);
45     public static final Timeout DEFAULT_SHARD_LEADER_ELECTION_TIMEOUT = new Timeout(30, TimeUnit.SECONDS);
46     public static final boolean DEFAULT_PERSISTENT = true;
47     public static final FileAkkaConfigurationReader DEFAULT_CONFIGURATION_READER = new FileAkkaConfigurationReader();
48     public static final int DEFAULT_SHARD_SNAPSHOT_DATA_THRESHOLD_PERCENTAGE = 12;
49     public static final int DEFAULT_SHARD_ELECTION_TIMEOUT_FACTOR = 2;
50     public static final int DEFAULT_TX_CREATION_INITIAL_RATE_LIMIT = 100;
51     public static final String UNKNOWN_DATA_STORE_TYPE = "unknown";
52     public static final int DEFAULT_SHARD_BATCHED_MODIFICATION_COUNT = 1000;
53     public static final long DEFAULT_SHARD_COMMIT_QUEUE_EXPIRY_TIMEOUT_IN_MS = TimeUnit.MILLISECONDS.convert(2, TimeUnit.MINUTES);
54     public static final int DEFAULT_SHARD_SNAPSHOT_CHUNK_SIZE = 2048000;
55
56     private static final Set<String> globalDatastoreNames = Sets.newConcurrentHashSet();
57
58     private InMemoryDOMDataStoreConfigProperties dataStoreProperties;
59     private Duration shardTransactionIdleTimeout = DatastoreContext.DEFAULT_SHARD_TRANSACTION_IDLE_TIMEOUT;
60     private long operationTimeoutInMillis = DEFAULT_OPERATION_TIMEOUT_IN_MS;
61     private String dataStoreMXBeanType;
62     private int shardTransactionCommitTimeoutInSeconds = DEFAULT_SHARD_TX_COMMIT_TIMEOUT_IN_SECONDS;
63     private int shardTransactionCommitQueueCapacity = DEFAULT_SHARD_TX_COMMIT_QUEUE_CAPACITY;
64     private Timeout shardInitializationTimeout = DEFAULT_SHARD_INITIALIZATION_TIMEOUT;
65     private Timeout shardLeaderElectionTimeout = DEFAULT_SHARD_LEADER_ELECTION_TIMEOUT;
66     private boolean persistent = DEFAULT_PERSISTENT;
67     private AkkaConfigurationReader configurationReader = DEFAULT_CONFIGURATION_READER;
68     private long transactionCreationInitialRateLimit = DEFAULT_TX_CREATION_INITIAL_RATE_LIMIT;
69     private final DefaultConfigParamsImpl raftConfig = new DefaultConfigParamsImpl();
70     private String dataStoreName = UNKNOWN_DATA_STORE_TYPE;
71     private LogicalDatastoreType logicalStoreType = LogicalDatastoreType.OPERATIONAL;
72     private int shardBatchedModificationCount = DEFAULT_SHARD_BATCHED_MODIFICATION_COUNT;
73     private boolean writeOnlyTransactionOptimizationsEnabled = true;
74     private long shardCommitQueueExpiryTimeoutInMillis = DEFAULT_SHARD_COMMIT_QUEUE_EXPIRY_TIMEOUT_IN_MS;
75     private boolean transactionDebugContextEnabled = false;
76     private String shardManagerPersistenceId;
77
78     public static Set<String> getGlobalDatastoreNames() {
79         return globalDatastoreNames;
80     }
81
82     private DatastoreContext() {
83         setShardJournalRecoveryLogBatchSize(DEFAULT_JOURNAL_RECOVERY_BATCH_SIZE);
84         setSnapshotBatchCount(DEFAULT_SNAPSHOT_BATCH_COUNT);
85         setHeartbeatInterval(DEFAULT_HEARTBEAT_INTERVAL_IN_MILLIS);
86         setIsolatedLeaderCheckInterval(DEFAULT_ISOLATED_LEADER_CHECK_INTERVAL_IN_MILLIS);
87         setSnapshotDataThresholdPercentage(DEFAULT_SHARD_SNAPSHOT_DATA_THRESHOLD_PERCENTAGE);
88         setElectionTimeoutFactor(DEFAULT_SHARD_ELECTION_TIMEOUT_FACTOR);
89         setShardSnapshotChunkSize(DEFAULT_SHARD_SNAPSHOT_CHUNK_SIZE);
90     }
91
92     private DatastoreContext(DatastoreContext other) {
93         this.dataStoreProperties = other.dataStoreProperties;
94         this.shardTransactionIdleTimeout = other.shardTransactionIdleTimeout;
95         this.operationTimeoutInMillis = other.operationTimeoutInMillis;
96         this.dataStoreMXBeanType = other.dataStoreMXBeanType;
97         this.shardTransactionCommitTimeoutInSeconds = other.shardTransactionCommitTimeoutInSeconds;
98         this.shardTransactionCommitQueueCapacity = other.shardTransactionCommitQueueCapacity;
99         this.shardInitializationTimeout = other.shardInitializationTimeout;
100         this.shardLeaderElectionTimeout = other.shardLeaderElectionTimeout;
101         this.persistent = other.persistent;
102         this.configurationReader = other.configurationReader;
103         this.transactionCreationInitialRateLimit = other.transactionCreationInitialRateLimit;
104         this.dataStoreName = other.dataStoreName;
105         this.logicalStoreType = other.logicalStoreType;
106         this.shardBatchedModificationCount = other.shardBatchedModificationCount;
107         this.writeOnlyTransactionOptimizationsEnabled = other.writeOnlyTransactionOptimizationsEnabled;
108         this.shardCommitQueueExpiryTimeoutInMillis = other.shardCommitQueueExpiryTimeoutInMillis;
109         this.transactionDebugContextEnabled = other.transactionDebugContextEnabled;
110         this.shardManagerPersistenceId = other.shardManagerPersistenceId;
111
112         setShardJournalRecoveryLogBatchSize(other.raftConfig.getJournalRecoveryLogBatchSize());
113         setSnapshotBatchCount(other.raftConfig.getSnapshotBatchCount());
114         setHeartbeatInterval(other.raftConfig.getHeartBeatInterval().toMillis());
115         setIsolatedLeaderCheckInterval(other.raftConfig.getIsolatedCheckIntervalInMillis());
116         setSnapshotDataThresholdPercentage(other.raftConfig.getSnapshotDataThresholdPercentage());
117         setElectionTimeoutFactor(other.raftConfig.getElectionTimeoutFactor());
118         setCustomRaftPolicyImplementation(other.raftConfig.getCustomRaftPolicyImplementationClass());
119         setShardSnapshotChunkSize(other.raftConfig.getSnapshotChunkSize());
120         setPeerAddressResolver(other.raftConfig.getPeerAddressResolver());
121     }
122
123     public static Builder newBuilder() {
124         return new Builder(new DatastoreContext());
125     }
126
127     public static Builder newBuilderFrom(DatastoreContext context) {
128         return new Builder(new DatastoreContext(context));
129     }
130
131     public InMemoryDOMDataStoreConfigProperties getDataStoreProperties() {
132         return dataStoreProperties;
133     }
134
135     public Duration getShardTransactionIdleTimeout() {
136         return shardTransactionIdleTimeout;
137     }
138
139     public String getDataStoreMXBeanType() {
140         return dataStoreMXBeanType;
141     }
142
143     public long getOperationTimeoutInMillis() {
144         return operationTimeoutInMillis;
145     }
146
147     public ConfigParams getShardRaftConfig() {
148         return raftConfig;
149     }
150
151     public int getShardTransactionCommitTimeoutInSeconds() {
152         return shardTransactionCommitTimeoutInSeconds;
153     }
154
155     public int getShardTransactionCommitQueueCapacity() {
156         return shardTransactionCommitQueueCapacity;
157     }
158
159     public Timeout getShardInitializationTimeout() {
160         return shardInitializationTimeout;
161     }
162
163     public Timeout getShardLeaderElectionTimeout() {
164         return shardLeaderElectionTimeout;
165     }
166
167     public boolean isPersistent() {
168         return persistent;
169     }
170
171     public AkkaConfigurationReader getConfigurationReader() {
172         return configurationReader;
173     }
174
175     public long getShardElectionTimeoutFactor(){
176         return raftConfig.getElectionTimeoutFactor();
177     }
178
179     public String getDataStoreName(){
180         return dataStoreName;
181     }
182
183     public LogicalDatastoreType getLogicalStoreType() {
184         return logicalStoreType;
185     }
186
187     public long getTransactionCreationInitialRateLimit() {
188         return transactionCreationInitialRateLimit;
189     }
190
191     public String getShardManagerPersistenceId() {
192         return shardManagerPersistenceId;
193     }
194
195     private void setPeerAddressResolver(PeerAddressResolver resolver) {
196         raftConfig.setPeerAddressResolver(resolver);
197     }
198
199     private void setHeartbeatInterval(long shardHeartbeatIntervalInMillis){
200         raftConfig.setHeartBeatInterval(new FiniteDuration(shardHeartbeatIntervalInMillis,
201                 TimeUnit.MILLISECONDS));
202     }
203
204     private void setShardJournalRecoveryLogBatchSize(int shardJournalRecoveryLogBatchSize){
205         raftConfig.setJournalRecoveryLogBatchSize(shardJournalRecoveryLogBatchSize);
206     }
207
208
209     private void setIsolatedLeaderCheckInterval(long shardIsolatedLeaderCheckIntervalInMillis) {
210         raftConfig.setIsolatedLeaderCheckInterval(
211                 new FiniteDuration(shardIsolatedLeaderCheckIntervalInMillis, TimeUnit.MILLISECONDS));
212     }
213
214     private void setElectionTimeoutFactor(long shardElectionTimeoutFactor) {
215         raftConfig.setElectionTimeoutFactor(shardElectionTimeoutFactor);
216     }
217
218     private void setCustomRaftPolicyImplementation(String customRaftPolicyImplementation) {
219         raftConfig.setCustomRaftPolicyImplementationClass(customRaftPolicyImplementation);
220     }
221
222     private void setSnapshotDataThresholdPercentage(int shardSnapshotDataThresholdPercentage) {
223         Preconditions.checkArgument(shardSnapshotDataThresholdPercentage >= 0
224                 && shardSnapshotDataThresholdPercentage <= 100);
225         raftConfig.setSnapshotDataThresholdPercentage(shardSnapshotDataThresholdPercentage);
226     }
227
228     private void setSnapshotBatchCount(long shardSnapshotBatchCount) {
229         raftConfig.setSnapshotBatchCount(shardSnapshotBatchCount);
230     }
231
232     private void setShardSnapshotChunkSize(int shardSnapshotChunkSize) {
233         raftConfig.setSnapshotChunkSize(shardSnapshotChunkSize);
234     }
235
236     public int getShardBatchedModificationCount() {
237         return shardBatchedModificationCount;
238     }
239
240     public boolean isWriteOnlyTransactionOptimizationsEnabled() {
241         return writeOnlyTransactionOptimizationsEnabled;
242     }
243
244     public long getShardCommitQueueExpiryTimeoutInMillis() {
245         return shardCommitQueueExpiryTimeoutInMillis;
246     }
247
248     public boolean isTransactionDebugContextEnabled() {
249         return transactionDebugContextEnabled;
250     }
251
252     public int getShardSnapshotChunkSize() {
253         return raftConfig.getSnapshotChunkSize();
254     }
255
256     public static class Builder {
257         private final DatastoreContext datastoreContext;
258         private int maxShardDataChangeExecutorPoolSize =
259                 InMemoryDOMDataStoreConfigProperties.DEFAULT_MAX_DATA_CHANGE_EXECUTOR_POOL_SIZE;
260         private int maxShardDataChangeExecutorQueueSize =
261                 InMemoryDOMDataStoreConfigProperties.DEFAULT_MAX_DATA_CHANGE_EXECUTOR_QUEUE_SIZE;
262         private int maxShardDataChangeListenerQueueSize =
263                 InMemoryDOMDataStoreConfigProperties.DEFAULT_MAX_DATA_CHANGE_LISTENER_QUEUE_SIZE;
264         private int maxShardDataStoreExecutorQueueSize =
265                 InMemoryDOMDataStoreConfigProperties.DEFAULT_MAX_DATA_STORE_EXECUTOR_QUEUE_SIZE;
266
267         private Builder(DatastoreContext datastoreContext) {
268             this.datastoreContext = datastoreContext;
269
270             if(datastoreContext.getDataStoreProperties() != null) {
271                 maxShardDataChangeExecutorPoolSize =
272                         datastoreContext.getDataStoreProperties().getMaxDataChangeExecutorPoolSize();
273                 maxShardDataChangeExecutorQueueSize =
274                         datastoreContext.getDataStoreProperties().getMaxDataChangeExecutorQueueSize();
275                 maxShardDataChangeListenerQueueSize =
276                         datastoreContext.getDataStoreProperties().getMaxDataChangeListenerQueueSize();
277                 maxShardDataStoreExecutorQueueSize =
278                         datastoreContext.getDataStoreProperties().getMaxDataStoreExecutorQueueSize();
279             }
280         }
281
282         public Builder boundedMailboxCapacity(int boundedMailboxCapacity) {
283             // TODO - this is defined in the yang DataStoreProperties but not currently used.
284             return this;
285         }
286
287         public Builder enableMetricCapture(boolean enableMetricCapture) {
288             // TODO - this is defined in the yang DataStoreProperties but not currently used.
289             return this;
290         }
291
292
293         public Builder shardTransactionIdleTimeout(long timeout, TimeUnit unit) {
294             datastoreContext.shardTransactionIdleTimeout = Duration.create(timeout, unit);
295             return this;
296         }
297
298         public Builder shardTransactionIdleTimeoutInMinutes(long timeout) {
299             return shardTransactionIdleTimeout(timeout, TimeUnit.MINUTES);
300         }
301
302         public Builder operationTimeoutInSeconds(int operationTimeoutInSeconds) {
303             datastoreContext.operationTimeoutInMillis = TimeUnit.SECONDS.toMillis(operationTimeoutInSeconds);
304             return this;
305         }
306
307         public Builder operationTimeoutInMillis(long operationTimeoutInMillis) {
308             datastoreContext.operationTimeoutInMillis = operationTimeoutInMillis;
309             return this;
310         }
311
312         public Builder dataStoreMXBeanType(String dataStoreMXBeanType) {
313             datastoreContext.dataStoreMXBeanType = dataStoreMXBeanType;
314             return this;
315         }
316
317         public Builder shardTransactionCommitTimeoutInSeconds(int shardTransactionCommitTimeoutInSeconds) {
318             datastoreContext.shardTransactionCommitTimeoutInSeconds = shardTransactionCommitTimeoutInSeconds;
319             return this;
320         }
321
322         public Builder shardJournalRecoveryLogBatchSize(int shardJournalRecoveryLogBatchSize) {
323             datastoreContext.setShardJournalRecoveryLogBatchSize(shardJournalRecoveryLogBatchSize);
324             return this;
325         }
326
327         public Builder shardSnapshotBatchCount(int shardSnapshotBatchCount) {
328             datastoreContext.setSnapshotBatchCount(shardSnapshotBatchCount);
329             return this;
330         }
331
332         public Builder shardSnapshotDataThresholdPercentage(int shardSnapshotDataThresholdPercentage) {
333             datastoreContext.setSnapshotDataThresholdPercentage(shardSnapshotDataThresholdPercentage);
334             return this;
335         }
336
337         public Builder shardHeartbeatIntervalInMillis(int shardHeartbeatIntervalInMillis) {
338             datastoreContext.setHeartbeatInterval(shardHeartbeatIntervalInMillis);
339             return this;
340         }
341
342         public Builder shardTransactionCommitQueueCapacity(int shardTransactionCommitQueueCapacity) {
343             datastoreContext.shardTransactionCommitQueueCapacity = shardTransactionCommitQueueCapacity;
344             return this;
345         }
346
347         public Builder shardInitializationTimeout(long timeout, TimeUnit unit) {
348             datastoreContext.shardInitializationTimeout = new Timeout(timeout, unit);
349             return this;
350         }
351
352         public Builder shardInitializationTimeoutInSeconds(long timeout) {
353             return shardInitializationTimeout(timeout, TimeUnit.SECONDS);
354         }
355
356         public Builder shardLeaderElectionTimeout(long timeout, TimeUnit unit) {
357             datastoreContext.shardLeaderElectionTimeout = new Timeout(timeout, unit);
358             return this;
359         }
360
361         public Builder shardLeaderElectionTimeoutInSeconds(long timeout) {
362             return shardLeaderElectionTimeout(timeout, TimeUnit.SECONDS);
363         }
364
365         public Builder configurationReader(AkkaConfigurationReader configurationReader){
366             datastoreContext.configurationReader = configurationReader;
367             return this;
368         }
369
370         public Builder persistent(boolean persistent){
371             datastoreContext.persistent = persistent;
372             return this;
373         }
374
375         public Builder shardIsolatedLeaderCheckIntervalInMillis(int shardIsolatedLeaderCheckIntervalInMillis) {
376             datastoreContext.setIsolatedLeaderCheckInterval(shardIsolatedLeaderCheckIntervalInMillis);
377             return this;
378         }
379
380         public Builder shardElectionTimeoutFactor(long shardElectionTimeoutFactor){
381             datastoreContext.setElectionTimeoutFactor(shardElectionTimeoutFactor);
382             return this;
383         }
384
385         public Builder transactionCreationInitialRateLimit(long initialRateLimit){
386             datastoreContext.transactionCreationInitialRateLimit = initialRateLimit;
387             return this;
388         }
389
390         public Builder logicalStoreType(LogicalDatastoreType logicalStoreType){
391             datastoreContext.logicalStoreType = Preconditions.checkNotNull(logicalStoreType);
392
393             // Retain compatible naming
394             switch (logicalStoreType) {
395             case CONFIGURATION:
396                 dataStoreName("config");
397                 break;
398             case OPERATIONAL:
399                 dataStoreName("operational");
400                 break;
401             default:
402                 dataStoreName(logicalStoreType.name());
403             }
404
405             return this;
406         }
407
408         public Builder dataStoreName(String dataStoreName){
409             datastoreContext.dataStoreName = Preconditions.checkNotNull(dataStoreName);
410             datastoreContext.dataStoreMXBeanType = "Distributed" + WordUtils.capitalize(dataStoreName) + "Datastore";
411             return this;
412         }
413
414         public Builder shardBatchedModificationCount(int shardBatchedModificationCount) {
415             datastoreContext.shardBatchedModificationCount = shardBatchedModificationCount;
416             return this;
417         }
418
419         public Builder writeOnlyTransactionOptimizationsEnabled(boolean value) {
420             datastoreContext.writeOnlyTransactionOptimizationsEnabled = value;
421             return this;
422         }
423
424         public Builder shardCommitQueueExpiryTimeoutInMillis(long value) {
425             datastoreContext.shardCommitQueueExpiryTimeoutInMillis = value;
426             return this;
427         }
428
429         public Builder shardCommitQueueExpiryTimeoutInSeconds(long value) {
430             datastoreContext.shardCommitQueueExpiryTimeoutInMillis = TimeUnit.MILLISECONDS.convert(
431                     value, TimeUnit.SECONDS);
432             return this;
433         }
434
435         public Builder transactionDebugContextEnabled(boolean value) {
436             datastoreContext.transactionDebugContextEnabled = value;
437             return this;
438         }
439
440         public Builder maxShardDataChangeExecutorPoolSize(int maxShardDataChangeExecutorPoolSize) {
441             this.maxShardDataChangeExecutorPoolSize = maxShardDataChangeExecutorPoolSize;
442             return this;
443         }
444
445         public Builder maxShardDataChangeExecutorQueueSize(int maxShardDataChangeExecutorQueueSize) {
446             this.maxShardDataChangeExecutorQueueSize = maxShardDataChangeExecutorQueueSize;
447             return this;
448         }
449
450         public Builder maxShardDataChangeListenerQueueSize(int maxShardDataChangeListenerQueueSize) {
451             this.maxShardDataChangeListenerQueueSize = maxShardDataChangeListenerQueueSize;
452             return this;
453         }
454
455         public Builder maxShardDataStoreExecutorQueueSize(int maxShardDataStoreExecutorQueueSize) {
456             this.maxShardDataStoreExecutorQueueSize = maxShardDataStoreExecutorQueueSize;
457             return this;
458         }
459
460         /**
461          * For unit tests only.
462          */
463         @VisibleForTesting
464         public Builder shardManagerPersistenceId(String id) {
465             datastoreContext.shardManagerPersistenceId = id;
466             return this;
467         }
468
469         public DatastoreContext build() {
470             datastoreContext.dataStoreProperties = InMemoryDOMDataStoreConfigProperties.create(
471                     maxShardDataChangeExecutorPoolSize, maxShardDataChangeExecutorQueueSize,
472                     maxShardDataChangeListenerQueueSize, maxShardDataStoreExecutorQueueSize);
473
474             if(datastoreContext.dataStoreName != null) {
475                 globalDatastoreNames.add(datastoreContext.dataStoreName);
476             }
477
478             return datastoreContext;
479         }
480
481         public Builder customRaftPolicyImplementation(String customRaftPolicyImplementation) {
482             datastoreContext.setCustomRaftPolicyImplementation(customRaftPolicyImplementation);
483             return this;
484         }
485
486         public Builder shardSnapshotChunkSize(int shardSnapshotChunkSize) {
487             datastoreContext.setShardSnapshotChunkSize(shardSnapshotChunkSize);
488             return this;
489         }
490
491         public Builder shardPeerAddressResolver(PeerAddressResolver resolver) {
492             datastoreContext.setPeerAddressResolver(resolver);
493             return this;
494         }
495     }
496 }

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