Skip to content

[xextension/storage] Add Walker interface for storage extensions#15190

Merged
mx-psi merged 6 commits into
open-telemetry:mainfrom
rdner:storage-walker-interface
May 14, 2026
Merged

[xextension/storage] Add Walker interface for storage extensions#15190
mx-psi merged 6 commits into
open-telemetry:mainfrom
rdner:storage-walker-interface

Conversation

@rdner

@rdner rdner commented Apr 24, 2026

Copy link
Copy Markdown
Member

Description

Introduce Walker interface, WalkFunc type, and SkipAll error value to support iterating over storage key/value entries.

This enables consumers to implement storage migrations or any functionality that involves ranging through the storage entries.

One of the examples of such functionality may be a TTL-based garbage collection. Values can have a TTL marker, a routine runs with an interval iterating through entries searching for expired entries. WalkFunc produces Delete operations for each expired entry.

The type names and the interface is close to the ones in filepath package of the standard library.

Example: TTL-based garbage collection using Walker

This example shows how to use the Walker interface to implement a TTL-based
garbage collection routine. Each stored value carries a TTL marker (an expiration
timestamp prefix). A periodic routine walks all entries, and the WalkFunc
produces Delete operations for every expired entry.

Values are prefixed with an 8-byte big-endian Unix timestamp (seconds) that
represents the expiration time. The rest of the bytes are the actual payload.

import (
	"encoding/binary"
	"time"
)

func encodeWithTTL(value []byte, ttl time.Duration) []byte {
	expireAt := time.Now().Add(ttl).Unix()
	buf := make([]byte, 8+len(value))
	binary.BigEndian.PutUint64(buf[:8], uint64(expireAt))
	copy(buf[8:], value)
	return buf
}

func decodeExpiration(value []byte) (time.Time, []byte) {
	ts := binary.BigEndian.Uint64(value[:8])
	return time.Unix(int64(ts), 0), value[8:]
}

Garbage collection with Walk:

import (
	"context"
	"fmt"
	"time"

	"go.opentelemetry.io/collector/extension/xextension/storage"
)

func collectExpired(ctx context.Context, client storage.Client) error {
	walker, ok := client.(storage.Walker)
	if !ok {
		return fmt.Errorf("storage client does not implement Walker")
	}

	now := time.Now()
	return walker.Walk(ctx, func(key string, value []byte) ([]*storage.Operation, error) {
		if len(value) < 8 {
			return nil, nil
		}
		expireAt, _ := decodeExpiration(value)
		if now.After(expireAt) {
			return []*storage.Operation{storage.DeleteOperation(key)}, nil
		}
		return nil, nil
	})
}

All Delete operations returned by the WalkFunc are collected during the walk
and applied atomically once the iteration completes. If the callback returns a
non-nil error (other than SkipAll), or if the walk encounters an internal
error, no operations are applied.

Documentation

The package's README.md has been updated. The new types have been fully documented.

Closes #15191

Introduce Walker interface, WalkFunc type, and SkipAll error value
to support iterating over storage key/value entries.

This enables consumers to implement storage migrations or any
functionality that involves ranging through the storage entries.

One of the examples of such functionality may be a TTL-based garbage
collection. Values can have a TTL marker, a routine runs with an
interval iterating through entries searching for expired entries. WalkFunc
produces Delete operations for each expired entry.
@rdner rdner changed the title Add Walker interface for storage extensions [xextension/storage] Add Walker interface for storage extensions Apr 24, 2026
@codspeed-hq

codspeed-hq Bot commented Apr 24, 2026

Copy link
Copy Markdown

Merging this PR will not alter performance

⚠️ Unknown Walltime execution environment detected

Using the Walltime instrument on standard Hosted Runners will lead to inconsistent data.

For the most accurate results, we recommend using CodSpeed Macro Runners: bare-metal machines fine-tuned for performance measurement consistency.

✅ 7 untouched benchmarks
⏩ 76 skipped benchmarks1


Comparing rdner:storage-walker-interface (3f80d60) with main (0814ca6)

Open in CodSpeed

Footnotes

  1. 76 benchmarks were skipped, so the baseline results were used instead. If they were deleted from the codebase, click here and archive them to remove them from the performance reports.

@codecov

codecov Bot commented Apr 24, 2026

Copy link
Copy Markdown

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 91.14%. Comparing base (d772145) to head (3f80d60).
⚠️ Report is 56 commits behind head on main.

Additional details and impacted files
@@            Coverage Diff             @@
##             main   #15190      +/-   ##
==========================================
- Coverage   91.32%   91.14%   -0.18%     
==========================================
  Files         700      701       +1     
  Lines       45316    45798     +482     
==========================================
+ Hits        41383    41742     +359     
- Misses       2782     2843      +61     
- Partials     1151     1213      +62     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:
  • ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

@rdner
rdner marked this pull request as ready for review May 4, 2026 14:58
@rdner
rdner requested a review from a team as a code owner May 4, 2026 14:58
@rdner
rdner requested a review from songy23 May 4, 2026 14:58
// SkipAll is used as a return value from WalkFunc to indicate that
// all remaining storage entries are to be skipped.
// The pending operations are still applied.
var SkipAll = errors.New("skip everything and stop the walk") //nolint:revive,staticcheck // mimics the existing API in the standard library, see filepath

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There is a linter issue that failed the CI:

Error: extension/xextension/storage/storage.go:87:5: error-naming: error var SkipAll should have name of the form ErrFoo (revive)
var SkipAll = errors.New("skip everything and stop the walk")
^

I'd rather prefer to keep the name as it is since it mimics the existing API in the standard library.

@rdner

rdner commented May 7, 2026

Copy link
Copy Markdown
Member Author

@songy23 please let me know if I can do anything to help with the review.

@songy23 songy23 left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM

@songy23 songy23 added the ready-to-merge Code review completed; ready to merge by maintainers label May 7, 2026
@VihasMakwana

Copy link
Copy Markdown
Contributor

This looks good to go @open-telemetry/collector-maintainers

@songy23
songy23 requested a review from mx-psi May 13, 2026 16:11
@jmacd

jmacd commented May 13, 2026

Copy link
Copy Markdown
Contributor

FYI @carsonip

@mx-psi
mx-psi added this pull request to the merge queue May 14, 2026
Merged via the queue into open-telemetry:main with commit d3b2c86 May 14, 2026
84 checks passed
@otelbot

otelbot Bot commented May 14, 2026

Copy link
Copy Markdown
Contributor

Thank you for your contribution @rdner! 🎉 We would like to hear from you about your experience contributing to OpenTelemetry by taking a few minutes to fill out this survey.

@rdner
rdner deleted the storage-walker-interface branch May 14, 2026 11:31
@dmitryax

Copy link
Copy Markdown
Member

Should we apply similar interface as we have for pcommon.Map

func (m Map) All() iter.Seq2[string, Value] {
?

swiatekm pushed a commit to swiatekm/opentelemetry-collector that referenced this pull request May 15, 2026
…n-telemetry#15190)

#### Description

Introduce `Walker` interface, `WalkFunc` type, and `SkipAll` error value
to support iterating over storage key/value entries.

This enables consumers to implement storage migrations or any
functionality that involves ranging through the storage entries.

One of the examples of such functionality may be a TTL-based garbage
collection. Values can have a TTL marker, a routine runs with an
interval iterating through entries searching for expired entries.
`WalkFunc` produces `Delete` operations for each expired entry.

The type names and the interface is close to the ones in
[`filepath`](https://pkg.go.dev/path/filepath#WalkFunc) package of the
standard library.

Example: TTL-based garbage collection using `Walker`

This example shows how to use the `Walker` interface to implement a
TTL-based
garbage collection routine. Each stored value carries a TTL marker (an
expiration
timestamp prefix). A periodic routine walks all entries, and the
`WalkFunc`
produces `Delete` operations for every expired entry.

Values are prefixed with an 8-byte big-endian Unix timestamp (seconds)
that
represents the expiration time. The rest of the bytes are the actual
payload.

```go
import (
	"encoding/binary"
	"time"
)

func encodeWithTTL(value []byte, ttl time.Duration) []byte {
	expireAt := time.Now().Add(ttl).Unix()
	buf := make([]byte, 8+len(value))
	binary.BigEndian.PutUint64(buf[:8], uint64(expireAt))
	copy(buf[8:], value)
	return buf
}

func decodeExpiration(value []byte) (time.Time, []byte) {
	ts := binary.BigEndian.Uint64(value[:8])
	return time.Unix(int64(ts), 0), value[8:]
}
```

Garbage collection with `Walk`:

```go
import (
	"context"
	"fmt"
	"time"

	"go.opentelemetry.io/collector/extension/xextension/storage"
)

func collectExpired(ctx context.Context, client storage.Client) error {
	walker, ok := client.(storage.Walker)
	if !ok {
		return fmt.Errorf("storage client does not implement Walker")
	}

	now := time.Now()
	return walker.Walk(ctx, func(key string, value []byte) ([]*storage.Operation, error) {
		if len(value) < 8 {
			return nil, nil
		}
		expireAt, _ := decodeExpiration(value)
		if now.After(expireAt) {
			return []*storage.Operation{storage.DeleteOperation(key)}, nil
		}
		return nil, nil
	})
}
```

All `Delete` operations returned by the `WalkFunc` are collected during
the walk
and applied atomically once the iteration completes. If the callback
returns a
non-nil error (other than `SkipAll`), or if the walk encounters an
internal
error, no operations are applied.

<!--Describe the documentation added.-->
#### Documentation

The package's README.md has been updated. The new types have been fully
documented.


Closes
open-telemetry#15191
@swiatekm

Copy link
Copy Markdown
Contributor

Should we apply similar interface as we have for pcommon.Map

func (m Map) All() iter.Seq2[string, Value] {
?

You can easily implement that interface with the new one. Are you thinking of a QoL thing?

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

ready-to-merge Code review completed; ready to merge by maintainers

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Add Walker interface to storage extension for iterating over entries

8 participants