WIP: Size based retention

docs/sources/configuration/_index.md

@@ -2103,6 +2103,8 @@ The `compactor` block configures the compactor component, which compacts index s
 # CLI flag: -boltdb.shipper.compactor.retention-delete-worker-count
 [retention_delete_worker_count: <int> | default = 150]
 
+[retention_disk_space_percentage: <int>]
+
 # The maximum amount of time to spend running retention and deletion on any
 # given table in the index.
 # CLI flag: -boltdb.shipper.compactor.retention-table-timeout
@@ -2225,6 +2227,11 @@ compactor_ring:
   # CLI flag: -boltdb.shipper.compactor.ring.instance-availability-zone
   [instance_availability_zone: <string> | default = ""]
 
+# Run the compactor one time to cleanup and compact index files only (no
+# retention applied)
+# CLI flag: -boltdb.shipper.compactor.run-once
+[_: <boolean> | default = false]
+
   # Enable using a IPv6 instance address.
   # CLI flag: -boltdb.shipper.compactor.ring.instance-enable-ipv6
   [instance_enable_ipv6: <boolean> | default = false]
@@ -3096,6 +3103,10 @@ storage:
     # CLI flag: -common.storage.filesystem.rules-directory
     [rules_directory: <string> | default = ""]
 
+    # Size based retention percentage
+    # CLI flag: -common.storage..size-based-retention-percentage
+    [size_based_retention_percentage: <int> | default = 80]
+
   hedging:
     # If set to a non-zero value a second request will be issued at the provided
     # duration. Default is 0 (disabled)

pkg/loki/common/common.go

@@ -91,11 +91,13 @@ func (s *Storage) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
 }
 
 type FilesystemConfig struct {
-	ChunksDirectory string `yaml:"chunks_directory"`
-	RulesDirectory  string `yaml:"rules_directory"`
+	ChunksDirectory              string `yaml:"chunks_directory"`
+	RulesDirectory               string `yaml:"rules_directory"`
+	SizeBasedRetentionPercentage int    `yaml:"size_based_retention_percentage"`
 }
 
 func (cfg *FilesystemConfig) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
 	f.StringVar(&cfg.ChunksDirectory, prefix+"filesystem.chunk-directory", "", "Directory to store chunks in.")
 	f.StringVar(&cfg.RulesDirectory, prefix+"filesystem.rules-directory", "", "Directory to store rules in.")
+	f.IntVar(&cfg.SizeBasedRetentionPercentage, prefix+".size-based-retention-percentage", 80, "Size based retention percentage")
 }

pkg/loki/config_wrapper.go

@@ -459,6 +459,8 @@ func applyStorageConfig(cfg, defaults *ConfigWrapper) error {
 			r.Ruler.StoreConfig.Type = "local"
 			r.Ruler.StoreConfig.Local = local.Config{Directory: r.Common.Storage.FSConfig.RulesDirectory}
 			r.StorageConfig.FSConfig.Directory = r.Common.Storage.FSConfig.ChunksDirectory
+			r.CompactorConfig.RetentionDiskSpacePercentage = r.Common.Storage.FSConfig.SizeBasedRetentionPercentage
+			r.CompactorConfig.SharedStoreType = config.StorageTypeFileSystem
 		}
 	}
 

pkg/loki/modules.go

@@ -1137,7 +1137,8 @@ func (t *Loki) initCompactor() (services.Service, error) {
 		level.Info(util_log.Logger).Log("msg", "-boltdb.shipper.compactor.shared-store not specified, initializing compactor to operator on the following object stores", "stores", strings.Join(stores, ", "))
 	}
 
-	t.compactor, err = compactor.NewCompactor(t.Cfg.CompactorConfig, objectClients, t.Cfg.SchemaConfig, t.Overrides, prometheus.DefaultRegisterer)
+	t.compactor, err = compactor.NewCompactor(t.Cfg.CompactorConfig, objectClients, t.Cfg.SchemaConfig, t.Overrides, prometheus.DefaultRegisterer, t.Cfg.StorageConfig.FSConfig)
+
 	if err != nil {
 		return nil, err
 	}