discovery

package module
v1.2.0 Latest Latest
Warning

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

 Go to latest Published: May 5, 2026 License: BSD-2-Clause Imports: 4 Imported by: 0

README

discovery

Package discovery implements cluster discovery helpers for Tarantool 3.0 implemented in Go language according to the design document.

The main example demonstrates how to create a pool of connections based on a cluster configuration in etcd.

NOTE: the public API is not stable and could be changed until release v1.0.0.

package main

import (
	"context"
	"errors"
	"fmt"
	"time"

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

	"github.com/tarantool/go-tarantool/v2"

	"github.com/tarantool/go-discovery"
	"github.com/tarantool/go-discovery/dial"
	"github.com/tarantool/go-discovery/discoverer"
	"github.com/tarantool/go-discovery/filter"
	"github.com/tarantool/go-discovery/pool"
	"github.com/tarantool/go-discovery/scheduler"
	"github.com/tarantool/go-discovery/subscriber"
)

func main() {
	// We need to create a etcd client. The client will use a default port
	// for a local etcd instance. You could change the settings.
	etcd, err := clientv3.New(clientv3.Config{
		Endpoints:   []string{"http://127.0.0.1:2379"},
	})
	if err != nil {
		fmt.Println("Unable to start etcd client:", err)
		return
	}
	defer etcd.Close()

	// The pool will try to connect to instances without TLS. It will send
	// requests in round-robin. But you could use pool.PriorityBalancer to
	// send requests by instances priority.
	examplePool, err := pool.NewPool(
		dial.NewNetDialerFactory("testuser", "testpass", tarantool.Opts{
			Timeout: 5 * time.Second,
		}),
		pool.NewRoundRobinBalancer(),
	)
	if err != nil {
		fmt.Println("Unable to create a pool:", err)
		return
	}

	for {
		// The scheduler will watch for updates from etcd.
		sched := scheduler.NewEtcdWatch(etcd, "/prefix")
		defer sched.Stop()

		disc := discoverer.NewFilter(
			// The base discoverer gets a list of instance configurations from
			// etcd.
			discoverer.NewEtcd(etcd, "/prefix"),
			// The filter filters the list and leaves only instances with
			// Group == "foo".
			filter.GroupOneOf{Groups: []string{"foo"}},
			// You can play with other filters here, see:
			// https://github.com/tarantool/go-discovery/filter/
			// subpackage.
		)

		// The Subscriber will send instance configurations into the pool
		// on updates.
		etcdSubscriber := subscriber.NewFilter(
			// The base subscriber watches for updates from etcd and generates
			// a list of update events from it.
			subscriber.NewSchedule(sched, disc),
			// The filter filters the list of updates and leaves only instances
			// with Name == "instance1".
			filter.NameOneOf{Names: []string{"instance1"}},
		)

		// Subscribe the pool for updates from the subscriber. Subscription
		// only to single subscriber at a moment is supported.
		err := etcdSubscriber.Subscribe(context.Background(), examplePool)
		if err != nil {
			fmt.Println("Failed to subscribe:", err)
			return
		}
		defer etcdSubscriber.Unsubscribe(examplePool)

		for {
			// Any request from go-tarantool could be used.
			request := tarantool.NewEvalRequest("return box.cfg.listen")
			// You could use discovery.ModeRW or discovery.ModeRO to send
			// requests to only RW or RO instances.
			result, err := examplePool.Do(request, discovery.ModeAny).Get()

			if errors.Is(err, pool.ErrUnsubscribed) {
				// Something happened with a scheduler (see log messages), we
				// need to recreate scheduler and subscribe to it again.
				time.Sleep(time.Second)
				fmt.Println("Unsubscribed from a scheduler.")
				break
			}

			if errors.Is(err, pool.ErrNoConnectedInstances) {
				// The pool is not connected to an instance yet. We could just
				// repeat the request. You need to make a more complex
				// business logic.
				fmt.Println("No instances.")
				time.Sleep(time.Second)
				continue
			}

			fmt.Println("Result:", result)
			fmt.Println("Error:", err)
			fmt.Println("Done.")
			return
		}
	}
}

Run tests

To run default set of tests:

go test -v ./...

Development

To generate code, you need to install two utilities:

go install golang.org/x/tools/cmd/stringer@latest
go install github.com/posener/goreadme/cmd/goreadme@latest

After that, you could generate the code.

go generate ./...

In addition, you need to install linter:

go install github.com/golangci/golangci-lint/cmd/golangci-lint@v1.51.1

You could run the linter with the command.

golangci-lint --config .golangci.yml run

Finally, codespell:

pip3 install codespell
codespell

You could also use our Makefile targets:

make deps
make format
make generate
make lint
make test
make testrace

Documentation

You could run the godoc server on localhost:6060 with the command:

make godoc_run

And open the generated documentation in another terminal or use the link:

make godoc_open

Examples

Documentation

Overview

Package discovery implements a discovery for Tarantool 3.0.

Example

Example demonstrates how to create and use the simplest variant of the pool. 

package main

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

	clientv3 "go.etcd.io/etcd/client/v3"
	"go.etcd.io/etcd/tests/v3/integration"

	"github.com/tarantool/go-tarantool/v2"
	"github.com/tarantool/go-tarantool/v2/test_helpers"

	"github.com/tarantool/go-discovery"
	"github.com/tarantool/go-discovery/dial"
	"github.com/tarantool/go-discovery/discoverer"
	"github.com/tarantool/go-discovery/filter"
	"github.com/tarantool/go-discovery/pool"
	"github.com/tarantool/go-discovery/scheduler"
	"github.com/tarantool/go-discovery/subscriber"
)

func init() {
	log.SetOutput(io.Discard)
}

func exampleStartTarantool(address string) (*test_helpers.TarantoolInstance, error) {
	dialer := tarantool.NetDialer{
		Address:  address,
		User:     "testuser",
		Password: "testpass",
	}
	startOpts := test_helpers.StartOpts{
		Dialer:       dialer,
		InitScript:   "testdata/init.lua",
		Listen:       address,
		WaitStart:    100 * time.Millisecond,
		ConnectRetry: 3,
		RetryTimeout: 500 * time.Millisecond,
	}

	return test_helpers.StartTarantool(startOpts)
}

func exampleStopTarantool(instance *test_helpers.TarantoolInstance) {
	test_helpers.StopTarantoolWithCleanup(instance)
}

func main() {
	// Start test Tarantool instance.
	ttInstance, err := exampleStartTarantool("127.0.0.1:3013")
	if err != nil {
		fmt.Println("Failed to start Tarantool:", err)
		return
	}
	defer exampleStopTarantool(ttInstance)

	// Start a test etcd cluster and connect with a client to it.
	cluster := integration.NewLazyCluster()
	defer cluster.Terminate()

	etcd, err := clientv3.New(clientv3.Config{
		Endpoints: cluster.EndpointsGRPC(),
	})
	if err != nil {
		fmt.Println("Unable to start etcd client:", err)
		return
	}
	defer func() { _ = etcd.Close() }()

	// And publish a cluster configuration to it.
	_, err = etcd.Put(context.Background(), "/prefix/config/all", `
groups:
  foo:
    replicasets:
      bar:
        instances:
          instance1:
            iproto:
              listen:
              - uri: 127.0.0.1:3011
              - uri: 127.0.0.1:3013
            roles: [crud]
            labels:
              tags: "any,bar,3"
          instance2:
            iproto:
              listen:
              - uri: 127.0.0.1:3014
          instance3:
            iproto:
              advertise:
                client: 127.0.0.1:3015
`)

	if err != nil {
		fmt.Println("Failed to publish cluster configuration error:", err)
		return
	}

	// The pool will try to connect to instances without TLS. It will send
	// requests in round-robin. But you could use pool.PriorityBalancer to
	// send request into instances by a priority.
	examplePool, err := pool.NewPool(
		dial.NewNetDialerFactory("testuser", "testpass", tarantool.Opts{
			Timeout: 5 * time.Second,
		}),
		pool.NewRoundRobinBalancer(),
	)
	if err != nil {
		fmt.Println("Unable to create a pool:", err)
		return
	}

	for {
		// The scheduler will watch for updates from etcd.
		sched := scheduler.NewEtcdWatch(etcd, "/prefix")
		defer sched.Stop()

		disc := discoverer.NewFilter(
			// The base discoverer gets a list of instance configurations from
			// etcd.
			discoverer.NewEtcd(etcd, "/prefix"),
			// The filter filters the list and leaves only instances with
			// Group == "foo".
			filter.GroupOneOf{Groups: []string{"foo"}},
			// You can play with other filters here, see:
			// https://github.com/tarantool/go-discovery/filter/
			// subpackage.
		)

		// The Subscriber will send instance configurations into the pool
		// on updates.
		etcdSubscriber := subscriber.NewFilter(
			// The base subscriber watches for updates from etcd and generates
			// a list of update events from it.
			subscriber.NewSchedule(sched, disc),
			// The filter filters the list of updates and leaves only instances
			// with Name == "instance1".
			filter.NameOneOf{Names: []string{"instance1"}},
		)

		// Subscribe the pool for updates from the subscriber. Subscription
		// only to single subscriber at a moment is supported.
		err := etcdSubscriber.Subscribe(context.Background(), examplePool)
		if err != nil {
			fmt.Println("Failed to subscribe:", err)
			return
		}
		for {
			// Any request from go-tarantool could be used.
			request := tarantool.NewEvalRequest("return box.cfg.listen")
			// You could use discovery.ModeRW or discovery.ModeRO to send
			// requests to only RW or RO instances.
			result, err := examplePool.Do(request, discovery.ModeAny).Get()

			if errors.Is(err, pool.ErrUnsubscribed) {
				// Something happened with a scheduler (see log messages), we
				// need to recreate scheduler and subscribe to it again.
				time.Sleep(time.Second)
				fmt.Println("Unsubscribed from a scheduler.")
				etcdSubscriber.Unsubscribe(examplePool)
				break
			}

			if errors.Is(err, pool.ErrNoConnectedInstances) {
				// The pool is not connected to an instance yet. We could just
				// repeat the request.
				//
				//  fmt.Println("No instances.")
				//  time.Sleep(time.Second)
				//
				// Commented to make the example repeatable and fast.
				continue
			}

			etcdSubscriber.Unsubscribe(examplePool)
			fmt.Println("Result:", result)
			fmt.Println("Error:", err)
			fmt.Println("Done.")
			return
		}
	}

}

Output:
Result: [127.0.0.1:3013]
Error: <nil>
Done.
Example (Discoverer_Connectable) 
package main

import (
	"context"
	"fmt"
	"io"
	"log"
	"time"

	clientv3 "go.etcd.io/etcd/client/v3"
	"go.etcd.io/etcd/tests/v3/integration"

	"github.com/tarantool/go-tarantool/v2"
	"github.com/tarantool/go-tarantool/v2/test_helpers"

	"github.com/tarantool/go-discovery/dial"
	"github.com/tarantool/go-discovery/discoverer"
)

func init() {
	log.SetOutput(io.Discard)
}

func exampleStartTarantool(address string) (*test_helpers.TarantoolInstance, error) {
	dialer := tarantool.NetDialer{
		Address:  address,
		User:     "testuser",
		Password: "testpass",
	}
	startOpts := test_helpers.StartOpts{
		Dialer:       dialer,
		InitScript:   "testdata/init.lua",
		Listen:       address,
		WaitStart:    100 * time.Millisecond,
		ConnectRetry: 3,
		RetryTimeout: 500 * time.Millisecond,
	}

	return test_helpers.StartTarantool(startOpts)
}

func exampleStopTarantool(instance *test_helpers.TarantoolInstance) {
	test_helpers.StopTarantoolWithCleanup(instance)
}

func main() {
	// Start test Tarantool instance.
	ttInstance, err := exampleStartTarantool("127.0.0.1:3013")
	if err != nil {
		fmt.Println("Failed to start Tarantool:", err)
		return
	}
	defer exampleStopTarantool(ttInstance)

	// Start a test etcd cluster and connect with a client to it.
	cluster := integration.NewLazyCluster()
	defer cluster.Terminate()

	etcd, err := clientv3.New(clientv3.Config{
		Endpoints: cluster.EndpointsGRPC(),
	})
	if err != nil {
		fmt.Println("Unable to start etcd client:", err)
		return
	}
	defer func() { _ = etcd.Close() }()

	// And publish a cluster configuration to it.
	_, err = etcd.Put(context.Background(), "/prefix/config/all", `
groups:
  foo:
    replicasets:
      bar:
        instances:
          instance1:
            iproto:
              advertise:
                client: 127.0.0.1:3013
            roles: [crud]
            labels:
              tags: "any,bar,3"
          instance2:
            iproto:
              advertise:
                client: 127.0.0.1:3014
          instance3:
            iproto:
              advertise:
                client: 127.0.0.1:3015
`)

	if err != nil {
		fmt.Println("Failed to publish cluster configuration error:", err)
		return
	}

	// Create a net dialer factory.
	factory := dial.NewNetDialerFactory("testuser", "testpass",
		tarantool.Opts{Timeout: 5 * time.Second})

	// Create a Connectable discoverer based on the Etcd discoverer.
	disc := discoverer.NewConnectable(factory,
		discoverer.NewEtcd(etcd, "/prefix"))

	// Get a list of connectable instances.
	inst, err := disc.Discovery(context.Background())
	if err != nil {
		fmt.Println("Failed to discover connectable instances:", err)
		return
	}

	for _, instance := range inst {
		fmt.Println("Instance found")
		fmt.Println("Name:", instance.Name)
		fmt.Println("Mode:", instance.Mode)
		fmt.Println("URI:", instance.URI)
		fmt.Println("Roles:", instance.Roles)
		fmt.Println("Labels:", instance.Labels)
		fmt.Println("Group:", instance.Group)
		fmt.Println("Replicaset:", instance.Replicaset)
	}
	fmt.Println("Done.")

}

Output:
Instance found
Name: instance1
Mode: rw
URI: [127.0.0.1:3013]
Roles: [crud]
Labels: map[tags:any,bar,3]
Group: foo
Replicaset: bar
Done.
Example (Discoverer_Connectable_with_Tarantool) 
// Start test Tarantool instance.
ttInstance, err := exampleStartTarantool("127.0.0.1:3013")
if err != nil {
	fmt.Println("Failed to start Tarantool:", err)
	return
}
defer exampleStopTarantool(ttInstance)

tcs, err := tcshelper.Start(0)
if err != nil {
	if errors.Is(err, tcshelper.ErrNotSupported) {
		fmt.Println("TcS is not supported:", err)
	} else {
		fmt.Println("Failed to start TCS:", err)
	}
	return
}
defer tcs.Stop()

err = tcs.Put(context.Background(), "/prefix/config/all", `
groups:
  foo:
    replicasets:
      bar:
        instances:
          instance1:
            iproto:
              advertise:
                client: 127.0.0.1:3013
            roles: [crud]
            labels:
              tags: "any,bar,3"
          instance2:
            iproto:
              advertise:
                client: 127.0.0.1:3014
          instance3:
            iproto:
              advertise:
                client: 127.0.0.1:3015
`)

if err != nil {
	fmt.Println("Failed to publish cluster configuration error:", err)
	return
}

// Create a net dialer factory.
factory := dial.NewNetDialerFactory("testuser", "testpass",
	tarantool.Opts{Timeout: 5 * time.Second})

// Create a Connectable discoverer based on the Etcd discoverer.
disc := discoverer.NewConnectable(factory,
	discoverer.NewTarantool(tcs.Doer(), "/prefix"))

// Get a list of connectable instances.
inst, err := disc.Discovery(context.Background())
if err != nil {
	fmt.Println("Failed to discover connectable instances:", err)
	return
}

for _, instance := range inst {
	fmt.Println("Instance found")
	fmt.Println("Name:", instance.Name)
	fmt.Println("Mode:", instance.Mode)
	fmt.Println("URI:", instance.URI)
	fmt.Println("Roles:", instance.Roles)
	fmt.Println("Labels:", instance.Labels)
	fmt.Println("Group:", instance.Group)
	fmt.Println("Replicaset:", instance.Replicaset)
}
fmt.Println("Done.")
Example (Observer_Accumulator) 
package main

import (
	"fmt"
	"sync"

	"github.com/tarantool/go-discovery"
	"github.com/tarantool/go-discovery/observer"
)

type exampleObserver struct {
	wgEvent sync.WaitGroup
}

func newExampleObserver() *exampleObserver {
	obs := &exampleObserver{}

	obs.wgEvent = sync.WaitGroup{}

	return obs
}

func (o *exampleObserver) Observe(events []discovery.Event, err error) {
	if len(events) > 0 || err == nil {
		fmt.Println("An event found.")
		for _, event := range events {
			instance := event.New
			if event.Type == discovery.EventTypeRemove {
				instance = event.Old
			}

			fmt.Println("Type:", event.Type)
			fmt.Println("Group:", instance.Group)
			fmt.Println("Replicaset:", instance.Replicaset)
			fmt.Println("Name:", instance.Name)
			fmt.Println("Mode:", instance.Mode.String())
			fmt.Println("URI:", instance.URI)
			fmt.Println("Roles:", instance.Roles)
			fmt.Println("Labels:", instance.Labels)
		}
		o.wgEvent.Done()
	}

	if err != nil {
		fmt.Println("Error from the observer:", err)
	}
}

func main() {
	events := []discovery.Event{
		{
			Type: discovery.EventTypeAdd,
			New: discovery.Instance{
				Name:  "Cpt. Jack Sparrow",
				Group: "Black Pearl",
			},
		},
		{
			Type: discovery.EventTypeUpdate,
			Old: discovery.Instance{
				Name:  "Cpt. Jack Sparrow",
				Group: "Black Pearl",
			},
			New: discovery.Instance{
				Name: "Cpt. Jack Sparrow",
			},
		},
		{
			Type: discovery.EventTypeAdd,
			New: discovery.Instance{
				Name:  "Cpt. Barbossa",
				Group: "Black Pearl",
			},
		},
		{
			Type: discovery.EventTypeUpdate,
			Old: discovery.Instance{
				Name:  "Cpt. Barbossa",
				Group: "Black Pearl",
			},
			New: discovery.Instance{
				Name: "Cpt. Barbossa",
			},
		},
		{
			Type: discovery.EventTypeRemove,
			Old: discovery.Instance{
				Name: "Cpt. Barbossa",
			},
		},
		{
			Type: discovery.EventTypeUpdate,
			Old: discovery.Instance{
				Name: "Cpt. Jack Sparrow",
			},
			New: discovery.Instance{
				Name:       "Cpt. Jack Sparrow",
				Group:      "Black Pearl",
				Replicaset: "Pirates",
			},
		},
	}

	obs := newExampleObserver()
	obs.wgEvent.Add(1)

	accumulator := observer.NewAccumulator(obs)

	accumulator.Observe(events, nil)

	accumulator.Observe(nil, fmt.Errorf("done"))
	fmt.Println("Done.")

}

Output:
An event found.
Type: add
Group: Black Pearl
Replicaset: Pirates
Name: Cpt. Jack Sparrow
Mode: any
URI: []
Roles: []
Labels: map[]
Error from the observer: done
Done.
Example (Subscriber_Connectable) 
package main

import (
	"context"
	"fmt"
	"io"
	"log"
	"sync"
	"time"

	clientv3 "go.etcd.io/etcd/client/v3"
	"go.etcd.io/etcd/tests/v3/integration"

	"github.com/tarantool/go-tarantool/v2"
	"github.com/tarantool/go-tarantool/v2/test_helpers"

	"github.com/tarantool/go-discovery"
	"github.com/tarantool/go-discovery/dial"
	"github.com/tarantool/go-discovery/discoverer"
	"github.com/tarantool/go-discovery/scheduler"
	"github.com/tarantool/go-discovery/subscriber"
)

func init() {
	log.SetOutput(io.Discard)
}

func exampleStartTarantool(address string) (*test_helpers.TarantoolInstance, error) {
	dialer := tarantool.NetDialer{
		Address:  address,
		User:     "testuser",
		Password: "testpass",
	}
	startOpts := test_helpers.StartOpts{
		Dialer:       dialer,
		InitScript:   "testdata/init.lua",
		Listen:       address,
		WaitStart:    100 * time.Millisecond,
		ConnectRetry: 3,
		RetryTimeout: 500 * time.Millisecond,
	}

	return test_helpers.StartTarantool(startOpts)
}

func exampleStopTarantool(instance *test_helpers.TarantoolInstance) {
	test_helpers.StopTarantoolWithCleanup(instance)
}

type exampleObserver struct {
	wgEvent sync.WaitGroup
}

func newExampleObserver() *exampleObserver {
	obs := &exampleObserver{}

	obs.wgEvent = sync.WaitGroup{}

	return obs
}

func (o *exampleObserver) Observe(events []discovery.Event, err error) {
	if len(events) > 0 || err == nil {
		fmt.Println("An event found.")
		for _, event := range events {
			instance := event.New
			if event.Type == discovery.EventTypeRemove {
				instance = event.Old
			}

			fmt.Println("Type:", event.Type)
			fmt.Println("Group:", instance.Group)
			fmt.Println("Replicaset:", instance.Replicaset)
			fmt.Println("Name:", instance.Name)
			fmt.Println("Mode:", instance.Mode.String())
			fmt.Println("URI:", instance.URI)
			fmt.Println("Roles:", instance.Roles)
			fmt.Println("Labels:", instance.Labels)
		}
		o.wgEvent.Done()
	}

	if err != nil {
		fmt.Println("Error from the observer:", err)
	}
}

func main() {
	// Start test Tarantool instance.
	ttInstance, err := exampleStartTarantool("127.0.0.1:3013")
	if err != nil {
		fmt.Println("Failed to start Tarantool:", err)
		return
	}
	defer exampleStopTarantool(ttInstance)

	cluster := integration.NewLazyCluster()
	defer cluster.Terminate()

	etcd, err := clientv3.New(clientv3.Config{
		Endpoints: cluster.EndpointsGRPC(),
	})
	if err != nil {
		fmt.Println("Unable to start etcd client:", err)
		return
	}
	defer func() { _ = etcd.Close() }()

	// Create a net dialer factory.
	factory := dial.NewNetDialerFactory("testuser", "testpass",
		tarantool.Opts{Timeout: 5 * time.Second})

	// Create a Connectable subscriber based on the Etcd subscriber.
	connectable := subscriber.NewConnectable(factory,
		subscriber.NewSchedule(scheduler.NewEtcdWatch(etcd, "foo"),
			discoverer.NewEtcd(etcd, "foo")))

	obs := newExampleObserver()
	obs.wgEvent.Add(1)

	err = connectable.Subscribe(context.Background(), obs)
	if err != nil {
		fmt.Println("Subscribe error:", err)
		return
	}
	defer connectable.Unsubscribe(obs)

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	_, err = etcd.Put(ctx, "foo/config/key", `
groups:
  foo:
    replicasets:
      bar:
        instances:
          instance1:
            iproto:
              listen:
              - uri: 127.0.0.1:3013
            roles: [crud]
            labels:
              tags: "any,bar,3"
          instance2:
            iproto:
              advertise:
                client: 127.0.0.1:3014
          instance3:
            iproto:
              advertise:
                client: 127.0.0.1:3015
`)

	if err != nil {
		fmt.Println("Failed to publish cluster configuration error:", err)
		return
	}

	obs.wgEvent.Wait()
	obs.wgEvent.Add(1)
	connectable.Unsubscribe(obs)

	fmt.Println("Done.")

}

Output:
An event found.
Type: add
Group: foo
Replicaset: bar
Name: instance1
Mode: rw
URI: [127.0.0.1:3013]
Roles: [crud]
Labels: map[tags:any,bar,3]
An event found.
Type: remove
Group: foo
Replicaset: bar
Name: instance1
Mode: rw
URI: [127.0.0.1:3013]
Roles: [crud]
Labels: map[tags:any,bar,3]
Error from the observer: unsubscribed
Done.
Example (Subscriber_Filter) 
package main

import (
	"context"
	"fmt"
	"io"
	"log"
	"sync"
	"time"

	clientv3 "go.etcd.io/etcd/client/v3"
	"go.etcd.io/etcd/tests/v3/integration"

	"github.com/tarantool/go-discovery"
	"github.com/tarantool/go-discovery/discoverer"
	"github.com/tarantool/go-discovery/scheduler"
	"github.com/tarantool/go-discovery/subscriber"
)

func init() {
	log.SetOutput(io.Discard)
}

type exampleObserver struct {
	wgEvent sync.WaitGroup
}

func newExampleObserver() *exampleObserver {
	obs := &exampleObserver{}

	obs.wgEvent = sync.WaitGroup{}

	return obs
}

func (o *exampleObserver) Observe(events []discovery.Event, err error) {
	if len(events) > 0 || err == nil {
		fmt.Println("An event found.")
		for _, event := range events {
			instance := event.New
			if event.Type == discovery.EventTypeRemove {
				instance = event.Old
			}

			fmt.Println("Type:", event.Type)
			fmt.Println("Group:", instance.Group)
			fmt.Println("Replicaset:", instance.Replicaset)
			fmt.Println("Name:", instance.Name)
			fmt.Println("Mode:", instance.Mode.String())
			fmt.Println("URI:", instance.URI)
			fmt.Println("Roles:", instance.Roles)
			fmt.Println("Labels:", instance.Labels)
		}
		o.wgEvent.Done()
	}

	if err != nil {
		fmt.Println("Error from the observer:", err)
	}
}

func main() {
	cluster := integration.NewLazyCluster()
	defer cluster.Terminate()

	etcd, err := clientv3.New(clientv3.Config{
		Endpoints: cluster.EndpointsGRPC(),
	})
	if err != nil {
		fmt.Println("Unable to start etcd client:", err)
		return
	}
	defer func() { _ = etcd.Close() }()

	schedule := subscriber.NewSchedule(scheduler.NewEtcdWatch(etcd, "foo"),
		discoverer.NewEtcd(etcd, "foo"))

	filters := []discovery.Filter{
		discovery.FilterFunc(func(inst discovery.Instance) bool {
			return inst.Name == "zoo"
		}),
		discovery.FilterFunc(func(inst discovery.Instance) bool {
			return inst.Group == "foo"
		}),
	}
	filter := subscriber.NewFilter(schedule, filters...)

	obs := newExampleObserver()
	obs.wgEvent.Add(2)

	err = filter.Subscribe(context.Background(), obs)
	if err != nil {
		fmt.Println("Subscribe error:", err)
		return
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	_, err = etcd.Put(ctx, "foo/config/key", `
database:
  mode: ro
groups:
  foo:
    replicasets:
      bar:
        instances:
          zoo:
            iproto:
              advertise:
                client: localhost:3011
                peer:
                  params:
                    transport: plain
            roles: [crud]
            labels:
              tags: "any,bar,3"
  foo2:
    replicasets:
      bar2:
        instances:
          zoo2:
            iproto:
              advertise:
                client: localhost:3012
                peer:
                  params:
                    transport: ssl
            roles: [crud]
            labels:
              tags: "any,bar,3"
`)
	cancel()

	if err != nil {
		fmt.Println("Put error:", err)
		return
	}

	obs.wgEvent.Wait()
	filter.Unsubscribe(obs)

	fmt.Println("Done.")

}

Output:
An event found.
An event found.
Type: add
Group: foo
Replicaset: bar
Name: zoo
Mode: ro
URI: [localhost:3011]
Roles: [crud]
Labels: map[tags:any,bar,3]
Error from the observer: unsubscribed
Done.
Example (Subscriber_Filter_Etcd_canceled) 
package main

import (
	"context"
	"fmt"
	"io"
	"log"
	"sync"

	clientv3 "go.etcd.io/etcd/client/v3"
	"go.etcd.io/etcd/tests/v3/integration"

	"github.com/tarantool/go-discovery"
	"github.com/tarantool/go-discovery/discoverer"
	"github.com/tarantool/go-discovery/scheduler"
	"github.com/tarantool/go-discovery/subscriber"
)

func init() {
	log.SetOutput(io.Discard)
}

type exampleObserver struct {
	wgEvent sync.WaitGroup
}

func newExampleObserver() *exampleObserver {
	obs := &exampleObserver{}

	obs.wgEvent = sync.WaitGroup{}

	return obs
}

func (o *exampleObserver) Observe(events []discovery.Event, err error) {
	if len(events) > 0 || err == nil {
		fmt.Println("An event found.")
		for _, event := range events {
			instance := event.New
			if event.Type == discovery.EventTypeRemove {
				instance = event.Old
			}

			fmt.Println("Type:", event.Type)
			fmt.Println("Group:", instance.Group)
			fmt.Println("Replicaset:", instance.Replicaset)
			fmt.Println("Name:", instance.Name)
			fmt.Println("Mode:", instance.Mode.String())
			fmt.Println("URI:", instance.URI)
			fmt.Println("Roles:", instance.Roles)
			fmt.Println("Labels:", instance.Labels)
		}
		o.wgEvent.Done()
	}

	if err != nil {
		fmt.Println("Error from the observer:", err)
	}
}

func main() {
	cluster := integration.NewLazyCluster()
	defer cluster.Terminate()

	etcd, err := clientv3.New(clientv3.Config{
		Endpoints: cluster.EndpointsGRPC(),
	})
	if err != nil {
		fmt.Println("Unable to start etcd client:", err)
		return
	}
	defer func() { _ = etcd.Close() }()

	schedule := subscriber.NewSchedule(scheduler.NewEtcdWatch(etcd, "foo"),
		discoverer.NewEtcd(etcd, "foo"))

	filters := []discovery.Filter{
		discovery.FilterFunc(func(inst discovery.Instance) bool {
			return inst.Name == "zoo"
		}),
	}
	filter := subscriber.NewFilter(schedule, filters...)

	obs := newExampleObserver()

	ctx, cancel := context.WithCancel(context.Background())
	cancel()

	err = filter.Subscribe(ctx, obs)
	if err != nil {
		fmt.Println("Subscribe error:", err)
	}
	fmt.Println("Done.")

}

Output:
Subscribe error: context canceled
Done.
Example (Subscriber_Schedule_Etcd) 
package main

import (
	"context"
	"fmt"
	"io"
	"log"
	"sync"
	"time"

	clientv3 "go.etcd.io/etcd/client/v3"
	"go.etcd.io/etcd/tests/v3/integration"

	"github.com/tarantool/go-discovery"
	"github.com/tarantool/go-discovery/discoverer"
	"github.com/tarantool/go-discovery/scheduler"
	"github.com/tarantool/go-discovery/subscriber"
)

func init() {
	log.SetOutput(io.Discard)
}

type exampleObserver struct {
	wgEvent sync.WaitGroup
}

func newExampleObserver() *exampleObserver {
	obs := &exampleObserver{}

	obs.wgEvent = sync.WaitGroup{}

	return obs
}

func (o *exampleObserver) Observe(events []discovery.Event, err error) {
	if len(events) > 0 || err == nil {
		fmt.Println("An event found.")
		for _, event := range events {
			instance := event.New
			if event.Type == discovery.EventTypeRemove {
				instance = event.Old
			}

			fmt.Println("Type:", event.Type)
			fmt.Println("Group:", instance.Group)
			fmt.Println("Replicaset:", instance.Replicaset)
			fmt.Println("Name:", instance.Name)
			fmt.Println("Mode:", instance.Mode.String())
			fmt.Println("URI:", instance.URI)
			fmt.Println("Roles:", instance.Roles)
			fmt.Println("Labels:", instance.Labels)
		}
		o.wgEvent.Done()
	}

	if err != nil {
		fmt.Println("Error from the observer:", err)
	}
}

func main() {
	cluster := integration.NewLazyCluster()
	defer cluster.Terminate()

	etcd, err := clientv3.New(clientv3.Config{
		Endpoints: cluster.EndpointsGRPC(),
	})
	if err != nil {
		fmt.Println("Unable to start etcd client:", err)
		return
	}
	defer func() { _ = etcd.Close() }()

	schedule := subscriber.NewSchedule(scheduler.NewEtcdWatch(etcd, "foo"),
		discoverer.NewEtcd(etcd, "foo"))

	obs := newExampleObserver()
	obs.wgEvent.Add(2)

	err = schedule.Subscribe(context.Background(), obs)
	if err != nil {
		fmt.Println("Subscribe error:", err)
		return
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	_, err = etcd.Put(ctx, "foo/config/key", `
database:
  mode: ro
groups:
  foo:
    replicasets:
      bar:
        instances:
          zoo:
            iproto:
              advertise:
                client: localhost:3011
            roles: [crud]
            labels:
              tags: "any,bar,3"
`)
	cancel()

	if err != nil {
		fmt.Println("Put error:", err)
		return
	}

	obs.wgEvent.Wait()
	schedule.Unsubscribe(obs)

	fmt.Println("Done.")

}

Output:
An event found.
An event found.
Type: add
Group: foo
Replicaset: bar
Name: zoo
Mode: ro
URI: [localhost:3011]
Roles: [crud]
Labels: map[tags:any,bar,3]
Error from the observer: unsubscribed
Done.
Example (Subscriber_Schedule_Etcd_canceled) 
package main

import (
	"context"
	"fmt"
	"io"
	"log"
	"sync"

	clientv3 "go.etcd.io/etcd/client/v3"
	"go.etcd.io/etcd/tests/v3/integration"

	"github.com/tarantool/go-discovery"
	"github.com/tarantool/go-discovery/discoverer"
	"github.com/tarantool/go-discovery/scheduler"
	"github.com/tarantool/go-discovery/subscriber"
)

func init() {
	log.SetOutput(io.Discard)
}

type exampleObserver struct {
	wgEvent sync.WaitGroup
}

func newExampleObserver() *exampleObserver {
	obs := &exampleObserver{}

	obs.wgEvent = sync.WaitGroup{}

	return obs
}

func (o *exampleObserver) Observe(events []discovery.Event, err error) {
	if len(events) > 0 || err == nil {
		fmt.Println("An event found.")
		for _, event := range events {
			instance := event.New
			if event.Type == discovery.EventTypeRemove {
				instance = event.Old
			}

			fmt.Println("Type:", event.Type)
			fmt.Println("Group:", instance.Group)
			fmt.Println("Replicaset:", instance.Replicaset)
			fmt.Println("Name:", instance.Name)
			fmt.Println("Mode:", instance.Mode.String())
			fmt.Println("URI:", instance.URI)
			fmt.Println("Roles:", instance.Roles)
			fmt.Println("Labels:", instance.Labels)
		}
		o.wgEvent.Done()
	}

	if err != nil {
		fmt.Println("Error from the observer:", err)
	}
}

func main() {
	cluster := integration.NewLazyCluster()
	defer cluster.Terminate()

	etcd, err := clientv3.New(clientv3.Config{
		Endpoints: cluster.EndpointsGRPC(),
	})
	if err != nil {
		fmt.Println("Unable to start etcd client:", err)
		return
	}
	defer func() { _ = etcd.Close() }()

	schedule := subscriber.NewSchedule(scheduler.NewEtcdWatch(etcd, "foo"),
		discoverer.NewEtcd(etcd, "foo"))

	obs := newExampleObserver()

	ctx, cancel := context.WithCancel(context.Background())
	cancel()

	err = schedule.Subscribe(ctx, obs)
	if err != nil {
		fmt.Println("Subscribe error:", err)
	}
	fmt.Println("Done.")

}

Output:
Subscribe error: context canceled
Done.

Index

Examples

Constants

This section is empty.

Variables

View Source 
var (
	// ErrMissingObserver is an error that tells that the provided
	// observer is nil.
	ErrMissingObserver = fmt.Errorf("observer is missing")
	// ErrUnsubscribe is an error argument for call to observer.Observe() while
	// unsubscribing the observer.
	ErrUnsubscribe = fmt.Errorf("unsubscribed")
)
View Source 
var ErrSchedulerStopped = fmt.Errorf("scheduler was stopped")

ErrSchedulerStopped is an error that tells that user stopped the scheduler. All Wait() calls will return this error after the Stop() call.

Functions

This section is empty.

Types

type DialerFactory 

type DialerFactory interface {
	// NewDialer creates a new tarantool connection dialer and connection
	// options.
	NewDialer(instance Instance) (tarantool.Dialer, tarantool.Opts, error)
}

DialerFactory is an interface that wraps the method for creating a tarantool connection dialer and connection options.

type Discoverer 

type Discoverer interface {
	// Discovery returns a list of instance configurations. In case of any
	// errors while trying to retrieve the list, it returns an error.
	Discovery(ctx context.Context) ([]Instance, error)
}

Discoverer is an interface for retrieving a list of instance configurations from a configuration source.

type Event 

type Event struct {
	// Type of the current event.
	Type EventType
	// Old refers to a previous configuration of the instance. The value
	// present for EventTypeUpdate and EventTypeRemove.
	Old Instance
	// New refers to a new configuration of the instance. This value is
	// present for EventTypeUpdate and EventTypeAdd.
	New Instance
}

Event is a base event for an instance configuration update.

type EventType 

type EventType int

EventType defines an enumeration of available event types. 

const (
	// EventTypeAdd defines an instance configuration add event.
	EventTypeAdd EventType = iota // add
	// EventTypeUpdate defines an instance configuration update event.
	EventTypeUpdate // update
	// EventTypeRemove defines an instance configuration remove event.
	EventTypeRemove // remove
)

func (EventType) String 

func (i EventType) String() string

type Filter 

type Filter interface {
	// Filter return true if an instance matches a condition.
	Filter(instance Instance) bool
}

Filter is an interface to filter an instance by some conditions. A set of filters implemented in the `filter` subpackage.

type FilterFunc 

type FilterFunc func(instance Instance) bool

FilterFunc allows to adopt a function to a filter, as example: 

var filter Filter = FilterFunc(func(intstance Instance) bool {
    return true
})

func (FilterFunc) Filter 

func (f FilterFunc) Filter(instance Instance) bool

Filter calls a function itself to implement the Filter interface.

type Instance 

type Instance struct {
	// Group is a name of the instance's group.
	Group string
	// Replicaset is a name of the instance's replicaset.
	Replicaset string
	// Name is a name of the instance.
	Name string
	// Mode is a current mode of the instance.
	Mode Mode
	// URI is a set of available URIs to connect to the instance.
	URI []string
	// Roles is a set of the instance's roles.
	Roles []string
	// Labels is a map of string labels of the instance.
	Labels map[string]string
	// RolesCfg is a map of role name to its configuration, as defined in
	// the roles_cfg section of the cluster config for this instance.
	RolesCfg map[string]any
}

Instance describes a single instance configuration.

type Mode 

type Mode int

Mode defines an enumeration of a read-write work mode for an instance. 

const (
	// ModeAny is used when a target instance could be both.
	ModeAny Mode = iota // any
	// ModeRO defines a read-only instance.
	ModeRO // ro
	// ModeRW defines a read-write instance.
	ModeRW // rw
)

func (Mode) String 

func (i Mode) String() string

type Observer 

type Observer interface {
	// Observe receives either a list of new events or an error. After
	// receiving an error, the observer will no longer receive new events.
	Observe(events []Event, err error)
}

Observer interface to subscribe for a new events from a configuration update source.

type Scheduler 

type Scheduler interface {
	// Wait blocks until a next event. It returns `nil` if an event occurs or
	// an `error` if the context expired or the Scheduler implementation has
	// been stopped.
	//
	// If the Scheduler was stopped by user calling the Stop() method, the
	// ErrSchedulerStopped error will be returned.
	Wait(ctx context.Context) error
	// Stop stops an event generation.
	Stop()
}

Scheduler is an interface that allows to schedule update events. An implementation of the interface should generate events, a client should wait on the `Wait()` method for a next event.

type Subscriber 

type Subscriber interface {
	// Subscribe subscribes an observer to new update configuration events. The
	// subscriber should send the first part of update events to the observer
	// before exit from the call to confirm subscription.
	// The context is used to cancel in-progress subscription and does not
	// Unsubscribe the observer on expiration after successful subscription.
	Subscribe(ctx context.Context, observer Observer) error
	// Unsubscribe unsubscribes the observer. A subscriber should not send new
	// events to the observer after the call.
	Unsubscribe(observer Observer)
}

Subscriber is an interface that allows to subscribe to an instance configuration update events.

Source Files

View all Source files

Directories

Path Synopsis
Package dial implements a dialer for Tarantool 3.0.
 Package dial implements a dialer for Tarantool 3.0.
Package discoverer implements a discoverer for Tarantool 3.0.
 Package discoverer implements a discoverer for Tarantool 3.0.
Package filter implements filters for Tarantool 3.0 discovery.
 Package filter implements filters for Tarantool 3.0 discovery.
Package observer implements an observer for Tarantool 3.0.
 Package observer implements an observer for Tarantool 3.0.
Package pool implements a pool of connections for Tarantool 3.0.
 Package pool implements a pool of connections for Tarantool 3.0.
internal/container/buffer
Package buffer implements buffers for a pool container.
 Package buffer implements buffers for a pool container.
Package scheduler implements a scheduler for Tarantool 3.0.
 Package scheduler implements a scheduler for Tarantool 3.0.
Package subscriber implements a subscriber for Tarantool 3.0.
 Package subscriber implements a subscriber for Tarantool 3.0.

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.