DenisandGitHub
12dfcf61ba
#### 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 https://github.com/open-telemetry/opentelemetry-collector/issues/15191
Storage
Status: under development; This is currently just the interface
A storage extension persists state beyond the collector process. Other components can request a storage client from the storage extension and use it to manage state.
The storage.Extension interface extends component.Extension by adding the following method:
GetClient(context.Context, component.Kind, component.ID, string) (Client, error)
The storage.Client interface contains the following methods:
Get(context.Context, string) ([]byte, error)
Set(context.Context, string, []byte) error
Delete(context.Context, string) error
Close(context.Context) error
It is possible to execute several operations in a single transaction via Batch. The method takes a collection of
Operation arguments (each of which contains Key, Value and Type properties):
Batch(context.Context, ...Operation) error
The elements itself can be created using:
SetOperation(string, []byte) Operation
GetOperation(string) Operation
DeleteOperation(string) Operation
Get operation results are stored in-place into the given Operation and can be retrieved using its Value property.
Note: All methods should return error only if a problem occurred. (For example, if a file is no longer accessible, or if a remote service is unavailable.)
Note: It is the responsibility of each component to Close a storage client that it has requested.
Walking storage entries
A storage.Client may optionally implement the storage.Walker interface to support iterating over all key/value entries:
Walk(context.Context, WalkFunc) error
The WalkFunc callback is called for every key/value pair. Key order is not guaranteed.
The callback returns a slice of Operation and an error:
WalkFunc = func(key string, value []byte) ([]*Operation, error)
Operations returned by WalkFunc are collected during the walk and applied in order when the walk completes successfully or when SkipAll is returned. Returning a nil slice is valid and contributes no operations.
All operation types (Get, Set, Delete) are valid in the returned slice. Get operations work as usual — the result is stored in the Operation instance.
If WalkFunc returns the SkipAll error, the walk stops early and all collected operations are still applied. If WalkFunc returns any other non-nil error, or if the walk encounters an internal error, the walk stops and no collected operations are applied.