storage

README

go-storage: library to manage centralized configuration storages

About

go-storage is a Go library that provides a uniform interface for managing centralized configuration storages, supporting multiple backends like etcd and Tarantool Config Storage (TCS). It offers transactional operations, conditional predicates, real-time watch, and data integrity features.

Overview

The library abstracts the complexities of different storage backends, providing a consistent API for configuration management. It is designed for distributed systems where configuration consistency, real-time updates, and transactional safety are critical.

Features

• Unified Storage Interface: Single API for multiple backend drivers (etcd, TCS)
• Transactional Operations: Atomic transactions with conditional predicates
• Real-time Watch: Monitor changes to keys and prefixes
• Conditional Execution: Value and version-based predicates for safe updates
• Data Integrity: Built-in signing and verification of stored data
• Schema-Driven Integrity API: Codec[T] / Store[T] for type-safe values plus multi-key atomic transactions via Tx
• Namespace Scoping: Prefixed wrapper that scopes every operation under a key prefix
• Key‑Value Operations: Get, Put, Delete with prefix support
• Range Queries: Efficient scanning of keys with filters
• Extensible Drivers: Easy to add new storage backends

Installation

go get github.com/tarantool/go-storage

Quick Start

Using etcd Driver

package main

import (
    "context"
    "log"

    "go.etcd.io/etcd/client/v3"
    "github.com/tarantool/go-storage/driver/etcd"
    "github.com/tarantool/go-storage/operation"
)

func main() {
    // Connect to etcd.
    cli, err := clientv3.New(clientv3.Config{
        Endpoints: []string{"localhost:2379"},
    })
    if err != nil {
        log.Fatal(err)
    }
    defer cli.Close()

    // Create etcd driver.
    driver := etcd.New(cli)

    // Execute a simple Put operation.
    ctx := context.Background()
    _, err = driver.Execute(ctx, nil, []operation.Operation{
        operation.Put([]byte("/config/app/version"), []byte("1.0.0")),
    }, nil)
    if err != nil {
        log.Fatal(err)
    }
}

Using TCS Driver

package main

import (
    "context"
    "log"

    "github.com/tarantool/go-tarantool/v2"
    "github.com/tarantool/go-storage/driver/tcs"
    "github.com/tarantool/go-storage/operation"
)

func main() {
    // Connect to Tarantool.
    conn, err := tarantool.Connect("localhost:3301", tarantool.Opts{})
    if err != nil {
        log.Fatal(err)
    }
    defer conn.Close()

    // Create TCS driver.
    driver := tcs.New(conn)

    // Execute a transaction.
    ctx := context.Background()
    resp, err := driver.Execute(ctx, nil, []operation.Operation{
        operation.Put([]byte("/config/app/name"), []byte("MyApp")),
    }, nil)
    if err != nil {
        log.Fatal(err)
    }
    log.Printf("Transaction succeeded: %v", resp.Succeeded)
}

Drivers

etcd Driver

The driver/etcd package implements the storage driver interface for etcd. It supports all etcd features including conditional transactions, leases, and watch.

TCS Driver

The driver/tcs package provides a driver for Tarantool Config Storage (TCS), a distributed key‑value storage built on Tarantool. It offers high performance and strong consistency.

Connection Utilities

The connect package provides a simplified way to connect to storage backends using a unified configuration. It handles connection establishment, SSL/TLS setup, and authentication.

Note: Connecting to Tarantool Config Storage (TCS) with SSL requires the go_storage_ssl build tag. Without this tag, SSL support is disabled and attempting to connect with SSL.Enable = true will return ErrSSLDisabled.

Quick Start with Connect

package main

import (
    "context"
    "log"

    "github.com/tarantool/go-storage/connect"
)

func main() {
    ctx := context.Background()

    cfg := connect.Config{
        Endpoints: []string{"localhost:2379"},
        Username:  "user",
        Password:  "pass",
    }

    // Automatically tries etcd first, then TCS.
    stor, cleanup, err := connect.NewStorage(ctx, cfg)
    if err != nil {
        log.Fatal(err)
    }
    defer cleanup()

    // Use the storage...
    _ = stor
}

Explicit Backend Selection

// Connect to etcd specifically.
stor, cleanup, err := connect.NewEtcdStorage(ctx, cfg)

// Connect to TCS specifically.
stor, cleanup, err := connect.NewTCSStorage(ctx, cfg)

SSL/TLS Configuration

cfg := connect.Config{
    Endpoints: []string{"localhost:2379"},
    SSL: connect.SSLConfig{
        Enable:     true,
        CaFile:     "/path/to/ca.crt",
        CertFile:   "/path/to/client.crt",
        KeyFile:    "/path/to/client.key",
        VerifyPeer: true,
    },
}

API Overview

Storage Interface

The core Storage interface (storage.Storage) provides high‑level methods:

• Watch(ctx, key, opts) <-chan watch.Event – watch for changes
• Tx(ctx) tx.Tx – create a transaction builder
• Range(ctx, opts) ([]kv.KeyValue, error) – range query with prefix/limit

Namespace Scoping with Prefixed

storage.Prefixed(prefix, inner) returns a Storage that transparently prepends prefix to every operation, predicate, Range, and Watch call, and strips it back from any keys returned to the caller. Nested wrappers are flattened at construction (Prefixed("/a", Prefixed("/b", base)) is equivalent to Prefixed("/a/b", base)).

scoped := storage.Prefixed("/ns", storage.NewStorage(driver))

// Caller writes /cfg/version; the driver actually stores /ns/cfg/version.
_, err := scoped.Tx(ctx).Then(
    operation.Put([]byte("/cfg/version"), []byte("1.0.0")),
).Commit()

Transaction Builder

The tx.Tx interface enables conditional transactions:

resp, err := storage.Tx(ctx).
    If(predicate.ValueEqual(key, "old")).
    Then(operation.Put(key, "new")).
    Else(operation.Delete(key)).
    Commit()

Operations

The operation package defines Get, Put, Delete operations. Each operation can be configured with options.

Predicates

The predicate package provides value and version comparisons:

• ValueEqual, ValueNotEqual
• VersionEqual, VersionNotEqual, VersionGreater, VersionLess

Watch

The watch package delivers real‑time change events. Watch can be set on a single key or a prefix.

Data Integrity with Typed Storage

The integrity package provides a high‑level Typed interface for storing and retrieving values with built‑in integrity protection. It automatically computes hashes and signatures (using configurable algorithms) and verifies them on retrieval.

Creating a Typed Storage Instance

package main

import (
    "context"
    "crypto/rand"
    "crypto/rsa"
    "log"

    clientv3 "go.etcd.io/etcd/client/v3"
    "github.com/tarantool/go-storage"
    "github.com/tarantool/go-storage/driver/etcd"
    "github.com/tarantool/go-storage/hasher"
    "github.com/tarantool/go-storage/crypto"
    "github.com/tarantool/go-storage/integrity"
)

func main() {
    // 1. Create a base storage (e.g., etcd driver).
    cli, err := clientv3.New(clientv3.Config{Endpoints: []string{"localhost:2379"}})
    if err != nil {
        log.Fatal(err)
    }
    defer cli.Close()

    driver := etcd.New(cli)
    baseStorage := storage.NewStorage(driver)

    // 2. Generate RSA keys (in production, load from secure storage).
    privKey, err := rsa.GenerateKey(rand.Reader, 2048)
    if err != nil {
        log.Fatal(err)
    }

    // 3. Build typed storage with integrity protection.
    typed := integrity.NewTypedBuilder[MyConfig](baseStorage).
        WithPrefix("/config").
        WithHasher(hasher.NewSHA256Hasher()).          // adds SHA‑256 hash verification.
        WithSignerVerifier(crypto.NewRSAPSSSignerVerifier(*privKey)). // adds RSA‑PSS signatures.
        Build()

    ctx := context.Background()

    // 4. Store a configuration object with automatic integrity data.
    config := MyConfig{Environment: "production", Timeout: 30}
    if err := typed.Put(ctx, "app/settings", config); err != nil {
        log.Fatal(err)
    }

    // 4.5 Store an object using predicates.
    p, _ := typed.ValueEqual(config)

    if err := typed.Put(ctx,
        "app/settings",
        config,
        integrity.WithPutPredicates(p),
    ); err != nil {
        log.Fatal(err)
    }

    // 5. Retrieve and verify integrity.
    result, err := typed.Get(ctx, "app/settings")
    if err != nil {
        log.Fatal(err)
    }

    if result.Error != nil {
        log.Printf("Integrity check failed: %v", result.Error)
    } else {
        cfg, _ := result.Value.Get()
        log.Printf("Retrieved valid config: %+v", cfg)
    }

    // 6. Range over all configurations under a prefix.
    results, err := typed.Range(ctx, "app/")
    if err != nil {
        log.Fatal(err)
    }
    for _, res := range results {
        log.Printf("Found config %s (valid: %v)", res.Name, res.Error == nil)
    }
}

type MyConfig struct {
    Environment string `yaml:"environment"`
    Timeout     int    `yaml:"timeout"`
}

Key Features

• Automatic Hash & Signature Generation: Values are stored together with their hashes and/or signatures.
• Validation on read: Get and Range operations verify hashes and signatures; invalid data is reported.
• Configurable Algorithms: Plug in any hasher (hasher.Hasher) and signer/verifier (crypto.SignerVerifier).
• Prefix Isolation: Each typed storage uses a configurable key prefix, avoiding collisions.
• Watch Support: Watch method filters events for the typed namespace.

The integrity.Typed builder also accepts custom marshallers (default is YAML), custom namers, and separate signer/verifier instances for asymmetric setups.

Schema-Driven Integrity API (Codec, Store, Tx)

Alongside integrity.Typed, the package exposes a schema-first API split into three pieces:

• integrity.Codec[T] describes the on-disk layout (object location, hashers, signers, marshaller) without binding to any storage handle. It is built via the fluent CodecBuilder[T], which validates location-override keys eagerly so typos like WithHashLocation("sah256", …) fail at Build() instead of being silently ignored.
• integrity.Store[T] is a codec bound to a storage.Storage and exposes the familiar Get / Put / Delete / Range / Watch methods.
• integrity.Tx accumulates TxGet / TxPut / TxDelete / TxRange calls from one or more codecs and commits them atomically through a single storage call. Reads return typed futures (GetFuture[T], RangeFuture[T]) whose Result() is populated after Commit.

Codec and Store

codec, err := integrity.NewCodecBuilder[MyConfig]().
    WithObjectLocation("config").
    WithHasher(hasher.NewSHA256Hasher()).
    Build()
if err != nil {
    log.Fatal(err)
}

store := codec.Bind(baseStorage)

if err := store.Put(ctx, "app/settings", MyConfig{...}); err != nil {
    log.Fatal(err)
}

res, err := store.Get(ctx, "app/settings")
if err != nil {
    log.Fatal(err)
}
cfg := res.Value.Unwrap()

Multi-Key Transactions

Tx batches reads and writes — across multiple codecs if needed — into one atomic storage call. If predicates are routed by Then / Else; futures attached to the branch that did not fire return ErrBranchNotFired.

txn := integrity.NewTx(baseStorage)

pred, _ := codec.ValueEqual(MyConfig{...})
bound, _ := codec.BindPredicate("app/settings", pred)
txn.If(bound)

newFut := codec.TxGet(txn.Then(), "app/new-settings")
_ = codec.TxPut(txn.Then(), "app/settings", MyConfig{...})

resp, err := txn.Commit(ctx)
if err != nil {
    log.Fatal(err)
}
if !resp.Succeeded {
    // The Then branch did not fire; newFut.Result() returns
    // integrity.ErrBranchNotFired.
}

Layered Key Layout

The new API uses namer.LayeredNamer by default, which places each key category under its own top-level location segment:

/<objectLocation>/<name>                       (value)
/hash/<hashLocation>/<objectLocation>/<name>   (one per hasher)
/sig/<sigLocation>/<objectLocation>/<name>     (one per signer)

objectLocation may itself be a multi-segment path (e.g. "settings/ldap") — useful when the on-disk hierarchy of a feature is fixed at codec build time rather than per item. hashLocation and sigLocation are still single tokens because they index per-hasher / per-signer maps. The first segment of objectLocation must not equal the reserved markers hash or sig (they would collide with the parser's category dispatch).

namer.CompactSingleHash() and namer.CompactSingleSig() drop the per-hasher / per-signer segment when exactly one is configured. ParseKey parses a raw key back to (name, KeyType, property) unambiguously.

Singleton Store: One Fixed Key per Codec

For configuration objects that live at a single, known key (e.g. /settings/auth) — not under a directory of <objectLocation>/<id> items — Codec[T].BindSingleton(storage, name) returns a *SingletonStore[T] that bakes the name in once. All operations (Get / Put / Delete / Watch, plus TxGet / TxPut / TxDelete for multi-op transactions) drop the name parameter:

codec, err := integrity.NewCodecBuilder[AuthConfig]().
    WithObjectLocation("settings").
    WithSignerVerifier(crypto.NewRSAPSSSignerVerifier(*privKey)).
    Build()
if err != nil {
    log.Fatal(err)
}

auth, err := codec.BindSingleton(baseStorage, "auth")
if err != nil {
    log.Fatal(err)
}

// Wire layout:
//   /settings/auth                      (value)
//   /hash/<hashLoc>/settings/auth       (hash, per hasher)
//   /sig/<sigLoc>/settings/auth         (sig,  per signer)

if err := auth.Put(ctx, AuthConfig{Issuer: "example"}); err != nil {
    log.Fatal(err)
}

res, err := auth.Get(ctx)

The same Codec[T] can serve both shapes: codec.Bind(storage) for a directory of items keyed by name, codec.BindSingleton(storage, name) for a fixed singleton. Predicates from the codec (ValueEqual, VersionEqual, etc.) are name-agnostic and work as-is when passed through WithPutPredicates(...) / WithDeletePredicates(...). For multi-op transactions, auth.BindPredicate(pred) resolves the predicate to the singleton's value-layer key for use in Tx.If.

Marshallers

Beyond the default YAML marshaller, the marshaller package now ships:

• TypedJSONMarshaller[T] — encoding/json-based marshalling for any Go type.
• TypedBytesMarshaller — passthrough TypedMarshaller[[]byte] for values that are already serialized or stored as opaque blobs.

Examples

Comprehensive examples are available in the driver packages:

• etcd examples: driver/etcd/examples_test.go
• TCS examples: driver/tcs/examples_test.go

Run them with go test -v -run Example ./driver/etcd or ./driver/tcs.

Build Tags

The library supports the following build tags:

go_storage_ssl

Enables SSL/TLS support for Tarantool Config Storage connections. This tag requires the go-tlsdialer dependency.

# Build with SSL support for TCS
go build -tags go_storage_ssl ./...

Without this tag:

• SSL support for TCS is disabled
• Connecting to TCS with SSL.Enable = true returns ErrSSLDisabled
• The go-tlsdialer dependency and CGO is not required on build-time

Contributing

Contributions are welcome! Please see the CONTRIBUTING.md file for guidelines (if present) or open an issue to discuss your ideas.

License

This project is licensed under the BSD 2‑Clause License – see the LICENSE file for details.

Documentation

Overview

Package storage provides a uniform way to handle various types of centralized config storages for Tarantool.

See the github.com/tarantool/go-storage/integrity package for high-level typed storage with automatic hash and signature verification.

Index

• Variables
• type Option
  • func WithRetry() Option
  • func WithTimeout() Option
• type RangeOption
  • func WithLimit(limit int) RangeOption
  • func WithPrefix(prefix string) RangeOption
• type Storage
  • func NewStorage(driver driver.Driver, _ ...Option) Storage
  • func Prefixed(prefix string, inner Storage) (Storage, error)

Examples

• Prefixed
• Prefixed (Nested)
• Prefixed (Predicates)

Variables

var ErrPrefixTrailingSlash = errors.New("storage.Prefixed: prefix must not end with '/'")

ErrPrefixTrailingSlash is returned by Prefixed when the prefix ends with "/". The codec namer prepends "/" to every key, so a trailing slash here would produce keys like "/foo//objectLocation/name".

Types

type Option

type Option func(*storageOptions)

Option is a function that configures storage options.

func WithRetry

func WithRetry() Option

WithRetry configures retry behavior for failed operations. This is a dummy option for demonstration purposes.

func WithTimeout

func WithTimeout() Option

WithTimeout configures a default timeout for storage operations. This is a dummy option for demonstration purposes.

type RangeOption

type RangeOption func(*rangeOptions)

RangeOption is a function that configures range operation options.

func WithLimit

func WithLimit(limit int) RangeOption

WithLimit configures a range operation to limit the number of results returned.

func WithPrefix

func WithPrefix(prefix string) RangeOption

WithPrefix configures a range operation to filter keys by the specified prefix.

type Storage

type Storage interface {
	// Watch streams changes for a specific key or prefix.
	// Options:
	//   - WithPrefix: watch for changes on keys with the specified prefix
	Watch(ctx context.Context, key []byte, opts ...watch.Option) <-chan watch.Event

	// Tx creates a new transaction.
	// The context manages timeouts and cancellation for the transaction.
	Tx(ctx context.Context) txPkg.Tx

	// Range queries a range of keys with optional filtering.
	// Options:
	//   - WithPrefix: filter keys by prefix
	//   - WithLimit: limit the number of results returned
	Range(ctx context.Context, opts ...RangeOption) ([]kv.KeyValue, error)
}

Storage is the main interface for key-value storage operations. It provides methods for watching changes, transaction management, and range queries.

func NewStorage

func NewStorage(driver driver.Driver, _ ...Option) Storage

NewStorage creates a new Storage instance with the specified driver. Optional StorageOption parameters can be provided to configure the storage.

func Prefixed

func Prefixed(prefix string, inner Storage) (Storage, error)

Prefixed returns a Storage that scopes every operation, predicate, range, and watch under prefix. Keys returned to the caller (RequestResponse.Values, kv.KeyValue, watch.Event.Prefix) have prefix stripped — callers never see absolute keys.

Composition: Prefixed("/a", Prefixed("/b", base)) ≡ Prefixed("/a/b", base). Nested wrappers are flattened at construction so the outer prefix is the leftmost segment.

An empty prefix yields a transparent wrapper. A non-empty prefix must not end with "/" (returns ErrPrefixTrailingSlash) — the codec namer prepends "/" to every key, so a trailing slash here would produce keys like "/foo//objectLocation/name".

Example

ExamplePrefixed shows how Prefixed scopes every Tx operation under a namespace. Callers work with logical keys ("/cfg/version"); the underlying driver sees absolute keys ("/ns/cfg/version"), and the namespace is stripped from any keys returned to the caller.

package main

import (
	"context"
	"fmt"
	"log"

	storage "github.com/tarantool/go-storage"
	"github.com/tarantool/go-storage/driver/dummy"
	"github.com/tarantool/go-storage/operation"
)

func main() {
	ctx := context.Background()
	base := storage.NewStorage(dummy.New())

	scoped, err := storage.Prefixed("/ns", base)
	if err != nil {
		log.Fatalf("prefix: %v", err)
	}

	_, err = scoped.Tx(ctx).Then(
		operation.Put([]byte("/cfg/version"), []byte("1.0.0")),
	).Commit()
	if err != nil {
		log.Fatalf("put: %v", err)
	}

	resp, err := scoped.Tx(ctx).Then(
		operation.Get([]byte("/cfg/version")),
	).Commit()
	if err != nil {
		log.Fatalf("get: %v", err)
	}

	got := resp.Results[0].Values[0]
	fmt.Printf("logical key: %s, value: %s\n", got.Key, got.Value)

}

Output:
logical key: /cfg/version, value: 1.0.0

Example (Nested)

ExamplePrefixed_nested shows that nested wrappers are flattened at construction time: Prefixed("/a", Prefixed("/b", base)) is equivalent to Prefixed("/a/b", base). This is observable by reading the same key through the un-wrapped base storage.

package main

import (
	"context"
	"fmt"
	"log"

	storage "github.com/tarantool/go-storage"
	"github.com/tarantool/go-storage/driver/dummy"
	"github.com/tarantool/go-storage/operation"
)

func main() {
	ctx := context.Background()
	base := storage.NewStorage(dummy.New())

	inner, err := storage.Prefixed("/b", base)
	if err != nil {
		log.Fatalf("prefix /b: %v", err)
	}

	outer, err := storage.Prefixed("/a", inner)
	if err != nil {
		log.Fatalf("prefix /a: %v", err)
	}

	_, err = outer.Tx(ctx).Then(
		operation.Put([]byte("/k"), []byte("v")),
	).Commit()
	if err != nil {
		log.Fatalf("put: %v", err)
	}

	// Read directly from base to see the absolute key the driver actually stored.
	resp, err := base.Tx(ctx).Then(
		operation.Get([]byte("/a/b/k")),
	).Commit()
	if err != nil {
		log.Fatalf("get: %v", err)
	}

	got := resp.Results[0].Values[0]
	fmt.Printf("absolute: %s = %s\n", got.Key, got.Value)

}

Output:
absolute: /a/b/k = v

Example (Predicates)

ExamplePrefixed_predicates shows that predicates are also rewritten under the configured prefix, so conditional transactions stay scoped.

package main

import (
	"context"
	"fmt"
	"log"

	storage "github.com/tarantool/go-storage"
	"github.com/tarantool/go-storage/driver/dummy"
	"github.com/tarantool/go-storage/operation"
	"github.com/tarantool/go-storage/predicate"
)

func main() {
	ctx := context.Background()

	scoped, err := storage.Prefixed("/ns", storage.NewStorage(dummy.New()))
	if err != nil {
		log.Fatalf("prefix: %v", err)
	}

	_, err = scoped.Tx(ctx).Then(
		operation.Put([]byte("/feature"), []byte("on")),
	).Commit()
	if err != nil {
		log.Fatalf("seed: %v", err)
	}

	resp, err := scoped.Tx(ctx).
		If(predicate.ValueEqual([]byte("/feature"), "on")).
		Then(operation.Put([]byte("/feature"), []byte("off"))).
		Else(operation.Put([]byte("/feature"), []byte("unchanged"))).
		Commit()
	if err != nil {
		log.Fatalf("commit: %v", err)
	}

	fmt.Println("succeeded:", resp.Succeeded)

}

Output:
succeeded: true