move clean_up from client to extension level

Each extension instance can create multiple clients.
Moving the logic to extension start is better aligned with the
extension lifecycle and prevent potential collisions, all clients will share the same compaction folder already cleaned up.

Author: pandres-varian

extension/storage/filestorage/client.go

@@ -8,7 +8,6 @@ import (
 	"errors"
 	"fmt"
 	"os"
-	"path/filepath"
 	"sync"
 	"syscall"
 	"time"
@@ -21,12 +20,13 @@ import (
 var defaultBucket = []byte(`default`)
 
 const (
+	TempDbPrefix = "tempdb"
+
 	elapsedKey       = "elapsed"
 	directoryKey     = "directory"
 	tempDirectoryKey = "tempDirectory"
 
-	tempDbPrefix = "tempdb"
-	oneMiB       = 1048576
+	oneMiB = 1048576
 )
 
 type fileStorageClient struct {
@@ -65,9 +65,6 @@ func newClient(logger *zap.Logger, filePath string, timeout time.Duration, compa
 	}
 
 	client := &fileStorageClient{logger: logger, db: db, compactionCfg: compactionCfg, openTimeout: timeout}
-	if compactionCfg.CleanupOnStart {
-		client.cleanup(compactionCfg.Directory)
-	}
 	if compactionCfg.OnRebound {
 		client.startCompactionLoop(context.Background())
 	}
@@ -157,7 +154,7 @@ func (c *fileStorageClient) Compact(compactionDirectory string, timeout time.Dur
 	var compactedDb *bbolt.DB
 
 	// create temporary file in compactionDirectory
-	file, err = os.CreateTemp(compactionDirectory, tempDbPrefix)
+	file, err = os.CreateTemp(compactionDirectory, TempDbPrefix)
 	if err != nil {
 		return err
 	}
@@ -347,30 +344,3 @@ func moveFileWithFallback(src string, dest string) error {
 	err = os.Remove(src)
 	return err
 }
-
-// cleanup left compaction temporary files from previous killed process
-func (c *fileStorageClient) cleanup(compactionDirectory string) error {
-	pattern := filepath.Join(compactionDirectory, fmt.Sprintf("%s*", tempDbPrefix))
-	contents, err := filepath.Glob(pattern)
-	if err != nil {
-		c.logger.Info("cleanup error listing temporary files",
-			zap.Error(err))
-		return err
-	}
-
-	var errs []error
-	for _, item := range contents {
-		err = os.Remove(item)
-		if err == nil {
-			c.logger.Debug("cleanup",
-				zap.String("deletedFile", item))
-		} else {
-			errs = append(errs, err)
-		}
-	}
-	if errs != nil {
-		c.logger.Info("cleanup errors",
-			zap.Error(errors.Join(errs...)))
-	}
-	return nil
-}

extension/storage/filestorage/client_test.go

@@ -404,34 +404,6 @@ func TestClientConcurrentCompaction(t *testing.T) {
 	}
 }
 
-func TestClientCleanupOnStart(t *testing.T) {
-	tempDir := t.TempDir()
-	dbFile := filepath.Join(tempDir, "my_db")
-	temp, _ := os.CreateTemp(tempDir, tempDbPrefix)
-	// simulate ongoing compaction in another instance
-	tempLocked, _ := os.CreateTemp(tempDir, tempDbPrefix)
-	temp.Close()
-
-	client, err := newClient(zap.NewNop(), dbFile, time.Second, &CompactionConfig{
-		Directory:      tempDir,
-		CleanupOnStart: true,
-	}, false)
-	require.NoError(t, err)
-
-	t.Cleanup(func() {
-		require.NoError(t, client.Close(context.TODO()))
-		tempLocked.Close()
-	})
-
-	// check if cleanup removed the unlocked file and left db and locked file
-	files, err := os.ReadDir(tempDir)
-	require.NoError(t, err)
-	require.Equal(t, 2, len(files))
-	require.Equal(t, "my_db", files[0].Name())
-	_, f := filepath.Split(tempLocked.Name())
-	require.Equal(t, f, files[1].Name())
-}
-
 func BenchmarkClientGet(b *testing.B) {
 	tempDir := b.TempDir()
 	dbFile := filepath.Join(tempDir, "my_db")

extension/storage/filestorage/extension.go

@@ -5,7 +5,9 @@ package filestorage // import "github.com/open-telemetry/opentelemetry-collector
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"os"
 	"path/filepath"
 	"strings"
 
@@ -42,6 +44,9 @@ func newLocalFileStorage(logger *zap.Logger, config *Config) (extension.Extensio
 
 // Start does nothing
 func (lfs *localFileStorage) Start(context.Context, component.Host) error {
+	if lfs.cfg.Compaction.CleanupOnStart {
+		lfs.cleanup(lfs.cfg.Compaction.Directory)
+	}
 	return nil
 }
 
@@ -135,3 +140,30 @@ func isSafe(character rune) bool {
 	}
 	return false
 }
+
+// cleanup left compaction temporary files from previous killed process
+func (c *localFileStorage) cleanup(compactionDirectory string) error {
+	pattern := filepath.Join(compactionDirectory, fmt.Sprintf("%s*", TempDbPrefix))
+	contents, err := filepath.Glob(pattern)
+	if err != nil {
+		c.logger.Info("cleanup error listing temporary files",
+			zap.Error(err))
+		return err
+	}
+
+	var errs []error
+	for _, item := range contents {
+		err = os.Remove(item)
+		if err == nil {
+			c.logger.Debug("cleanup",
+				zap.String("deletedFile", item))
+		} else {
+			errs = append(errs, err)
+		}
+	}
+	if errs != nil {
+		c.logger.Info("cleanup errors",
+			zap.Error(errors.Join(errs...)))
+	}
+	return nil
+}

extension/storage/filestorage/extension_test.go

@@ -14,6 +14,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/extension/experimental/storage"
 	"go.opentelemetry.io/collector/extension/extensiontest"
 )
@@ -454,7 +455,7 @@ func TestCleanupOnStart(t *testing.T) {
 
 	tempDir := t.TempDir()
 	// simulate left temporary compaction file from killed process
-	temp, _ := os.CreateTemp(tempDir, tempDbPrefix)
+	temp, _ := os.CreateTemp(tempDir, TempDbPrefix)
 	temp.Close()
 
 	f := NewFactory()
@@ -467,6 +468,7 @@ func TestCleanupOnStart(t *testing.T) {
 
 	se, ok := extension.(storage.Extension)
 	require.True(t, ok)
+	require.NoError(t, se.Start(ctx, componenttest.NewNopHost()))
 
 	client, err := se.GetClient(
 		ctx,