[WIP] Add ES Auto Index Cleaner

pkg/es/client/index_client.go

@@ -11,6 +11,9 @@
 	"strconv"
 	"strings"
 	"time"
+
+	"github.com/jaegertracing/jaeger/pkg/es/config"
+	"go.uber.org/zap"
 )
 
 // Index represents ES index.
@@ -41,6 +44,32 @@
 	MasterTimeoutSeconds int
 }
 
+// Create the indices only client using config.Configuration
+func CreateIndicesClient(c *config.Configuration, logger *zap.Logger) (*IndicesClient, error) {
+	if len(c.Servers) < 1 {
+		return nil, errors.New("no servers specified")
+	}
+
+	httpClient := &http.Client{
+		Timeout: c.QueryTimeout,
+	}
+
+	transport, err := config.GetHTTPRoundTripper(c, logger)
+	if err != nil {
+		return nil, err
+	}
+	httpClient.Transport = transport
+
+	return &IndicesClient{
+		Client: Client{
+			Client:    httpClient,
+			Endpoint:  c.Servers[0],
+			BasicAuth: BasicAuth(c.Authentication.BasicAuthentication.Username, c.Authentication.BasicAuthentication.Password),
+		},
+		MasterTimeoutSeconds: int(c.QueryTimeout) / int(time.Second),
+	}, nil
+}
+
 // GetJaegerIndices queries all Jaeger indices including the archive and rollover.
 // Jaeger daily indices are:
 // - jaeger-span-2019-01-01

pkg/es/config/config.go

@@ -136,6 +136,14 @@ type Configuration struct {
 	Tags                     TagsAsFields  `mapstructure:"tags_as_fields"`
 	// Enabled, if set to true, enables the namespace for storage pointed to by this configuration.
 	Enabled bool `mapstructure:"-"`
+
+	// IndexCleaner cleans indices older than MaxSpanAge
+	IndexCleaner IndexCleaner `mapstructure:"index_cleaner"`
+}
+
+type IndexCleaner struct {
+	// Enabled, if set to true, enables the cleaner feature
+	Enabled bool `mapstructure:"enabled"`
 }
 
 // TagsAsFields holds configuration for tag schema.

plugin/storage/es/cleaner/cleaner.go

@@ -0,0 +1,68 @@
+package cleaner
+
+import (
+	"time"
+
+	"github.com/jaegertracing/jaeger/pkg/es/client"
+	"github.com/jaegertracing/jaeger/pkg/es/filter"
+	"go.uber.org/zap"
+)
+
+type IndexCleaner struct {
+	// client is the client used for interacting with es
+	client *client.IndicesClient
+	logger *zap.Logger
+	// indexPrefix is the user-defined index prefix, required for searching indices
+	indexPrefix string
+	// timePeriod is the interval at which cleaner would run
+	// also, cleaning is done for index with creationTime < now - timePeriod
+	timePeriod time.Duration
+	ticker     *time.Ticker
+}
+
+func NewIndexCleaner(client *client.IndicesClient, logger *zap.Logger, indexPrefix string, timePeriod time.Duration) *IndexCleaner {
+	return &IndexCleaner{
+		client:      client,
+		logger:      logger,
+		indexPrefix: indexPrefix,
+		timePeriod:  timePeriod,
+		ticker:      time.NewTicker(timePeriod),
+	}
+}
+
+func (i *IndexCleaner) Start() {
+	go func() {
+		for _ = range i.ticker.C {
+			err := i.Clean()
+			if err != nil {
+				i.logger.Error("Index cleaning failed", zap.Error(err))
+			}
+		}
+	}()
+}
+
+func (i *IndexCleaner) Clean() error {
+	indices, err := i.client.GetJaegerIndices(i.indexPrefix)
+	if err != nil {
+		return err
+	}
+
+	year, month, day := time.Now().UTC().Date()
+	tomorrowMidnight := time.Date(year, month, day, 0, 0, 0, 0, time.UTC).AddDate(0, 0, 1)
+	deleteIndicesBefore := tomorrowMidnight.Add(-1 * i.timePeriod)
+	i.logger.Info("Indices before this date will be deleted", zap.String("date", deleteIndicesBefore.Format(time.RFC3339)))
+
+	indices = filter.ByDate(indices, deleteIndicesBefore)
+
+	if len(indices) == 0 {
+		i.logger.Info("No indices to delete")
+		return nil
+	}
+
+	i.logger.Info("Deleting indices", zap.Any("indices", indices))
+	return i.client.DeleteIndices(indices)
+}
+
+func (i *IndexCleaner) Stop() {
+	i.ticker.Stop()
+}

plugin/storage/es/cleaner/cleaner_test.go

@@ -0,0 +1 @@
+package cleaner

plugin/storage/es/factory.go

@@ -21,10 +21,12 @@
 	"go.uber.org/zap"
 
 	"github.com/jaegertracing/jaeger/pkg/es"
+	"github.com/jaegertracing/jaeger/pkg/es/client"
 	"github.com/jaegertracing/jaeger/pkg/es/config"
 	"github.com/jaegertracing/jaeger/pkg/fswatcher"
 	"github.com/jaegertracing/jaeger/pkg/metrics"
 	"github.com/jaegertracing/jaeger/plugin"
+	"github.com/jaegertracing/jaeger/plugin/storage/es/cleaner"
 	esDepStore "github.com/jaegertracing/jaeger/plugin/storage/es/dependencystore"
 	"github.com/jaegertracing/jaeger/plugin/storage/es/mappings"
 	esSampleStore "github.com/jaegertracing/jaeger/plugin/storage/es/samplingstore"
@@ -67,6 +69,8 @@
 	archiveClient atomic.Pointer[es.Client]
 
 	watchers []*fswatcher.FSWatcher
+
+	primaryIndexCleaner *cleaner.IndexCleaner
 }
 
 // NewFactory creates a new Factory.
@@ -161,6 +165,15 @@
 		f.watchers = append(f.watchers, primaryWatcher)
 	}
 
+	if f.primaryConfig.IndexCleaner.Enabled {
+		indexCleanerClient, err := client.CreateIndicesClient(f.primaryConfig, logger)
+		if err != nil {
+			return fmt.Errorf("failed to create index cleaner client: %w", err)
+		}
+		primaryIndexCleaner := cleaner.NewIndexCleaner(indexCleanerClient, logger, string(f.primaryConfig.Indices.IndexPrefix), f.primaryConfig.MaxSpanAge)
+		primaryIndexCleaner.Start()
+	}
+
 	if f.archiveConfig.Enabled {
 		archiveClient, err := f.newClientFn(f.archiveConfig, logger, metricsFactory)
 		if err != nil {
@@ -370,6 +383,8 @@
 		errs = append(errs, client.Close())
 	}
 
+	f.primaryIndexCleaner.Stop()
+
 	return errors.Join(errs...)
 }