compactor for compacting boltdb files uploaded by shipper

pkg/loki/loki.go

@@ -7,6 +7,8 @@ import (
 	"fmt"
 	"net/http"
 
+	"github.com/grafana/loki/pkg/storage/stores/shipper/compactor"
+
 	"github.com/cortexproject/cortex/pkg/util/flagext"
 	"github.com/cortexproject/cortex/pkg/util/modules"
 	"github.com/prometheus/client_golang/prometheus"
@@ -61,6 +63,7 @@ type Config struct {
 	RuntimeConfig    runtimeconfig.ManagerConfig `yaml:"runtime_config,omitempty"`
 	MemberlistKV     memberlist.KVConfig         `yaml:"memberlist"`
 	Tracing          tracing.Config              `yaml:"tracing"`
+	CompactorConfig  compactor.Config            `yaml:"compactor,omitempty"`
 }
 
 // RegisterFlags registers flag.
@@ -135,6 +138,7 @@ type Loki struct {
 	stopper       queryrange.Stopper
 	runtimeConfig *runtimeconfig.Manager
 	memberlistKV  *memberlist.KVInitService
+	compactor     *compactor.Compactor
 
 	httpAuthMiddleware middleware.Interface
 }
@@ -305,6 +309,7 @@ func (t *Loki) setupModuleManager() error {
 	mm.RegisterModule(Querier, t.initQuerier)
 	mm.RegisterModule(QueryFrontend, t.initQueryFrontend)
 	mm.RegisterModule(TableManager, t.initTableManager)
+	mm.RegisterModule(Compactor, t.initCompactor)
 	mm.RegisterModule(All, nil)
 
 	// Add dependencies
@@ -317,6 +322,7 @@ func (t *Loki) setupModuleManager() error {
 		Querier:       {Store, Ring, Server},
 		QueryFrontend: {Server, Overrides},
 		TableManager:  {Server},
+		Compactor:     {Server},
 		All:           {Querier, Ingester, Distributor, TableManager},
 	}
 

pkg/loki/modules.go

@@ -9,6 +9,8 @@ import (
 	"os"
 	"time"
 
+	"github.com/grafana/loki/pkg/storage/stores/shipper/compactor"
+
 	"github.com/cortexproject/cortex/pkg/chunk"
 	"github.com/cortexproject/cortex/pkg/chunk/cache"
 	"github.com/cortexproject/cortex/pkg/chunk/storage"
@@ -55,6 +57,7 @@ const (
 	Store         string = "store"
 	TableManager  string = "table-manager"
 	MemberlistKV  string = "memberlist-kv"
+	Compactor     string = "compactor"
 	All           string = "all"
 )
 
@@ -360,6 +363,16 @@ func (t *Loki) initMemberlistKV() (services.Service, error) {
 	return t.memberlistKV, nil
 }
 
+func (t *Loki) initCompactor() (services.Service, error) {
+	var err error
+	t.compactor, err = compactor.NewCompactor(t.cfg.CompactorConfig, t.cfg.StorageConfig.Config)
+	if err != nil {
+		return nil, err
+	}
+
+	return t.compactor, nil
+}
+
 func calculateMaxLookBack(pc chunk.PeriodConfig, maxLookBackConfig, maxChunkAge time.Duration) (time.Duration, error) {
 	if pc.ObjectType != shipper.FilesystemObjectStoreType && maxLookBackConfig.Nanoseconds() != 0 {
 		return 0, errors.New("it is an error to specify a non zero `query_store_max_look_back_period` value when using any object store other than `filesystem`")

pkg/storage/stores/shipper/compactor/compactor.go

@@ -0,0 +1,91 @@
+package compactor
+
+import (
+	"context"
+	"flag"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"github.com/cortexproject/cortex/pkg/chunk"
+	"github.com/cortexproject/cortex/pkg/chunk/storage"
+	chunk_util "github.com/cortexproject/cortex/pkg/chunk/util"
+	pkg_util "github.com/cortexproject/cortex/pkg/util"
+	"github.com/cortexproject/cortex/pkg/util/services"
+	"github.com/go-kit/kit/log/level"
+
+	"github.com/grafana/loki/pkg/storage/stores/shipper"
+	"github.com/grafana/loki/pkg/storage/stores/util"
+)
+
+type Config struct {
+	WorkingDirectory string `yaml:"working_directory"`
+	SharedStoreType  string `yaml:"shared_store"`
+}
+
+// RegisterFlags registers flags.
+func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
+	f.StringVar(&cfg.WorkingDirectory, "working-directory", "", "Directory where files can be downloaded for compaction.")
+	f.StringVar(&cfg.SharedStoreType, "shared-store", "", "Shared store used for storing boltdb files. Supported types: gcs, s3, azure, swift, filesystem")
+}
+
+type Compactor struct {
+	services.Service
+
+	cfg          Config
+	objectClient chunk.ObjectClient
+}
+
+func NewCompactor(cfg Config, storageConfig storage.Config) (*Compactor, error) {
+	objectClient, err := storage.NewObjectClient(cfg.SharedStoreType, storageConfig)
+	if err != nil {
+		return nil, err
+	}
+
+	err = chunk_util.EnsureDirectory(cfg.WorkingDirectory)
+	if err != nil {
+		return nil, err
+	}
+
+	compactor := Compactor{
+		cfg:          cfg,
+		objectClient: util.NewPrefixedObjectClient(objectClient, shipper.StorageKeyPrefix),
+	}
+
+	compactor.Service = services.NewTimerService(4*time.Hour, nil, compactor.Run, nil)
+	return &compactor, nil
+}
+
+func (c *Compactor) Run(ctx context.Context) error {
+	_, dirs, err := c.objectClient.List(ctx, "")
+	if err != nil {
+		return err
+	}
+
+	tables := make([]string, len(dirs))
+	for i, dir := range dirs {
+		tables[i] = strings.TrimSuffix(string(dir), "/")
+	}
+
+	for _, tableName := range tables {
+		table, err := newTable(ctx, filepath.Join(c.cfg.WorkingDirectory, tableName), c.objectClient)
+		if err != nil {
+			level.Error(pkg_util.Logger).Log("msg", "failed to initialize table for compaction", "err", err)
+			continue
+		}
+
+		err = table.compact()
+		if err != nil {
+			level.Error(pkg_util.Logger).Log("msg", "failed to compact files", "err", err)
+		}
+
+		// check if context was cancelled before going for next table.
+		select {
+		case <-ctx.Done():
+			return nil
+		default:
+		}
+	}
+
+	return nil
+}