[filestorage]: Provide an option to the user to create a directory (#34985)

**Description:** 
This PR introduces a new option that will create a directory for the
user if it doesn't already exist.

**Link to tracking Issue:**
#34939

**Testing:** Added

---------

Co-authored-by: Daniel Jaglowski <[email protected]>
Co-authored-by: Andrzej Stencel <[email protected]>

Author: 3 people

.chloggen/filestorage-create-directort.yaml

@@ -0,0 +1,13 @@
+# 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: file_storage
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: provide a new option to the user to create a directory on start
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [34939]

extension/storage/filestorage/README.md

@@ -24,6 +24,9 @@ The default timeout is `1s`.
 
 `fsync` when set, will force the database to perform an fsync after each write.  This helps to ensure database integrity if there is an interruption to the database process, but at the cost of performance.  See [DB.NoSync](https://pkg.go.dev/go.etcd.io/bbolt#DB) for more information.
 
+`create_directory` when set, will create the data storage and compaction directory if it does not already exist. The directory will be created with `0750 (rwxr-x--)` permissions, by default. Use `directory_permissions` to customize directory creation permissions.
+
+
 ## Compaction
 `compaction` defines how and when files should be compacted. There are two modes of compaction available (both of which can be set concurrently):
 - `compaction.on_start` (default: false), which happens when collector starts

extension/storage/filestorage/config.go

@@ -8,9 +8,13 @@ import (
 	"fmt"
 	"io/fs"
 	"os"
+	"strconv"
 	"time"
 )
 
+var errInvalidOctal = errors.New("directory_permissions value must be a valid octal representation")
+var errInvalidPermissionBits = errors.New("directory_permissions contain invalid bits for file access")
+
 // Config defines configuration for file storage extension.
 type Config struct {
 	Directory string        `mapstructure:"directory,omitempty"`
@@ -20,6 +24,11 @@ type Config struct {
 
 	// FSync specifies that fsync should be called after each database write
 	FSync bool `mapstructure:"fsync,omitempty"`
+
+	// CreateDirectory specifies that the directory should be created automatically by the extension on start
+	CreateDirectory            bool   `mapstructure:"create_directory,omitempty"`
+	DirectoryPermissions       string `mapstructure:"directory_permissions,omitempty"`
+	directoryPermissionsParsed int64  `mapstructure:"-,omitempty"`
 }
 
 // CompactionConfig defines configuration for optional file storage compaction.
@@ -59,18 +68,16 @@ func (cfg *Config) Validate() error {
 		dirs = []string{cfg.Directory}
 	}
 	for _, dir := range dirs {
-		info, err := os.Stat(dir)
-		if err != nil {
-			if os.IsNotExist(err) {
-				return fmt.Errorf("directory must exist: %w", err)
+		if info, err := os.Stat(dir); err != nil {
+			if !cfg.CreateDirectory && os.IsNotExist(err) {
+				return fmt.Errorf("directory must exist: %w. You can enable the create_directory option to automatically create it", err)
 			}
 
 			fsErr := &fs.PathError{}
-			if errors.As(err, &fsErr) {
+			if errors.As(err, &fsErr) && !os.IsNotExist(err) {
 				return fmt.Errorf("problem accessing configured directory: %s, err: %w", dir, fsErr)
 			}
-		}
-		if !info.IsDir() {
+		} else if !info.IsDir() {
 			return fmt.Errorf("%s is not a directory", dir)
 		}
 	}
@@ -83,5 +90,15 @@ func (cfg *Config) Validate() error {
 		return errors.New("compaction check interval must be positive when rebound compaction is set")
 	}
 
+	if cfg.CreateDirectory {
+		permissions, err := strconv.ParseInt(cfg.DirectoryPermissions, 8, 32)
+		if err != nil {
+			return errInvalidOctal
+		} else if permissions&int64(os.ModePerm) != permissions {
+			return errInvalidPermissionBits
+		}
+		cfg.directoryPermissionsParsed = permissions
+	}
+
 	return nil
 }

extension/storage/filestorage/config_test.go

@@ -10,10 +10,12 @@ import (
 	"testing"
 	"time"
 
+	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/confmap/confmaptest"
+	"go.opentelemetry.io/collector/extension"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/storage/filestorage/internal/metadata"
 )
@@ -47,8 +49,10 @@ func TestLoadConfig(t *testing.T) {
 					CheckInterval:              time.Second * 5,
 					CleanupOnStart:             true,
 				},
-				Timeout: 2 * time.Second,
-				FSync:   true,
+				Timeout:              2 * time.Second,
+				FSync:                true,
+				CreateDirectory:      false,
+				DirectoryPermissions: "0750",
 			},
 		},
 	}
@@ -95,6 +99,115 @@ func TestHandleProvidingFilePathAsDirWithAnError(t *testing.T) {
 	require.Error(t, err)
 	require.EqualError(t, err, file.Name()+" is not a directory")
 }
+func TestDirectoryCreateConfig(t *testing.T) {
+	tests := []struct {
+		name   string
+		config func(*testing.T, extension.Factory) *Config
+		err    error
+	}{
+		{
+			name: "create directory true - no error",
+			config: func(t *testing.T, f extension.Factory) *Config {
+				storageDir := filepath.Join(t.TempDir(), uuid.NewString())
+				cfg := f.CreateDefaultConfig().(*Config)
+				cfg.Directory = storageDir
+				cfg.CreateDirectory = true
+				return cfg
+			},
+			err: nil,
+		},
+		{
+			name: "create directory true - no error - 0700 permissions",
+			config: func(t *testing.T, f extension.Factory) *Config {
+				storageDir := filepath.Join(t.TempDir(), uuid.NewString())
+				cfg := f.CreateDefaultConfig().(*Config)
+				cfg.Directory = storageDir
+				cfg.CreateDirectory = true
+				cfg.DirectoryPermissions = "0700"
+				return cfg
+			},
+			err: nil,
+		},
+		{
+			name: "create directory false - error",
+			config: func(t *testing.T, f extension.Factory) *Config {
+				storageDir := filepath.Join(t.TempDir(), uuid.NewString())
+				cfg := f.CreateDefaultConfig().(*Config)
+				cfg.Directory = storageDir
+				cfg.CreateDirectory = false
+				return cfg
+			},
+			err: os.ErrNotExist,
+		},
+		{
+			name: "create directory true - invalid permissions",
+			config: func(t *testing.T, f extension.Factory) *Config {
+				storageDir := filepath.Join(t.TempDir(), uuid.NewString())
+				cfg := f.CreateDefaultConfig().(*Config)
+				cfg.Directory = storageDir
+				cfg.CreateDirectory = true
+				cfg.DirectoryPermissions = "invalid string"
+				return cfg
+			},
+			err: errInvalidOctal,
+		},
+		{
+			name: "create directory true - rwxr--r-- (should be octal string)",
+			config: func(t *testing.T, f extension.Factory) *Config {
+				storageDir := filepath.Join(t.TempDir(), uuid.NewString())
+				cfg := f.CreateDefaultConfig().(*Config)
+				cfg.Directory = storageDir
+				cfg.CreateDirectory = true
+				cfg.DirectoryPermissions = "rwxr--r--"
+				return cfg
+			},
+			err: errInvalidOctal,
+		},
+		{
+			name: "create directory true - 0778 (invalid octal)",
+			config: func(t *testing.T, f extension.Factory) *Config {
+				storageDir := filepath.Join(t.TempDir(), uuid.NewString())
+				cfg := f.CreateDefaultConfig().(*Config)
+				cfg.Directory = storageDir
+				cfg.CreateDirectory = true
+				cfg.DirectoryPermissions = "0778"
+				return cfg
+			},
+			err: errInvalidOctal,
+		},
+		{
+			name: "create directory true - 07771 (invalid permission bits)",
+			config: func(t *testing.T, f extension.Factory) *Config {
+				storageDir := filepath.Join(t.TempDir(), uuid.NewString())
+				cfg := f.CreateDefaultConfig().(*Config)
+				cfg.Directory = storageDir
+				cfg.CreateDirectory = true
+				cfg.DirectoryPermissions = "07771"
+				return cfg
+			},
+			err: errInvalidPermissionBits,
+		},
+		{
+			name: "create directory false - 07771 (invalid string) - no error",
+			config: func(t *testing.T, f extension.Factory) *Config {
+				cfg := f.CreateDefaultConfig().(*Config)
+				cfg.Directory = t.TempDir()
+				cfg.CreateDirectory = false
+				cfg.DirectoryPermissions = "07771"
+				return cfg
+
+			},
+			err: nil,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			f := NewFactory()
+			config := tt.config(t, f)
+			require.ErrorIs(t, config.Validate(), tt.err)
+		})
+	}
+}
 
 func TestCompactionDirectory(t *testing.T) {
 	f := NewFactory()
@@ -157,5 +270,4 @@ func TestCompactionDirectory(t *testing.T) {
 			require.ErrorIs(t, component.ValidateConfig(test.config(t)), test.err)
 		})
 	}
-
 }

extension/storage/filestorage/extension.go

@@ -26,6 +26,19 @@ type localFileStorage struct {
 var _ storage.Extension = (*localFileStorage)(nil)
 
 func newLocalFileStorage(logger *zap.Logger, config *Config) (extension.Extension, error) {
+	if config.CreateDirectory {
+		var dirs []string
+		if config.Compaction.OnStart || config.Compaction.OnRebound {
+			dirs = []string{config.Directory, config.Compaction.Directory}
+		} else {
+			dirs = []string{config.Directory}
+		}
+		for _, dir := range dirs {
+			if err := ensureDirectoryExists(dir, os.FileMode(config.directoryPermissionsParsed)); err != nil {
+				return nil, err
+			}
+		}
+	}
 	return &localFileStorage{
 		cfg:    config,
 		logger: logger,
@@ -129,6 +142,14 @@ func isSafe(character rune) bool {
 	return false
 }
 
+func ensureDirectoryExists(path string, perm os.FileMode) error {
+	if _, err := os.Stat(path); os.IsNotExist(err) {
+		return os.MkdirAll(path, perm)
+	}
+	// we already handled other errors in config.Validate(), so it's okay to return nil
+	return nil
+}
+
 // cleanup left compaction temporary files from previous killed process
 func (lfs *localFileStorage) cleanup(compactionDirectory string) error {
 	pattern := filepath.Join(compactionDirectory, fmt.Sprintf("%s*", TempDbPrefix))

extension/storage/filestorage/extension_test.go

@@ -8,13 +8,16 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"runtime"
 	"sync"
 	"testing"
 
+	"github.com/google/uuid"
 	"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"
 	"go.opentelemetry.io/collector/extension/experimental/storage"
 	"go.opentelemetry.io/collector/extension/extensiontest"
 	"go.uber.org/zap"
@@ -525,3 +528,89 @@ func TestCompactionOnStart(t *testing.T) {
 		require.NoError(t, client.Close(context.TODO()))
 	})
 }
+
+func TestDirectoryCreation(t *testing.T) {
+	tests := []struct {
+		name     string
+		config   func(*testing.T, extension.Factory) *Config
+		validate func(*testing.T, *Config)
+	}{
+		{
+			name: "create directory true - no error",
+			config: func(t *testing.T, f extension.Factory) *Config {
+				tempDir := t.TempDir()
+				storageDir := filepath.Join(tempDir, uuid.NewString())
+				cfg := f.CreateDefaultConfig().(*Config)
+				cfg.Directory = storageDir
+				cfg.CreateDirectory = true
+				cfg.DirectoryPermissions = "0750"
+				require.NoError(t, cfg.Validate())
+				return cfg
+			},
+			validate: func(t *testing.T, cfg *Config) {
+				require.DirExists(t, cfg.Directory)
+				s, err := os.Stat(cfg.Directory)
+				require.NoError(t, err)
+				var expectedFileMode os.FileMode
+				if runtime.GOOS == "windows" { // on Windows, we get 0777 for writable directories
+					expectedFileMode = os.FileMode(0777)
+				} else {
+					expectedFileMode = os.FileMode(0750)
+				}
+				require.Equal(t, expectedFileMode, s.Mode()&os.ModePerm)
+			},
+		},
+		{
+			name: "create directory true - no error - 0700 permissions",
+			config: func(t *testing.T, f extension.Factory) *Config {
+				tempDir := t.TempDir()
+				storageDir := filepath.Join(tempDir, uuid.NewString())
+				cfg := f.CreateDefaultConfig().(*Config)
+				cfg.Directory = storageDir
+				cfg.DirectoryPermissions = "0700"
+				cfg.CreateDirectory = true
+				require.NoError(t, cfg.Validate())
+				return cfg
+			},
+			validate: func(t *testing.T, cfg *Config) {
+				require.DirExists(t, cfg.Directory)
+				s, err := os.Stat(cfg.Directory)
+				require.NoError(t, err)
+				var expectedFileMode os.FileMode
+				if runtime.GOOS == "windows" { // on Windows, we get 0777 for writable directories
+					expectedFileMode = os.FileMode(0777)
+				} else {
+					expectedFileMode = os.FileMode(0700)
+				}
+				require.Equal(t, expectedFileMode, s.Mode()&os.ModePerm)
+			},
+		},
+		{
+			name: "create directory false - error",
+			config: func(t *testing.T, f extension.Factory) *Config {
+				tempDir := t.TempDir()
+				storageDir := filepath.Join(tempDir, uuid.NewString())
+				cfg := f.CreateDefaultConfig().(*Config)
+				cfg.Directory = storageDir
+				cfg.CreateDirectory = false
+				require.ErrorIs(t, cfg.Validate(), os.ErrNotExist)
+				return cfg
+			},
+			validate: func(t *testing.T, cfg *Config) {
+				require.NoDirExists(t, cfg.Directory)
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			f := NewFactory()
+			config := tt.config(t, f)
+			if config != nil {
+				ext, err := f.CreateExtension(context.Background(), extensiontest.NewNopSettings(), config)
+				require.NoError(t, err)
+				require.NotNil(t, ext)
+				tt.validate(t, config)
+			}
+		})
+	}
+}

extension/storage/filestorage/factory.go

@@ -47,8 +47,10 @@ func createDefaultConfig() component.Config {
 			CheckInterval:              defaultCompactionInterval,
 			CleanupOnStart:             false,
 		},
-		Timeout: time.Second,
-		FSync:   false,
+		Timeout:              time.Second,
+		FSync:                false,
+		CreateDirectory:      false,
+		DirectoryPermissions: "0750",
 	}
 }