add MAX_COMPACTION_LEVELS option

DOCUMENTATION.md

@@ -1385,6 +1385,10 @@ You can tune the LOG_LEVEL of the Datahub. The supported values are DEBUG and IN
 
 The Datahub supports reporting metrics trough a StatsD server. This is turned off if left empty, and you can turn it on by giving it an ip-address and a port combination.
 
+```MAX_COMPACTION_LEVELS```
+
+Can be used to override Badger's default 7 LSM levels. When more that 1.1TB disk space usage are exceeded or expected to be exceeded, 8 compaction levels are needed.
+
 #### Securing the Data Hub
 
 There are four main security models for the data hub.

internal/conf/config.go

@@ -80,6 +80,7 @@ func loadEnv(basePath *string, loadFromHome bool) (*Env, error) {
 		FullsyncLeaseTimeout:    viper.GetDuration("FULLSYNC_LEASE_TIMEOUT"),
 		BlockCacheSize:          viper.GetInt64("BLOCK_CACHE_SIZE"),
 		ValueLogFileSize:        viper.GetInt64("VALUE_LOG_FILE_SIZE"),
+		MaxCompactionLevels:     viper.GetInt("MAX_COMPACTION_LEVELS"),
 		AdminUserName:           viper.GetString("ADMIN_USERNAME"),
 		AdminPassword:           viper.GetString("ADMIN_PASSWORD"),
 		NodeId:                  viper.GetString("NODE_ID"),

internal/conf/environment.go

@@ -36,6 +36,7 @@ type Env struct {
 	FullsyncLeaseTimeout    time.Duration
 	BlockCacheSize          int64
 	ValueLogFileSize        int64
+	MaxCompactionLevels     int
 	AdminUserName           string
 	AdminPassword           string
 	NodeId                  string

internal/server/store.go

@@ -59,6 +59,7 @@ type Store struct {
 	fullsyncLeaseTimeout time.Duration
 	blockCacheSize       int64
 	valueLogFileSize     int64
+	maxCompactionLevels  int
 }
 
 type BadgerLogger struct { // we use this to implement the Badger Logger interface
@@ -86,6 +87,7 @@ func NewStore(lc fx.Lifecycle, env *conf.Env, statsdClient statsd.ClientInterfac
 		fullsyncLeaseTimeout: fsTimeout,
 		blockCacheSize:       env.BlockCacheSize,
 		valueLogFileSize:     env.ValueLogFileSize,
+		maxCompactionLevels:  env.MaxCompactionLevels,
 	}
 	store.NamespaceManager = NewNamespaceManager(store)
 
@@ -342,7 +344,13 @@ func (s *Store) GetGlobalContext() *Context {
 func (s *Store) Open() error {
 	s.logger.Info("Open database")
 	opts := badger.DefaultOptions(s.storeLocation)
-	opts.MaxLevels = 8 // make badger accept data larger than 1.1TB
+
+	if s.maxCompactionLevels > 0 {
+		// default is 7, set to 8 to make badger accept data larger than 1.1TB at the cost of larger compactions
+		opts.MaxLevels = s.maxCompactionLevels
+	}
+	s.logger.Infof("Max Compaction Levels: %v", opts.MaxLevels)
+
 	if s.blockCacheSize > 0 {
 		opts.BlockCacheSize = s.blockCacheSize
 	} else {