mirror of
https://github.com/hanzoai/telemetry.git
synced 2026-10-30 20:01:49 +00:00
#### 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
140 lines
4.9 KiB
Go
140 lines
4.9 KiB
Go
// Copyright The OpenTelemetry Authors
|
|
// SPDX-License-Identifier: Apache-2.0
|
|
|
|
package storage // import "go.opentelemetry.io/collector/extension/xextension/storage"
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
|
|
"go.opentelemetry.io/collector/component"
|
|
"go.opentelemetry.io/collector/extension"
|
|
)
|
|
|
|
// Extension is the interface that storage extensions must implement
|
|
type Extension interface {
|
|
extension.Extension
|
|
|
|
// GetClient will create a client for use by the specified component.
|
|
// Each component can have multiple storages (e.g. one for each signal),
|
|
// which can be identified using storageName parameter.
|
|
// The component can use the client to manage state
|
|
GetClient(ctx context.Context, kind component.Kind, id component.ID, storageName string) (Client, error)
|
|
}
|
|
|
|
// Client is the interface that storage clients must implement
|
|
// All methods should return error only if a problem occurred.
|
|
// This mirrors the behavior of a golang map:
|
|
// - Set doesn't error if a key already exists - it just overwrites the value.
|
|
// - Get doesn't error if a key is not found - it just returns nil.
|
|
// - Delete doesn't error if the key doesn't exist - it just no-ops.
|
|
//
|
|
// Similarly:
|
|
// - Batch doesn't error if any of the above happens for either retrieved or updated keys
|
|
//
|
|
// This also provides a way to differentiate data operations
|
|
//
|
|
// [overwrite | not-found | no-op] from "real" problems
|
|
type Client interface {
|
|
// Get will retrieve data from storage that corresponds to the
|
|
// specified key. It should return (nil, nil) if not found
|
|
Get(ctx context.Context, key string) ([]byte, error)
|
|
|
|
// Set will store data. The data can be retrieved by the same
|
|
// component after a process restart, using the same key
|
|
Set(ctx context.Context, key string, value []byte) error
|
|
|
|
// Delete will delete data associated with the specified key
|
|
Delete(ctx context.Context, key string) error
|
|
|
|
// Batch handles specified operations in batch. Get operation results are put in-place
|
|
Batch(ctx context.Context, ops ...*Operation) error
|
|
|
|
// Close will release any resources held by the client
|
|
Close(ctx context.Context) error
|
|
}
|
|
|
|
// WalkFunc is the type of the function called by Walk to visit each entry in the storage.
|
|
//
|
|
// The key argument contains the key of the stored entry.
|
|
// Key order is not guaranteed and may vary between storage implementations.
|
|
//
|
|
// The value bytes are only valid for the duration of the
|
|
// function call and need to be copied if later access is needed.
|
|
//
|
|
// The function may return operations that are collected and applied in order
|
|
// either when the end of the Walk is reached or after SkipAll is returned.
|
|
// Returning a nil slice is valid and simply 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 the storage supports transactions, all returned operations must be
|
|
// applied in the same transaction the key/value entries for WalkFunc were obtained from.
|
|
//
|
|
// The error result returned by the function controls how Walk continues:
|
|
//
|
|
// If the function returns the special value SkipAll, Walk skips all remaining
|
|
// storage entries and all collected operations still get applied in order.
|
|
//
|
|
// Otherwise, if the function returns a non-nil error,
|
|
// Walk stops entirely and returns that error without applying
|
|
// any of the collected operations.
|
|
type WalkFunc func(key string, value []byte) ([]*Operation, error)
|
|
|
|
// 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
|
|
|
|
// Walker is the interface that a storage extension Client may implement
|
|
// in order to support ranging through the storage entries.
|
|
type Walker interface {
|
|
// Walk calls fn for every key/value pair in the storage.
|
|
//
|
|
// If fn returns a non-nil error other than SkipAll, or if Walk
|
|
// encounters an internal error, ranging stops immediately and no
|
|
// collected operations are applied.
|
|
Walk(ctx context.Context, fn WalkFunc) error
|
|
}
|
|
|
|
type OpType int
|
|
|
|
const (
|
|
Get OpType = iota
|
|
Set
|
|
Delete
|
|
)
|
|
|
|
type Operation struct {
|
|
// Key specifies key which is going to be get/set/deleted
|
|
Key string
|
|
// Value specifies value that is going to be set or holds result of get operation
|
|
Value []byte
|
|
// Type describes the operation type
|
|
Type OpType
|
|
}
|
|
|
|
func SetOperation(key string, value []byte) *Operation {
|
|
return &Operation{
|
|
Key: key,
|
|
Value: value,
|
|
Type: Set,
|
|
}
|
|
}
|
|
|
|
func GetOperation(key string) *Operation {
|
|
return &Operation{
|
|
Key: key,
|
|
Type: Get,
|
|
}
|
|
}
|
|
|
|
func DeleteOperation(key string) *Operation {
|
|
return &Operation{
|
|
Key: key,
|
|
Type: Delete,
|
|
}
|
|
}
|
|
|
|
var ErrStorageFull = errors.New("the storage extension has run out of available space")
|