Merge branch 'main' into extension/encoding/googlecloudlogentryencodingextension

Author: anarwal

.chloggen/add-lambda-helpers.yaml

@@ -0,0 +1,37 @@
+# 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. receiver/filelog)
+component: pkg/ottl
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add `ottl.NewTestingLambdaExpression` and internal helpers for implementing lambda-based OTTL functions.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [49180]
+
+# (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: |
+  Adds `ottl.NewTestingLambdaExpression`, a test helper that builds a `LambdaExpression`
+  with a custom value body. The callback receives `resolveBinding func(string) any` to
+  read formal parameters from the active local scope during evaluation.
+
+  Also introduces internal helpers to support lambda-based OTTL functions:
+  - `funcutil.GetSliceOrMapValue` coerces a getter result to `pcommon.Slice` or `pcommon.Map`.
+  - `funcutil.EvaluateBiFunction` and `funcutil.EvaluateBiPredicate` bind two positional
+    arguments, normalize them, and evaluate a `LambdaActivation`.
+  - `ottlcommon.NormalizeValue` converts values, including `pcommon.Value`, into types
+    used by OTTL comparisons.
+
+# 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: [api]

.chloggen/file-storage-max-size.yaml

@@ -0,0 +1,27 @@
+# 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/file_storage
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Added `max_size` support to the filestorage extension to cap per-component database growth.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [38620]
+
+# (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:
+
+# 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: []

.chloggen/promreceiver-ignore-scope-info-beta.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. receiver/filelog)
+component: receiver/prometheus
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Promote `receiver.prometheusreceiver.IgnoreScopeInfoMetric` feature gate to beta.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [47312]
+
+# (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: |
+  The `otel_scope_info` metric is now ignored for scope attribute extraction by default.
+  To temporarily restore the previous behavior, disable the feature gate with `--feature-gates=-receiver.prometheusreceiver.IgnoreScopeInfoMetric`.
+
+# 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

@@ -27,6 +27,13 @@ The default directory is `%ProgramData%\Otelcol\FileStorage` on Windows and `/va
 `timeout` is the maximum time to wait for a file lock. This value does not need to be modified in most circumstances.
 The default timeout is `1s`.
 
+`max_size` sets the maximum on-disk size of each bbolt database file in bytes.
+When a write would need the file to grow past this limit, the write is rejected with a storage-full error.
+A value of `0` means unlimited size.
+Writes that fit into already-allocated free space are still allowed, even when the file is already at the configured limit.
+When rebound compaction is enabled, `max_size` must be greater than or equal to both
+`compaction.rebound_needed_threshold_mib * 1,048,576` and `compaction.rebound_trigger_threshold_mib * 1,048,576`.
+
 `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 directories if they do not already exist.
@@ -56,6 +63,9 @@ A value of zero will ignore transaction sizes.
 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 `max_size` is set, both `compaction.rebound_needed_threshold_mib` and `compaction.rebound_trigger_threshold_mib`
+must be less than or equal to that limit after converting MiB to bytes.
+
 ### Rebound (online) compaction
 
 For rebound compaction, there are two additional parameters available:
@@ -94,6 +104,7 @@ extensions:
   file_storage/all_settings:
     directory: /var/lib/otelcol/mydir
     timeout: 1s
+    max_size: 268435456
     recreate: true
     compaction:
       on_start: true

extension/storage/filestorage/client.go

@@ -13,6 +13,7 @@ import (
 	"time"
 
 	"go.etcd.io/bbolt"
+	berrors "go.etcd.io/bbolt/errors"
 	"go.opentelemetry.io/collector/extension/xextension/storage"
 	"go.uber.org/zap"
 )
@@ -38,25 +39,29 @@ type fileStorageClient struct {
 	logger          *zap.Logger
 	compactionMutex sync.RWMutex
 	db              *bbolt.DB
+	dbOptions       bbolt.Options
 	compactionCfg   *CompactionConfig
-	openTimeout     time.Duration
 	stopCh          chan struct{}
 	wg              sync.WaitGroup
 	closed          bool
 }
 
-func bboltOptions(timeout time.Duration, noSync bool) *bbolt.Options {
-	return &bbolt.Options{
-		Timeout:        timeout,
-		NoSync:         noSync,
+func bboltOptions(cfg *Config) bbolt.Options {
+	options := bbolt.Options{
+		Timeout:        cfg.Timeout,
+		NoSync:         !cfg.FSync,
 		NoFreelistSync: true,
 		FreelistType:   bbolt.FreelistMapType,
 	}
+	if cfg.MaxSize > 0 {
+		options.MaxSize = int(cfg.MaxSize)
+	}
+	return options
 }
 
-func newClient(logger *zap.Logger, filePath string, timeout time.Duration, compactionCfg *CompactionConfig, noSync bool) (*fileStorageClient, error) {
-	options := bboltOptions(timeout, noSync)
-	db, err := bbolt.Open(filePath, 0o600, options)
+func newClient(logger *zap.Logger, filePath string, cfg *Config) (*fileStorageClient, error) {
+	options := bboltOptions(cfg)
+	db, err := bbolt.Open(filePath, 0o600, &options)
 	if err != nil {
 		return nil, err
 	}
@@ -73,12 +78,12 @@ func newClient(logger *zap.Logger, filePath string, timeout time.Duration, compa
 	client := &fileStorageClient{
 		logger:        logger,
 		db:            db,
-		compactionCfg: compactionCfg,
-		openTimeout:   timeout,
+		dbOptions:     options,
+		compactionCfg: cfg.Compaction,
 		stopCh:        make(chan struct{}),
 		wg:            sync.WaitGroup{},
 	}
-	if compactionCfg.OnRebound {
+	if cfg.Compaction.OnRebound {
 		client.startCompactionLoop()
 	}
 
@@ -119,7 +124,7 @@ func (c *fileStorageClient) Batch(_ context.Context, ops ...*storage.Operation)
 
 	c.compactionMutex.RLock()
 	defer c.compactionMutex.RUnlock()
-	return c.db.Update(batch)
+	return normalizeStorageError(c.db.Update(batch))
 }
 
 // Walk implements storage.Walker.
@@ -135,7 +140,7 @@ func (c *fileStorageClient) Walk(ctx context.Context, fn storage.WalkFunc) error
 		return errors.New("storage is closed")
 	}
 
-	return c.db.Update(func(tx *bbolt.Tx) error {
+	return normalizeStorageError(c.db.Update(func(tx *bbolt.Tx) error {
 		bucket := tx.Bucket(defaultBucket)
 		if bucket == nil {
 			return errors.New("storage not initialized")
@@ -157,8 +162,9 @@ func (c *fileStorageClient) Walk(ctx context.Context, fn storage.WalkFunc) error
 			}
 			opBatch = append(opBatch, ops...)
 		}
+
 		return updateBucket(bucket, opBatch...)
-	})
+	}))
 }
 
 // updateBucket executes the specified operations in order for a given bucket. Get operation results are updated in place
@@ -212,7 +218,6 @@ func (c *fileStorageClient) Close(_ context.Context) error {
 func (c *fileStorageClient) Compact(compactionDirectory string, timeout time.Duration, maxTransactionSize int64) error {
 	var err error
 	var file *os.File
-	var compactedDb *bbolt.DB
 
 	// create temporary file in compactionDirectory
 	file, err = os.CreateTemp(compactionDirectory, TempDbPrefix)
@@ -235,7 +240,8 @@ func (c *fileStorageClient) Compact(compactionDirectory string, timeout time.Dur
 	}()
 
 	// use temporary file as compaction target
-	options := bboltOptions(timeout, c.db.NoSync)
+	options := c.dbOptions
+	options.Timeout = timeout
 
 	c.compactionMutex.Lock()
 	defer c.compactionMutex.Unlock()
@@ -249,7 +255,7 @@ func (c *fileStorageClient) Compact(compactionDirectory string, timeout time.Dur
 		zap.String(tempDirectoryKey, file.Name()))
 
 	// cannot reuse newClient as db shouldn't contain any bucket
-	compactedDb, err = bbolt.Open(file.Name(), 0o600, options)
+	compactedDb, err := bbolt.Open(file.Name(), 0o600, &options)
 	if err != nil {
 		return err
 	}
@@ -258,6 +264,7 @@ func (c *fileStorageClient) Compact(compactionDirectory string, timeout time.Dur
 
 	err = bbolt.Compact(compactedDb, c.db, maxTransactionSize)
 	if err != nil {
+		_ = compactedDb.Close()
 		return err
 	}
 
@@ -270,7 +277,7 @@ func (c *fileStorageClient) Compact(compactionDirectory string, timeout time.Dur
 	// replace current db file with compacted db file
 	// we reopen the DB file irrespective of the success of the replace, as we can't leave it closed
 	moveErr := moveFileWithFallback(compactedDbPath, dbPath)
-	newDb, openErr := bbolt.Open(dbPath, 0o600, options)
+	newDb, openErr := bbolt.Open(dbPath, 0o600, &options)
 	if openErr != nil {
 		// Leave c.db pointing at the old (closed) DB so that callers get
 		// errors from the closed DB instead of panicking on a nil pointer.
@@ -314,7 +321,7 @@ func (c *fileStorageClient) startCompactionLoop() {
 			select {
 			case <-compactionTicker.C:
 				if c.shouldCompact() {
-					err := c.Compact(c.compactionCfg.Directory, c.openTimeout, c.compactionCfg.MaxTransactionSize)
+					err := c.Compact(c.compactionCfg.Directory, c.dbOptions.Timeout, c.compactionCfg.MaxTransactionSize)
 					if err != nil {
 						c.logger.Error("compaction failure",
 							zap.String(directoryKey, c.compactionCfg.Directory),
@@ -329,6 +336,13 @@ func (c *fileStorageClient) startCompactionLoop() {
 	})
 }
 
+func normalizeStorageError(err error) error {
+	if errors.Is(err, berrors.ErrMaxSizeReached) {
+		return storage.ErrStorageFull
+	}
+	return err
+}
+
 // shouldCompact checks whether the conditions for online compaction are met
 func (c *fileStorageClient) shouldCompact() bool {
 	if !c.compactionCfg.OnRebound {