etcd

package
v1.6.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jun 22, 2026 License: BSD-2-Clause Imports: 14 Imported by: 0

Documentation

Overview

Package etcd provides an etcd implementation of the storage driver interface. It enables using etcd as a distributed key-value storage backend.

Index

Examples

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Client 

type Client interface {
	// Watch watches for changes on a key (using etcd's signature).
	Watch(ctx context.Context, key string, opts ...etcd.OpOption) etcd.WatchChan
	// Txn creates a new transaction.
	Txn(ctx context.Context) etcd.Txn
}

Client defines the minimal interface needed for etcd operations. It is compatible with etcdclientv3.Client.

This allows for easier testing and mock implementations.

type Driver 

type Driver struct {
	// contains filtered or unexported fields
}

Driver is an etcd implementation of the storage driver interface. It uses etcd as the underlying key-value storage backend.

func New 

func New(client Client) *Driver

New creates a new etcd driver instance using an existing etcd client. The client should be properly configured and connected to an etcd cluster.

NewLocker is only supported when client is a concrete *etcd.Client — the concurrency package needs the concrete type which the Client interface does not expose. Drivers built from a non-concrete Client return locker.ErrUnsupported from NewLocker.

func (Driver) Execute 

func (d Driver) Execute(
	ctx context.Context,
	predicates []predicate.Predicate,
	thenOps []operation.Operation,
	elseOps []operation.Operation,
) (tx.Response, error)

Execute executes a transactional operation with conditional logic. It processes predicates to determine whether to execute thenOps or elseOps.

Example (SimpleDelete)

ExampleDriver_Execute_simpleDelete demonstrates a simple Delete operation with the etcd driver. 

package main

import (
	"context"
	"fmt"
	"log"
	"testing"
	"time"

	etcdclient "go.etcd.io/etcd/client/v3"

	"github.com/tarantool/go-storage/driver/etcd"
	testingUtils "github.com/tarantool/go-storage/internal/testing"
	"github.com/tarantool/go-storage/operation"

	etcdtest "github.com/tarantool/go-storage/test_helpers/etcd"
)

const exampleTestDialTimeout = 5 * time.Second

// createEtcdDriver creates an etcd driver for examples using the integration framework.
// Returns driver and cleanup function.
func createEtcdDriver(tb testing.TB) (*etcd.Driver, func()) {
	tb.Helper()

	cluster := etcdtest.New(tb, etcdtest.ClusterConfig{Size: 1})

	endpoints := cluster.EndpointsGRPC()

	client, err := etcdclient.New(etcdclient.Config{
		Endpoints:   endpoints,
		DialTimeout: exampleTestDialTimeout,

		AutoSyncInterval:      0,
		DialKeepAliveTime:     0,
		DialKeepAliveTimeout:  0,
		MaxCallSendMsgSize:    0,
		MaxCallRecvMsgSize:    0,
		TLS:                   nil,
		Username:              "",
		Password:              "",
		RejectOldCluster:      false,
		DialOptions:           nil,
		Context:               nil,
		Logger:                nil,
		LogConfig:             nil,
		PermitWithoutStream:   false,
		MaxUnaryRetries:       0,
		BackoffWaitBetween:    0,
		BackoffJitterFraction: 0,
	})
	if err != nil {
		tb.Fatalf("Failed to create etcd client: %v", err)
	}

	tb.Cleanup(func() { _ = client.Close() })

	driver := etcd.New(client)

	return driver, func() {}
}

func main() {
	// Create a testing context for the example.
	t := testingUtils.NewT()
	defer t.Cleanups()

	ctx := context.Background()

	driver, cleanup := createEtcdDriver(t)
	defer cleanup()

	key := []byte("/config/app/version")

	// First store a value.
	_, _ = driver.Execute(ctx, nil, []operation.Operation{
		operation.Put(key, []byte("1.0.0")),
	}, nil)

	// Then delete it.
	_, err := driver.Execute(ctx, nil, []operation.Operation{
		operation.Delete(key),
	}, nil)
	if err != nil {
		log.Printf("Delete operation failed: %v", err)
		return
	}

	fmt.Println("Successfully deleted key:", string(key))

}

Output:
Successfully deleted key: /config/app/version
Example (SimpleGet)

ExampleDriver_Execute_simpleGet demonstrates a simple Get operation with the etcd driver. 

package main

import (
	"context"
	"fmt"
	"log"
	"testing"
	"time"

	etcdclient "go.etcd.io/etcd/client/v3"

	"github.com/tarantool/go-storage/driver/etcd"
	testingUtils "github.com/tarantool/go-storage/internal/testing"
	"github.com/tarantool/go-storage/operation"

	etcdtest "github.com/tarantool/go-storage/test_helpers/etcd"
)

const exampleTestDialTimeout = 5 * time.Second

// createEtcdDriver creates an etcd driver for examples using the integration framework.
// Returns driver and cleanup function.
func createEtcdDriver(tb testing.TB) (*etcd.Driver, func()) {
	tb.Helper()

	cluster := etcdtest.New(tb, etcdtest.ClusterConfig{Size: 1})

	endpoints := cluster.EndpointsGRPC()

	client, err := etcdclient.New(etcdclient.Config{
		Endpoints:   endpoints,
		DialTimeout: exampleTestDialTimeout,

		AutoSyncInterval:      0,
		DialKeepAliveTime:     0,
		DialKeepAliveTimeout:  0,
		MaxCallSendMsgSize:    0,
		MaxCallRecvMsgSize:    0,
		TLS:                   nil,
		Username:              "",
		Password:              "",
		RejectOldCluster:      false,
		DialOptions:           nil,
		Context:               nil,
		Logger:                nil,
		LogConfig:             nil,
		PermitWithoutStream:   false,
		MaxUnaryRetries:       0,
		BackoffWaitBetween:    0,
		BackoffJitterFraction: 0,
	})
	if err != nil {
		tb.Fatalf("Failed to create etcd client: %v", err)
	}

	tb.Cleanup(func() { _ = client.Close() })

	driver := etcd.New(client)

	return driver, func() {}
}

func main() {
	// Create a testing context for the example.
	t := testingUtils.NewT()
	defer t.Cleanups()

	ctx := context.Background()

	driver, cleanup := createEtcdDriver(t)
	defer cleanup()

	key := []byte("/config/app/version")

	// First store a value.
	_, _ = driver.Execute(ctx, nil, []operation.Operation{
		operation.Put(key, []byte("1.0.0")),
	}, nil)

	// Then retrieve it.
	response, err := driver.Execute(ctx, nil, []operation.Operation{
		operation.Get(key),
	}, nil)
	if err != nil {
		log.Printf("Get operation failed: %v", err)
		return
	}

	if response.Succeeded && len(response.Results) > 0 {
		if len(response.Results[0].Values) > 0 {
			kv := response.Results[0].Values[0]
			fmt.Printf("Retrieved key: %s, value: %s, version: %d\n",
				string(kv.Key), string(kv.Value), kv.ModRevision)
		}
	}

}

Output:
Retrieved key: /config/app/version, value: 1.0.0, version: 2
Example (SimpleMultiPut)

ExampleDriver_Execute_simpleMultiPut demonstrates multiple Put operations in a single transaction with the etcd driver. 

package main

import (
	"context"
	"fmt"
	"log"
	"testing"
	"time"

	etcdclient "go.etcd.io/etcd/client/v3"

	"github.com/tarantool/go-storage/driver/etcd"
	testingUtils "github.com/tarantool/go-storage/internal/testing"
	"github.com/tarantool/go-storage/operation"

	etcdtest "github.com/tarantool/go-storage/test_helpers/etcd"
)

const exampleTestDialTimeout = 5 * time.Second

// createEtcdDriver creates an etcd driver for examples using the integration framework.
// Returns driver and cleanup function.
func createEtcdDriver(tb testing.TB) (*etcd.Driver, func()) {
	tb.Helper()

	cluster := etcdtest.New(tb, etcdtest.ClusterConfig{Size: 1})

	endpoints := cluster.EndpointsGRPC()

	client, err := etcdclient.New(etcdclient.Config{
		Endpoints:   endpoints,
		DialTimeout: exampleTestDialTimeout,

		AutoSyncInterval:      0,
		DialKeepAliveTime:     0,
		DialKeepAliveTimeout:  0,
		MaxCallSendMsgSize:    0,
		MaxCallRecvMsgSize:    0,
		TLS:                   nil,
		Username:              "",
		Password:              "",
		RejectOldCluster:      false,
		DialOptions:           nil,
		Context:               nil,
		Logger:                nil,
		LogConfig:             nil,
		PermitWithoutStream:   false,
		MaxUnaryRetries:       0,
		BackoffWaitBetween:    0,
		BackoffJitterFraction: 0,
	})
	if err != nil {
		tb.Fatalf("Failed to create etcd client: %v", err)
	}

	tb.Cleanup(func() { _ = client.Close() })

	driver := etcd.New(client)

	return driver, func() {}
}

func main() {
	// Create a testing context for the example.
	t := testingUtils.NewT()
	defer t.Cleanups()

	ctx := context.Background()

	driver, cleanup := createEtcdDriver(t)
	defer cleanup()

	response, err := driver.Execute(ctx, nil, []operation.Operation{
		operation.Put([]byte("/config/app/name"), []byte("MyApp")),
		operation.Put([]byte("/config/app/environment"), []byte("production")),
	}, nil)
	if err != nil {
		log.Printf("Multi-put operation failed: %v", err)
		return
	}

	fmt.Println("Successfully stored", len(response.Results), "configuration items")

}

Output:
Successfully stored 2 configuration items
Example (SimplePut)

ExampleDriver_Execute_simplePut demonstrates a simple Put operation with the etcd driver. 

package main

import (
	"context"
	"fmt"
	"log"
	"testing"
	"time"

	etcdclient "go.etcd.io/etcd/client/v3"

	"github.com/tarantool/go-storage/driver/etcd"
	testingUtils "github.com/tarantool/go-storage/internal/testing"
	"github.com/tarantool/go-storage/operation"

	etcdtest "github.com/tarantool/go-storage/test_helpers/etcd"
)

const exampleTestDialTimeout = 5 * time.Second

// createEtcdDriver creates an etcd driver for examples using the integration framework.
// Returns driver and cleanup function.
func createEtcdDriver(tb testing.TB) (*etcd.Driver, func()) {
	tb.Helper()

	cluster := etcdtest.New(tb, etcdtest.ClusterConfig{Size: 1})

	endpoints := cluster.EndpointsGRPC()

	client, err := etcdclient.New(etcdclient.Config{
		Endpoints:   endpoints,
		DialTimeout: exampleTestDialTimeout,

		AutoSyncInterval:      0,
		DialKeepAliveTime:     0,
		DialKeepAliveTimeout:  0,
		MaxCallSendMsgSize:    0,
		MaxCallRecvMsgSize:    0,
		TLS:                   nil,
		Username:              "",
		Password:              "",
		RejectOldCluster:      false,
		DialOptions:           nil,
		Context:               nil,
		Logger:                nil,
		LogConfig:             nil,
		PermitWithoutStream:   false,
		MaxUnaryRetries:       0,
		BackoffWaitBetween:    0,
		BackoffJitterFraction: 0,
	})
	if err != nil {
		tb.Fatalf("Failed to create etcd client: %v", err)
	}

	tb.Cleanup(func() { _ = client.Close() })

	driver := etcd.New(client)

	return driver, func() {}
}

func main() {
	// Create a testing context for the example.
	t := testingUtils.NewT()
	defer t.Cleanups()

	ctx := context.Background()

	driver, cleanup := createEtcdDriver(t)
	defer cleanup()

	key := []byte("/config/app/version")
	value := []byte("1.0.0")

	_, err := driver.Execute(ctx, nil, []operation.Operation{
		operation.Put(key, value),
	}, nil)
	if err != nil {
		log.Printf("Put operation failed: %v", err)
		return
	}

	fmt.Println("Key", string(key), "stored with value:", string(value))

}

Output:
Key /config/app/version stored with value: 1.0.0
Example (With_predicates_multiplePredicates)

ExampleDriver_Execute_with_predicates_multiplePredicates demonstrates multiple predicates with Else operations. 

package main

import (
	"context"
	"fmt"
	"log"
	"testing"
	"time"

	etcdclient "go.etcd.io/etcd/client/v3"

	"github.com/tarantool/go-storage/driver/etcd"
	testingUtils "github.com/tarantool/go-storage/internal/testing"
	"github.com/tarantool/go-storage/operation"
	"github.com/tarantool/go-storage/predicate"

	etcdtest "github.com/tarantool/go-storage/test_helpers/etcd"
)

const exampleTestDialTimeout = 5 * time.Second

// createEtcdDriver creates an etcd driver for examples using the integration framework.
// Returns driver and cleanup function.
func createEtcdDriver(tb testing.TB) (*etcd.Driver, func()) {
	tb.Helper()

	cluster := etcdtest.New(tb, etcdtest.ClusterConfig{Size: 1})

	endpoints := cluster.EndpointsGRPC()

	client, err := etcdclient.New(etcdclient.Config{
		Endpoints:   endpoints,
		DialTimeout: exampleTestDialTimeout,

		AutoSyncInterval:      0,
		DialKeepAliveTime:     0,
		DialKeepAliveTimeout:  0,
		MaxCallSendMsgSize:    0,
		MaxCallRecvMsgSize:    0,
		TLS:                   nil,
		Username:              "",
		Password:              "",
		RejectOldCluster:      false,
		DialOptions:           nil,
		Context:               nil,
		Logger:                nil,
		LogConfig:             nil,
		PermitWithoutStream:   false,
		MaxUnaryRetries:       0,
		BackoffWaitBetween:    0,
		BackoffJitterFraction: 0,
	})
	if err != nil {
		tb.Fatalf("Failed to create etcd client: %v", err)
	}

	tb.Cleanup(func() { _ = client.Close() })

	driver := etcd.New(client)

	return driver, func() {}
}

func main() {
	// Create a testing context for the example.
	t := testingUtils.NewT()
	defer t.Cleanups()

	ctx := context.Background()

	driver, cleanup := createEtcdDriver(t)
	defer cleanup()

	key1 := []byte("/config/database/host")
	key2 := []byte("/config/database/port")

	_, _ = driver.Execute(ctx, nil, []operation.Operation{
		operation.Put(key1, []byte("localhost")),
		operation.Put(key2, []byte("5432")),
	}, nil)

	response, err := driver.Execute(ctx, []predicate.Predicate{
		predicate.ValueEqual(key1, []byte("localhost")),
		predicate.ValueEqual(key2, []byte("5432")),
	}, []operation.Operation{
		operation.Put(key1, []byte("new-host")),
		operation.Put(key2, []byte("6432")),
	}, []operation.Operation{
		operation.Delete(key1),
		operation.Delete(key2),
	})
	if err != nil {
		log.Printf("Multi-predicate transaction failed: %v", err)
		return
	}

	if response.Succeeded {
		fmt.Println("Multi-predicate transaction succeeded - values were updated")
	} else {
		fmt.Println("Multi-predicate transaction failed - cleanup operations executed")
	}

}

Output:
Multi-predicate transaction succeeded - values were updated
Example (With_predicates_valueBased)

ExampleDriver_Execute_with_predicates_valueBased demonstrates value-based conditional updates using predicates. 

package main

import (
	"context"
	"fmt"
	"log"
	"testing"
	"time"

	etcdclient "go.etcd.io/etcd/client/v3"

	"github.com/tarantool/go-storage/driver/etcd"
	testingUtils "github.com/tarantool/go-storage/internal/testing"
	"github.com/tarantool/go-storage/operation"
	"github.com/tarantool/go-storage/predicate"

	etcdtest "github.com/tarantool/go-storage/test_helpers/etcd"
)

const exampleTestDialTimeout = 5 * time.Second

// createEtcdDriver creates an etcd driver for examples using the integration framework.
// Returns driver and cleanup function.
func createEtcdDriver(tb testing.TB) (*etcd.Driver, func()) {
	tb.Helper()

	cluster := etcdtest.New(tb, etcdtest.ClusterConfig{Size: 1})

	endpoints := cluster.EndpointsGRPC()

	client, err := etcdclient.New(etcdclient.Config{
		Endpoints:   endpoints,
		DialTimeout: exampleTestDialTimeout,

		AutoSyncInterval:      0,
		DialKeepAliveTime:     0,
		DialKeepAliveTimeout:  0,
		MaxCallSendMsgSize:    0,
		MaxCallRecvMsgSize:    0,
		TLS:                   nil,
		Username:              "",
		Password:              "",
		RejectOldCluster:      false,
		DialOptions:           nil,
		Context:               nil,
		Logger:                nil,
		LogConfig:             nil,
		PermitWithoutStream:   false,
		MaxUnaryRetries:       0,
		BackoffWaitBetween:    0,
		BackoffJitterFraction: 0,
	})
	if err != nil {
		tb.Fatalf("Failed to create etcd client: %v", err)
	}

	tb.Cleanup(func() { _ = client.Close() })

	driver := etcd.New(client)

	return driver, func() {}
}

func main() {
	// Create a testing context for the example.
	t := testingUtils.NewT()
	defer t.Cleanups()

	ctx := context.Background()

	driver, cleanup := createEtcdDriver(t)
	defer cleanup()

	key := []byte("/config/app/settings")
	currentValue := []byte("old-settings")
	newValue := []byte("new-settings")

	_, _ = driver.Execute(ctx, nil, []operation.Operation{
		operation.Put(key, currentValue),
	}, nil)

	response, err := driver.Execute(ctx, []predicate.Predicate{
		predicate.ValueEqual(key, []byte("old-settings")),
	}, []operation.Operation{
		operation.Put(key, newValue),
	}, nil)
	if err != nil {
		log.Printf("Conditional update failed: %v", err)
		return
	}

	if response.Succeeded {
		fmt.Println("Conditional update succeeded - value was updated")
	} else {
		fmt.Println("Conditional update failed - value did not match")
	}

}

Output:
Conditional update succeeded - value was updated
Example (With_predicates_versionBased)

ExampleDriver_Execute_with_predicates_versionBased demonstrates version-based conditional updates using predicates. 

package main

import (
	"context"
	"fmt"
	"log"
	"testing"
	"time"

	etcdclient "go.etcd.io/etcd/client/v3"

	"github.com/tarantool/go-storage/driver/etcd"
	testingUtils "github.com/tarantool/go-storage/internal/testing"
	"github.com/tarantool/go-storage/operation"
	"github.com/tarantool/go-storage/predicate"

	etcdtest "github.com/tarantool/go-storage/test_helpers/etcd"
)

const exampleTestDialTimeout = 5 * time.Second

// createEtcdDriver creates an etcd driver for examples using the integration framework.
// Returns driver and cleanup function.
func createEtcdDriver(tb testing.TB) (*etcd.Driver, func()) {
	tb.Helper()

	cluster := etcdtest.New(tb, etcdtest.ClusterConfig{Size: 1})

	endpoints := cluster.EndpointsGRPC()

	client, err := etcdclient.New(etcdclient.Config{
		Endpoints:   endpoints,
		DialTimeout: exampleTestDialTimeout,

		AutoSyncInterval:      0,
		DialKeepAliveTime:     0,
		DialKeepAliveTimeout:  0,
		MaxCallSendMsgSize:    0,
		MaxCallRecvMsgSize:    0,
		TLS:                   nil,
		Username:              "",
		Password:              "",
		RejectOldCluster:      false,
		DialOptions:           nil,
		Context:               nil,
		Logger:                nil,
		LogConfig:             nil,
		PermitWithoutStream:   false,
		MaxUnaryRetries:       0,
		BackoffWaitBetween:    0,
		BackoffJitterFraction: 0,
	})
	if err != nil {
		tb.Fatalf("Failed to create etcd client: %v", err)
	}

	tb.Cleanup(func() { _ = client.Close() })

	driver := etcd.New(client)

	return driver, func() {}
}

func main() {
	// Create a testing context for the example.
	t := testingUtils.NewT()
	defer t.Cleanups()

	ctx := context.Background()

	driver, cleanup := createEtcdDriver(t)
	defer cleanup()

	key := []byte("/config/app/feature")
	value := []byte("enabled")

	_, err := driver.Execute(ctx, nil, []operation.Operation{
		operation.Put(key, value),
	}, nil)
	if err != nil {
		log.Printf("Initial update failed: %v", err)
		return
	}

	getResponse, err := driver.Execute(ctx, nil, []operation.Operation{
		operation.Get(key),
	}, nil)
	if err != nil {
		log.Printf("Get operation failed: %v", err)
		return
	}

	var currentVersion int64
	if len(getResponse.Results) > 0 && len(getResponse.Results[0].Values) > 0 {
		currentVersion = getResponse.Results[0].Values[0].ModRevision
	}

	response, err := driver.Execute(ctx, []predicate.Predicate{
		predicate.VersionEqual(key, currentVersion),
	}, []operation.Operation{
		operation.Put(key, []byte("disabled")),
	}, nil)
	if err != nil {
		log.Printf("Version-based update failed: %v", err)
		return
	}

	if response.Succeeded {
		fmt.Println("Version-based update succeeded - no concurrent modification")
	} else {
		fmt.Println("Version-based update failed - version conflict detected")
	}

}

Output:
Version-based update succeeded - no concurrent modification

func (Driver) NewLocker added in v1.6.0 

func (d Driver) NewLocker(ctx context.Context, name string, opts ...locker.Option) (locker.Locker, error)

NewLocker returns locker.ErrUnsupported when the Driver was built with a Client that is not a concrete *etcd.Client — the concurrency package needs the concrete type which the Client interface does not expose.

Example

ExampleDriver_NewLocker demonstrates acquiring an etcd-backed distributed lock, observing contention from a second Locker on the same name, and releasing the lock so the contender can proceed.

The driver must be built with a concrete *etcd.Client — drivers built with a non-concrete Client return locker.ErrUnsupported from NewLocker. 

package main

import (
	"context"
	"errors"
	"fmt"
	"log"
	"testing"
	"time"

	etcdclient "go.etcd.io/etcd/client/v3"

	"github.com/tarantool/go-storage/driver/etcd"
	testingUtils "github.com/tarantool/go-storage/internal/testing"
	"github.com/tarantool/go-storage/locker"

	etcdtest "github.com/tarantool/go-storage/test_helpers/etcd"
)

const exampleTestDialTimeout = 5 * time.Second

// createEtcdDriver creates an etcd driver for examples using the integration framework.
// Returns driver and cleanup function.
func createEtcdDriver(tb testing.TB) (*etcd.Driver, func()) {
	tb.Helper()

	cluster := etcdtest.New(tb, etcdtest.ClusterConfig{Size: 1})

	endpoints := cluster.EndpointsGRPC()

	client, err := etcdclient.New(etcdclient.Config{
		Endpoints:   endpoints,
		DialTimeout: exampleTestDialTimeout,

		AutoSyncInterval:      0,
		DialKeepAliveTime:     0,
		DialKeepAliveTimeout:  0,
		MaxCallSendMsgSize:    0,
		MaxCallRecvMsgSize:    0,
		TLS:                   nil,
		Username:              "",
		Password:              "",
		RejectOldCluster:      false,
		DialOptions:           nil,
		Context:               nil,
		Logger:                nil,
		LogConfig:             nil,
		PermitWithoutStream:   false,
		MaxUnaryRetries:       0,
		BackoffWaitBetween:    0,
		BackoffJitterFraction: 0,
	})
	if err != nil {
		tb.Fatalf("Failed to create etcd client: %v", err)
	}

	tb.Cleanup(func() { _ = client.Close() })

	driver := etcd.New(client)

	return driver, func() {}
}

func main() {
	t := testingUtils.NewT()
	defer t.Cleanups()

	ctx := context.Background()

	driver, cleanup := createEtcdDriver(t)
	defer cleanup()

	holder, err := driver.NewLocker(ctx, "/locks/leader")
	if err != nil {
		log.Printf("NewLocker (holder) failed: %v", err)
		return
	}

	err = holder.Lock(ctx)
	if err != nil {
		log.Printf("Lock failed: %v", err)
		return
	}

	fmt.Println("holder acquired")

	contender, err := driver.NewLocker(ctx, "/locks/leader")
	if err != nil {
		log.Printf("NewLocker (contender) failed: %v", err)
		return
	}

	err = contender.TryLock(ctx)
	if errors.Is(err, locker.ErrLocked) {
		fmt.Println("contender saw the lock as held")
	}

	err = holder.Unlock(ctx)
	if err != nil {
		log.Printf("Unlock failed: %v", err)
		return
	}

	fmt.Println("holder released")

	err = contender.TryLock(ctx)
	if err != nil {
		log.Printf("contender TryLock after release failed: %v", err)
		return
	}

	fmt.Println("contender acquired")

	err = contender.Unlock(ctx)
	if err != nil {
		log.Printf("Unlock failed: %v", err)
		return
	}

}

Output:
holder acquired
contender saw the lock as held
holder released
contender acquired

func (Driver) Watch 

func (d Driver) Watch(ctx context.Context, key []byte, _ ...watch.Option) (<-chan watch.Event, func(), error)

Watch monitors changes to a specific key and returns a stream of events. event.Prefix is the watched key with any trailing "/" stripped.

Example (Basic)

ExampleDriver_Watch_basic demonstrates basic watch on a single key. 

package main

import (
	"context"
	"fmt"
	"log"
	"testing"
	"time"

	etcdclient "go.etcd.io/etcd/client/v3"

	"github.com/tarantool/go-storage/driver/etcd"
	testingUtils "github.com/tarantool/go-storage/internal/testing"
	"github.com/tarantool/go-storage/operation"

	etcdtest "github.com/tarantool/go-storage/test_helpers/etcd"
)

const exampleTestDialTimeout = 5 * time.Second

// createEtcdDriver creates an etcd driver for examples using the integration framework.
// Returns driver and cleanup function.
func createEtcdDriver(tb testing.TB) (*etcd.Driver, func()) {
	tb.Helper()

	cluster := etcdtest.New(tb, etcdtest.ClusterConfig{Size: 1})

	endpoints := cluster.EndpointsGRPC()

	client, err := etcdclient.New(etcdclient.Config{
		Endpoints:   endpoints,
		DialTimeout: exampleTestDialTimeout,

		AutoSyncInterval:      0,
		DialKeepAliveTime:     0,
		DialKeepAliveTimeout:  0,
		MaxCallSendMsgSize:    0,
		MaxCallRecvMsgSize:    0,
		TLS:                   nil,
		Username:              "",
		Password:              "",
		RejectOldCluster:      false,
		DialOptions:           nil,
		Context:               nil,
		Logger:                nil,
		LogConfig:             nil,
		PermitWithoutStream:   false,
		MaxUnaryRetries:       0,
		BackoffWaitBetween:    0,
		BackoffJitterFraction: 0,
	})
	if err != nil {
		tb.Fatalf("Failed to create etcd client: %v", err)
	}

	tb.Cleanup(func() { _ = client.Close() })

	driver := etcd.New(client)

	return driver, func() {}
}

func main() {
	// Create a testing context for the example.
	t := testingUtils.NewT()
	defer t.Cleanups()

	ctx := context.Background()

	driver, cleanup := createEtcdDriver(t)
	defer cleanup()

	key := []byte("/config/app/status")

	watchCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
	defer cancel()

	eventCh, stopWatch, err := driver.Watch(watchCtx, key)
	if err != nil {
		log.Printf("Failed to start watch: %v", err)
		return
	}
	defer stopWatch()

	fmt.Println("Watching for changes on:", string(key))

	go func() {
		time.Sleep(100 * time.Millisecond)

		_, _ = driver.Execute(ctx, nil, []operation.Operation{
			operation.Put(key, []byte("running")),
		}, nil)
	}()

	select {
	case event := <-eventCh:
		fmt.Printf("Received watch event for key: %s\n", string(event.Prefix))
	case <-watchCtx.Done():
		fmt.Println("Watch context expired")
	}

}

Output:
Watching for changes on: /config/app/status
Received watch event for key: /config/app/status
Example (GracefulTermination)

ExampleDriver_Watch_gracefulTermination demonstrates graceful watch termination with manual control. 

package main

import (
	"context"
	"fmt"
	"log"
	"testing"
	"time"

	etcdclient "go.etcd.io/etcd/client/v3"

	"github.com/tarantool/go-storage/driver/etcd"
	testingUtils "github.com/tarantool/go-storage/internal/testing"

	etcdtest "github.com/tarantool/go-storage/test_helpers/etcd"
)

const exampleTestDialTimeout = 5 * time.Second

// createEtcdDriver creates an etcd driver for examples using the integration framework.
// Returns driver and cleanup function.
func createEtcdDriver(tb testing.TB) (*etcd.Driver, func()) {
	tb.Helper()

	cluster := etcdtest.New(tb, etcdtest.ClusterConfig{Size: 1})

	endpoints := cluster.EndpointsGRPC()

	client, err := etcdclient.New(etcdclient.Config{
		Endpoints:   endpoints,
		DialTimeout: exampleTestDialTimeout,

		AutoSyncInterval:      0,
		DialKeepAliveTime:     0,
		DialKeepAliveTimeout:  0,
		MaxCallSendMsgSize:    0,
		MaxCallRecvMsgSize:    0,
		TLS:                   nil,
		Username:              "",
		Password:              "",
		RejectOldCluster:      false,
		DialOptions:           nil,
		Context:               nil,
		Logger:                nil,
		LogConfig:             nil,
		PermitWithoutStream:   false,
		MaxUnaryRetries:       0,
		BackoffWaitBetween:    0,
		BackoffJitterFraction: 0,
	})
	if err != nil {
		tb.Fatalf("Failed to create etcd client: %v", err)
	}

	tb.Cleanup(func() { _ = client.Close() })

	driver := etcd.New(client)

	return driver, func() {}
}

func main() {
	// Create a testing context for the example.
	t := testingUtils.NewT()
	defer t.Cleanups()

	ctx := context.Background()

	driver, cleanup := createEtcdDriver(t)
	defer cleanup()

	key := []byte("/config/monitoring/metrics")

	watchCtx, cancel := context.WithCancel(ctx)

	eventCh, stopWatch, err := driver.Watch(watchCtx, key)
	if err != nil {
		log.Printf("Failed to start watch: %v", err)
		cancel()

		return
	}

	fmt.Println("Started watch with manual control")

	go func() {
		time.Sleep(100 * time.Millisecond)
		fmt.Println("Stopping watch gracefully...")
		stopWatch()
		cancel()
	}()

	for {
		select {
		case event, ok := <-eventCh:
			if !ok {
				return
			}

			fmt.Printf("Received event: %s\n", string(event.Prefix))
		case <-watchCtx.Done():
			return
		}
	}

}

Output:
Started watch with manual control
Stopping watch gracefully...
Example (MultipleOperations)

ExampleDriver_Watch_multipleOperations demonstrates watching for multiple operations on a key prefix. 

package main

import (
	"context"
	"fmt"
	"log"
	"testing"
	"time"

	etcdclient "go.etcd.io/etcd/client/v3"

	"github.com/tarantool/go-storage/driver/etcd"
	testingUtils "github.com/tarantool/go-storage/internal/testing"
	"github.com/tarantool/go-storage/operation"

	etcdtest "github.com/tarantool/go-storage/test_helpers/etcd"
)

const exampleTestDialTimeout = 5 * time.Second

// createEtcdDriver creates an etcd driver for examples using the integration framework.
// Returns driver and cleanup function.
func createEtcdDriver(tb testing.TB) (*etcd.Driver, func()) {
	tb.Helper()

	cluster := etcdtest.New(tb, etcdtest.ClusterConfig{Size: 1})

	endpoints := cluster.EndpointsGRPC()

	client, err := etcdclient.New(etcdclient.Config{
		Endpoints:   endpoints,
		DialTimeout: exampleTestDialTimeout,

		AutoSyncInterval:      0,
		DialKeepAliveTime:     0,
		DialKeepAliveTimeout:  0,
		MaxCallSendMsgSize:    0,
		MaxCallRecvMsgSize:    0,
		TLS:                   nil,
		Username:              "",
		Password:              "",
		RejectOldCluster:      false,
		DialOptions:           nil,
		Context:               nil,
		Logger:                nil,
		LogConfig:             nil,
		PermitWithoutStream:   false,
		MaxUnaryRetries:       0,
		BackoffWaitBetween:    0,
		BackoffJitterFraction: 0,
	})
	if err != nil {
		tb.Fatalf("Failed to create etcd client: %v", err)
	}

	tb.Cleanup(func() { _ = client.Close() })

	driver := etcd.New(client)

	return driver, func() {}
}

func main() {
	// Create a testing context for the example.
	t := testingUtils.NewT()
	defer t.Cleanups()

	ctx := context.Background()

	driver, cleanup := createEtcdDriver(t)
	defer cleanup()

	key := []byte("/config/database/")

	watchCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
	defer cancel()

	eventCh, stopWatch, err := driver.Watch(watchCtx, key)
	if err != nil {
		log.Printf("Failed to start watch: %v", err)
		return
	}
	defer stopWatch()

	fmt.Println("Watching for changes on prefix:", string(key))

	go func() {
		_, _ = driver.Execute(ctx, nil, []operation.Operation{
			operation.Put([]byte("/config/database/host"), []byte("db1")),
		}, nil)

		_, _ = driver.Execute(ctx, nil, []operation.Operation{
			operation.Put([]byte("/config/database/port"), []byte("5432")),
		}, nil)

		_, _ = driver.Execute(ctx, nil, []operation.Operation{
			operation.Delete([]byte("/config/database/host")),
		}, nil)
	}()

	eventCount := 0
	for eventCount < 3 {
		select {
		case event := <-eventCh:
			fmt.Printf("Event %d: change detected on %s\n", eventCount+1, string(event.Prefix))

			eventCount++
		case <-watchCtx.Done():
			fmt.Println("Watch context expired")
			return
		}
	}

}

Output:
Watching for changes on prefix: /config/database/
Event 1: change detected on /config/database
Event 2: change detected on /config/database
Event 3: change detected on /config/database

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.