Component(s)

extension/x/storage

Is your feature request related to a problem? Please describe.

The current storage.Client interface supports point lookups (Get), writes (Set), deletes (Delete), and batch operations (Batch), but provides no way to iterate over all stored key/value entries. This makes it impossible to implement common storage management patterns without reaching outside the abstraction.

Components that persist state through the storage extension today have no portable way to:

• Discover what keys exist in the storage.
• Migrate stored data from one schema to another.
• Clean up stale or expired entries.
• Export or inspect stored state for debugging.

Each of these use cases requires enumerating entries, which forces implementers to either maintain a separate key index (error-prone and redundant) or depend on the concrete storage backend directly (breaking the abstraction).

Describe the solution you'd like

Introduce an optional Walker interface that a storage.Client may implement:

type WalkFunc func(key string, value []byte) ([]*Operation, error)

var SkipAll = errors.New("skip everything and stop the walk")

type Walker interface {
    Walk(ctx context.Context, fn WalkFunc) error
}

Design decisions

• Optional interface via type assertion. Walker is separate from Client so that existing storage extensions continue to compile without changes. Consumers check support with walker, ok := client.(storage.Walker).
• WalkFunc returns operations. Rather than mutating storage during iteration (which is unsafe for most backends), the callback returns []*Operation that are collected and applied in order after the walk completes (or after SkipAll). This keeps the read and write phases separate and lets implementations apply all mutations in a single batch or transaction.
• SkipAll for early termination. Follows the fs.SkipAll / filepath.SkipDir pattern from the standard library. When returned, the walk stops but collected operations are still applied.
• Error semantics. If the callback returns any error other than SkipAll, or if the walk itself encounters an internal error, iteration stops and no collected operations are applied.
• Transaction support. If the underlying storage supports transactions, all returned operations must be applied within the same transaction the entries were read from, ensuring a consistent snapshot.

Usage examples

1. TTL-based garbage collection

Values carry an expiration timestamp prefix. A periodic routine walks all entries and produces Delete operations for expired ones.

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 := time.Unix(int64(binary.BigEndian.Uint64(value[:8])), 0)
        if now.After(expireAt) {
            return []*storage.Operation{storage.DeleteOperation(key)}, nil
        }
        return nil, nil
    })
}

2. Schema migration

When the value encoding changes between versions, walk all entries and rewrite them in the new format.

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

    return walker.Walk(ctx, func(key string, value []byte) ([]*storage.Operation, error) {
        if !bytes.HasPrefix(value, []byte("v1:")) {
            return nil, nil
        }
        newValue, err := convertV1ToV2(value)
        if err != nil {
            return nil, fmt.Errorf("migrating key %q: %w", key, err)
        }
        return []*storage.Operation{storage.SetOperation(key, newValue)}, nil
    })
}

3. Storage size reporting

Count all entries and sum their sizes for monitoring or debugging.

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

    var count, totalBytes int
    err := walker.Walk(ctx, func(key string, value []byte) ([]*storage.Operation, error) {
        count++
        totalBytes += len(key) + len(value)
        return nil, nil
    })
    return count, totalBytes, err
}

4. Key prefix search with early termination

Find all entries matching a prefix and stop once a limit is reached.

func findByPrefix(ctx context.Context, client storage.Client, prefix string, limit int) (map[string][]byte, error) {
    walker, ok := client.(storage.Walker)
    if !ok {
        return nil, fmt.Errorf("storage client does not implement Walker")
    }

    results := make(map[string][]byte)
    err := walker.Walk(ctx, func(key string, value []byte) ([]*storage.Operation, error) {
        if strings.HasPrefix(key, prefix) {
            cp := make([]byte, len(value))
            copy(cp, value)
            results[key] = cp
            if len(results) >= limit {
                return nil, storage.SkipAll
            }
        }
        return nil, nil
    })
    if err != nil {
        return nil, err
    }
    return results, nil
}

5. Orphan cleanup after component reconfiguration

After a component's configuration changes (e.g. a queue is renamed or removed), delete entries that belong to the old namespace.

func cleanupOrphans(ctx context.Context, client storage.Client, activeKeys map[string]struct{}) error {
    walker, ok := client.(storage.Walker)
    if !ok {
        return fmt.Errorf("storage client does not implement Walker")
    }

    return walker.Walk(ctx, func(key string, value []byte) ([]*storage.Operation, error) {
        if _, active := activeKeys[key]; !active {
            return []*storage.Operation{storage.DeleteOperation(key)}, nil
        }
        return nil, nil
    })
}

References

• Proposed implementation: [xextension/storage] Add Walker interface for storage extensions #15190
• Walker implementation PoC for the File storage extension: [storage/filestorage] Implement storage.Walker in the file storage extension opentelemetry-collector-contrib#47755

Describe alternatives you've considered

Alternatively, Walk could become a part of the storage.Client interface and we could implement this function for all existing storage extensions – which would be a breaking change for anyone outside opentelemetry-collector-contrib.

Additional context

No response

Tip

_{React with 👍 to help prioritize this issue. Please use comments to provide useful context, avoiding +1 or me too, to help us triage it. Learn more here.}