Add configuration parameter to expose rate limit for TSM compaction.

etc/config.sample.toml

@@ -97,6 +97,15 @@
   # to cache snapshotting.
   # max-concurrent-compactions = 0
 
+  # CompactThroughput is the rate limit in bytes per second that we
+  # will allow TSM compactions to write to disk. Note that short bursts are allowed
+  # to happen at a possibly larger value, set by CompactThroughputBurst
+  # compact-throughput = "48m"
+
+  # CompactThroughputBurst is the rate limit in bytes per second that we
+  # will allow TSM compactions to write to disk.
+  # compact-throughput-burst = "48m"
+
   # The threshold, in bytes, when an index write-ahead log file will compact
   # into an index file. Lower sizes will cause log files to be compacted more
   # quickly and result in lower heap usage at the expense of write throughput.

tsdb/config.go

@@ -37,6 +37,17 @@ const (
 	// will compact all TSM files in a shard if it hasn't received a write or delete
 	DefaultCompactFullWriteColdDuration = time.Duration(4 * time.Hour)
 
+	// DefaultCompactThroughput is the rate limit in bytes per second that we
+	// will allow TSM compactions to write to disk. Not that short bursts are allowed
+	// to happen at a possibly larger value, set by DefaultCompactThroughputBurst.
+	// A value of 0 here will disable compaction rate limiting
+	DefaultCompactThroughput = 48 * 1024 * 1024
+
+	// DefaultCompactThroughputBurst is the rate limit in bytes per second that we
+	// will allow TSM compactions to write to disk. If this is not set, the burst value
+	// will be set to equal the normal throughput
+	DefaultCompactThroughputBurst = 48 * 1024 * 1024
+
 	// DefaultMaxPointsPerBlock is the maximum number of points in an encoded
 	// block in a TSM file
 	DefaultMaxPointsPerBlock = 1000
@@ -79,6 +90,8 @@ type Config struct {
 	CacheSnapshotMemorySize        toml.Size     `toml:"cache-snapshot-memory-size"`
 	CacheSnapshotWriteColdDuration toml.Duration `toml:"cache-snapshot-write-cold-duration"`
 	CompactFullWriteColdDuration   toml.Duration `toml:"compact-full-write-cold-duration"`
+	CompactThroughput              toml.Size     `toml:"compact-throughput"`
+	CompactThroughputBurst         toml.Size     `toml:"compact-throughput-burst"`
 
 	// Limits
 
@@ -125,6 +138,8 @@ func NewConfig() Config {
 		CacheSnapshotMemorySize:        toml.Size(DefaultCacheSnapshotMemorySize),
 		CacheSnapshotWriteColdDuration: toml.Duration(DefaultCacheSnapshotWriteColdDuration),
 		CompactFullWriteColdDuration:   toml.Duration(DefaultCompactFullWriteColdDuration),
+		CompactThroughput:              toml.Size(DefaultCompactThroughput),
+		CompactThroughputBurst:         toml.Size(DefaultCompactThroughputBurst),
 
 		MaxSeriesPerDatabase:     DefaultMaxSeriesPerDatabase,
 		MaxValuesPerTag:          DefaultMaxValuesPerTag,

tsdb/store.go

@@ -230,15 +230,36 @@ func (s *Store) loadShards() error {
 
 	s.EngineOptions.CompactionLimiter = limiter.NewFixed(lim)
 
-	// Env var to disable throughput limiter.  This will be moved to a config option in 1.5.
 	compactionSettings := []zapcore.Field{zap.Int("max_concurrent_compactions", lim)}
-	if os.Getenv("INFLUXDB_DATA_COMPACTION_THROUGHPUT") == "" {
-		rate, burst := 48*1024*1024, 48*1024*1024
-		s.EngineOptions.CompactionThroughputLimiter = limiter.NewRate(48*1024*1024, 48*1024*1024)
-		compactionSettings = append(compactionSettings, zap.Int("throughput_bytes_per_second", rate), zap.Int("throughput_burst_bytes", burst))
+	throughput := int(s.EngineOptions.Config.CompactThroughput)
+	throughputBurst := int(s.EngineOptions.Config.CompactThroughputBurst)
+	compactionSettings = append(
+		compactionSettings,
+		zap.Int("throughput_bytes_per_second", throughput),
+		zap.Int("throughput_bytes_per_second_burst", throughputBurst),
+	)
+	if throughput > 0 {
+		if throughputBurst < throughput {
+			throughputBurst = throughput
+		}
+		s.Logger.Info(
+			"Compaction throughput enabled",
+			zap.String("throughput_bytes_per_second", fmt.Sprintf("%d%%", throughput)),
+			zap.String("throughput_bytes_per_second_burst", fmt.Sprintf("%d%%", throughputBurst)),
+		)
+		s.EngineOptions.CompactionThroughputLimiter = limiter.NewRate(
+			throughput,
+			throughputBurst,
+		)
 	} else {
-		compactionSettings = append(compactionSettings, zap.String("throughput_bytes_per_second", "unlimited"), zap.String("throughput_burst", "unlimited"))
+		compactionSettings = append(
+			compactionSettings,
+			zap.String("throughput_bytes_per_second", "unlimited"),
+			zap.String("throughput_bytes_per_second_burst", "unlimited"),
+		)
+		s.Logger.Info("Compaction throughput limit disabled")
 	}
+
 	s.Logger.Info("Compaction settings", compactionSettings...)
 
 	log, logEnd := logger.NewOperation(s.Logger, "Open store", "tsdb_open")