[extension/storage/filestorage] Add CleanupOnStart flag for compaction temporary files

.chloggen/extension-storage-filestorage-cleanup_on_start.yaml

@@ -0,0 +1,29 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: extension/storage/filestorage
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: New flag cleanup_on_start for the compaction section (default=false).
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [32863]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext: |
+  It will remove all temporary files in the compaction directory (those which start with `tempdb`),
+  temp files will be left if a previous run of the process is killed while compacting.
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

extension/storage/filestorage/README.md

@@ -34,6 +34,10 @@ The default timeout is `1s`.
 `compaction.max_transaction_size` (default: 65536): defines maximum size of the compaction transaction.
 A value of zero will ignore transaction sizes.
 
+`compaction.cleanup_on_start` (default: false) - specifies if removal of compaction temporary files is performed on start.
+It will remove all temporary files in the compaction directory (those which start with `tempdb`),
+temp files will be left if a previous run of the process is killed while compacting.
+
 ### Rebound (online) compaction
 
 For rebound compaction, there are two additional parameters available:

extension/storage/filestorage/client.go

@@ -20,6 +20,8 @@ import (
 var defaultBucket = []byte(`default`)
 
 const (
+	TempDbPrefix = "tempdb"
+
 	elapsedKey       = "elapsed"
 	directoryKey     = "directory"
 	tempDirectoryKey = "tempDirectory"
@@ -152,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, "tempdb")
+	file, err = os.CreateTemp(compactionDirectory, TempDbPrefix)
 	if err != nil {
 		return err
 	}

extension/storage/filestorage/config.go

@@ -45,6 +45,10 @@ type CompactionConfig struct {
 	MaxTransactionSize int64 `mapstructure:"max_transaction_size,omitempty"`
 	// CheckInterval specifies frequency of compaction check
 	CheckInterval time.Duration `mapstructure:"check_interval,omitempty"`
+	// CleanupOnStart specifies removal of temporary files is performed on start.
+	// It will remove all the files in the compaction directory starting with tempdb,
+	// temp files will be left if a previous run of the process is killed while compacting.
+	CleanupOnStart bool `mapstructure:"cleanup_on_start,omitempty"`
 }
 
 func (cfg *Config) Validate() error {

extension/storage/filestorage/config_test.go

@@ -45,6 +45,7 @@ func TestLoadConfig(t *testing.T) {
 					ReboundTriggerThresholdMiB: 16,
 					ReboundNeededThresholdMiB:  128,
 					CheckInterval:              time.Second * 5,
+					CleanupOnStart:             true,
 				},
 				Timeout: 2 * time.Second,
 				FSync:   true,

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"
 )
@@ -448,3 +449,39 @@ func TestCompactionRemoveTemp(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, 0, len(files))
 }
+
+func TestCleanupOnStart(t *testing.T) {
+	ctx := context.Background()
+
+	tempDir := t.TempDir()
+	// simulate left temporary compaction file from killed process
+	temp, _ := os.CreateTemp(tempDir, TempDbPrefix)
+	temp.Close()
+
+	f := NewFactory()
+	cfg := f.CreateDefaultConfig().(*Config)
+	cfg.Directory = tempDir
+	cfg.Compaction.Directory = tempDir
+	cfg.Compaction.CleanupOnStart = true
+	extension, err := f.CreateExtension(context.Background(), extensiontest.NewNopCreateSettings(), cfg)
+	require.NoError(t, err)
+
+	se, ok := extension.(storage.Extension)
+	require.True(t, ok)
+	require.NoError(t, se.Start(ctx, componenttest.NewNopHost()))
+
+	client, err := se.GetClient(
+		ctx,
+		component.KindReceiver,
+		newTestEntity("my_component"),
+		"",
+	)
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		require.NoError(t, client.Close(ctx))
+	})
+
+	files, err := os.ReadDir(tempDir)
+	require.NoError(t, err)
+	require.Equal(t, 1, len(files))
+}

extension/storage/filestorage/factory.go

@@ -45,6 +45,7 @@ func createDefaultConfig() component.Config {
 			ReboundNeededThresholdMiB:  defaultReboundNeededThresholdMib,
 			ReboundTriggerThresholdMiB: defaultReboundTriggerThresholdMib,
 			CheckInterval:              defaultCompactionInterval,
+			CleanupOnStart:             false,
 		},
 		Timeout: time.Second,
 		FSync:   false,

extension/storage/filestorage/testdata/config.yaml

@@ -12,5 +12,6 @@ file_storage/all_settings:
     rebound_trigger_threshold_mib: 16
     rebound_needed_threshold_mib: 128
     max_transaction_size: 2048
+    cleanup_on_start: true
   timeout: 2s
   fsync: true