[vtadmin] custom discovery resolver (#9977)

* More test fixes: 1. Register `VtctldServer`, not `VtctlServer`. 2. Abuse `GetKeyspace` to allow the client to inspect the listen address of a given server to use in assertions. 3. Building on (2), stop asserting on proxy.host, which will be going away> Signed-off-by: Andrew Mason <andrew@planetscale.com> * Add initial proof-of-concept for custom grpc resolver Signed-off-by: Andrew Mason <andrew@planetscale.com> * Add Options for passing through cluster-specific configs to Resolvers Still need to actually add flags for these, but this will make it easier. Also make some things unexported. Signed-off-by: Andrew Mason <andrew@planetscale.com> * tidy imports Signed-off-by: Andrew Mason <andrew@planetscale.com> * Better, more structured, logging, beginning vtgate/vtsql support Signed-off-by: Andrew Mason <andrew@planetscale.com> * More documentation, plus unexport our resolver implementation Signed-off-by: Andrew Mason <andrew@planetscale.com> * More docs Signed-off-by: Andrew Mason <andrew@planetscale.com> * gut WaitForReady Signed-off-by: Andrew Mason <andrew@planetscale.com> * Add copyright notice Signed-off-by: Andrew Mason <andrew@planetscale.com> * Add unit tests for cluster resolver Signed-off-by: Andrew Mason <andrew@planetscale.com> * Add wrappers to detect ResolveNow calls and remove Sleep from test 🎉 Signed-off-by: Andrew Mason <andrew@planetscale.com> * Document improvement ideas. I'll follow-up on these in a second change Signed-off-by: Andrew Mason <andrew@planetscale.com> * Add *Addrs variants of discovery for gates and vtctlds Signed-off-by: Andrew Mason <andrew@planetscale.com> * Refactors! Better discoveryFunc abstraction and still update addrs with empty list, and tests Signed-off-by: Andrew Mason <andrew@planetscale.com> * Add debug.Debuggable support for resolvers Signed-off-by: Andrew Mason <andrew@planetscale.com> * Use cluster discovery resolver for vtgate/vtsql as well Signed-off-by: Andrew Mason <andrew@planetscale.com> * Fix tests Signed-off-by: Andrew Mason <andrew@planetscale.com> * remove waitforready-related flag and config Signed-off-by: Andrew Mason <andrew@planetscale.com> * post merge cleanup Signed-off-by: Andrew Mason <andrew@planetscale.com> * The big refactor to enable configurable resolver options Signed-off-by: Andrew Mason <andrew@planetscale.com> * remove unnecssary discovery call Signed-off-by: Andrew Mason <andrew@planetscale.com> * remove WithBlock debugging aid Signed-off-by: Andrew Mason <andrew@planetscale.com> * rename resolver debug fields for consistency Signed-off-by: Andrew Mason <andrew@planetscale.com> * Push discovery down to resolver level, update tests Signed-off-by: Andrew Mason <andrew@planetscale.com> * annotate resolver span with addrs Signed-off-by: Andrew Mason <andrew@planetscale.com> * add util function for dial addrs Signed-off-by: Andrew Mason <andrew@planetscale.com> * Remove .host field as it does not have a use anymore Signed-off-by: Andrew Mason <andrew@planetscale.com> * inline vtsql trace helper Signed-off-by: Andrew Mason <andrew@planetscale.com> * Support balancer policies Signed-off-by: Andrew Mason <andrew@planetscale.com> * add link to service config docs Signed-off-by: Andrew Mason <andrew@planetscale.com> * lol ...... Signed-off-by: Andrew Mason <andrew@planetscale.com> * add documentation Signed-off-by: Andrew Mason <andrew@planetscale.com> * Update code to use new target.URL field (all other fields are deprecated) Signed-off-by: Andrew Mason <andrew@planetscale.com> * Update tests with URL field Signed-off-by: Andrew Mason <andrew@planetscale.com>
vitessio · Apr 4, 2022 · 94a6b76 · 94a6b76
1 parent 3d4f400
commit 94a6b76
Show file tree

Hide file tree

Showing 19 changed files with 919 additions and 218 deletions.
diff --git a/go/vt/vtadmin/cluster/cluster_test.go b/go/vt/vtadmin/cluster/cluster_test.go
@@ -36,6 +36,7 @@ import (
 	"vitess.io/vitess/go/vt/vitessdriver"
 	"vitess.io/vitess/go/vt/vtadmin/cluster"
 	"vitess.io/vitess/go/vt/vtadmin/cluster/discovery/fakediscovery"
+	"vitess.io/vitess/go/vt/vtadmin/cluster/resolver"
 	vtadminerrors "vitess.io/vitess/go/vt/vtadmin/errors"
 	"vitess.io/vitess/go/vt/vtadmin/testutil"
 	"vitess.io/vitess/go/vt/vtadmin/vtctldclient/fakevtctldclient"
@@ -2652,7 +2653,9 @@ func TestGetTablets(t *testing.T) {
 			Id:   "c1",
 			Name: "one",
 		},
-		Discovery: disco,
+		ResolverOptions: &resolver.Options{
+			Discovery: disco,
+		},
 	})
 	db.DialFunc = func(cfg vitessdriver.Configuration) (*sql.DB, error) {
 		return nil, assert.AnError

diff --git a/go/vt/vtadmin/cluster/discovery/discovery.go b/go/vt/vtadmin/cluster/discovery/discovery.go
@@ -53,6 +53,10 @@ type Discovery interface {
 	// return an address is not specified by the interface, and can be
 	// implementation-specific.
 	DiscoverVTGateAddr(ctx context.Context, tags []string) (string, error)
+	// DiscoverVTGateAddrs returns a list of addresses of vtgates found in the
+	// discovery service. This is semantically equivalent to the result of
+	// DiscoverVTGateAddr for each gate returned by a call to DiscoverVTGates.
+	DiscoverVTGateAddrs(ctx context.Context, tags []string) ([]string, error)
 	// DiscoverVTGates returns a list of vtgates found in the discovery service.
 	// Tags can optionally be used to filter gates. Order of the gates is not
 	// specified by the interface, and can be implementation-specific.
@@ -68,6 +72,10 @@ type Discovery interface {
 	// return an address is not specified by the interface, and can be
 	// implementation-specific.
 	DiscoverVtctldAddr(ctx context.Context, tags []string) (string, error)
+	// DiscoverVtctldAddrs returns a list of addresses of vtctlds found in the
+	// discovery service. This is semantically equivalent to the result of
+	// DiscoverVtctldAddr for each gate returned by a call to DiscoverVtctlds.
+	DiscoverVtctldAddrs(ctx context.Context, tags []string) ([]string, error)
 	// DiscoverVtctlds returns a list of vtctlds found in the discovery service.
 	// Tags can optionally be used to filter vtctlds. Order of the vtctlds is
 	// not specified by the interface, and can be implementation-specific.

diff --git a/go/vt/vtadmin/cluster/discovery/discovery_consul.go b/go/vt/vtadmin/cluster/discovery/discovery_consul.go
@@ -228,6 +228,31 @@ func (c *ConsulDiscovery) DiscoverVTGateAddr(ctx context.Context, tags []string)
 	return addr, nil
 }
 
+// DiscoverVTGateAddrs is part of the Discovery interface.
+func (c *ConsulDiscovery) DiscoverVTGateAddrs(ctx context.Context, tags []string) ([]string, error) {
+	span, ctx := trace.NewSpan(ctx, "ConsulDiscovery.DiscoverVTGateAddrs")
+	defer span.Finish()
+
+	executeFQDNTemplate := false
+
+	vtgates, err := c.discoverVTGates(ctx, tags, executeFQDNTemplate)
+	if err != nil {
+		return nil, err
+	}
+
+	addrs := make([]string, len(vtgates))
+	for i, vtgate := range vtgates {
+		addr, err := textutil.ExecuteTemplate(c.vtgateAddrTmpl, vtgate)
+		if err != nil {
+			return nil, fmt.Errorf("failed to execute vtgate address template for %v: %w", vtgate, err)
+		}
+
+		addrs[i] = addr
+	}
+
+	return addrs, nil
+}
+
 // DiscoverVTGates is part of the Discovery interface.
 func (c *ConsulDiscovery) DiscoverVTGates(ctx context.Context, tags []string) ([]*vtadminpb.VTGate, error) {
 	span, ctx := trace.NewSpan(ctx, "ConsulDiscovery.DiscoverVTGates")
@@ -348,6 +373,31 @@ func (c *ConsulDiscovery) DiscoverVtctldAddr(ctx context.Context, tags []string)
 	return addr, nil
 }
 
+// DiscoverVtctldAddrs is part of the Discovery interface.
+func (c *ConsulDiscovery) DiscoverVtctldAddrs(ctx context.Context, tags []string) ([]string, error) {
+	span, ctx := trace.NewSpan(ctx, "ConsulDiscovery.DiscoverVtctldAddrs")
+	defer span.Finish()
+
+	executeFQDNTemplate := false
+
+	vtctlds, err := c.discoverVtctlds(ctx, tags, executeFQDNTemplate)
+	if err != nil {
+		return nil, err
+	}
+
+	addrs := make([]string, len(vtctlds))
+	for i, vtctld := range vtctlds {
+		addr, err := textutil.ExecuteTemplate(c.vtctldAddrTmpl, vtctld)
+		if err != nil {
+			return nil, fmt.Errorf("failed to execute vtctld address template for %v: %w", vtctld, err)
+		}
+
+		addrs[i] = addr
+	}
+
+	return addrs, nil
+}
+
 // DiscoverVtctlds is part of the Discovery interface.
 func (c *ConsulDiscovery) DiscoverVtctlds(ctx context.Context, tags []string) ([]*vtadminpb.Vtctld, error) {
 	span, ctx := trace.NewSpan(ctx, "ConsulDiscovery.DiscoverVtctlds")

diff --git a/go/vt/vtadmin/cluster/discovery/discovery_json.go b/go/vt/vtadmin/cluster/discovery/discovery_json.go
@@ -23,6 +23,7 @@ import (
 	"math/rand"
 
 	"vitess.io/vitess/go/trace"
+
 	vtadminpb "vitess.io/vitess/go/vt/proto/vtadmin"
 )
 
@@ -146,6 +147,24 @@ func (d *JSONDiscovery) DiscoverVTGateAddr(ctx context.Context, tags []string) (
 	return gate.Hostname, nil
 }
 
+// DiscoverVTGateAddrs is part of the Discovery interface.
+func (d *JSONDiscovery) DiscoverVTGateAddrs(ctx context.Context, tags []string) ([]string, error) {
+	span, ctx := trace.NewSpan(ctx, "JSONDiscovery.DiscoverVTGateAddrs")
+	defer span.Finish()
+
+	gates, err := d.discoverVTGates(ctx, tags)
+	if err != nil {
+		return nil, err
+	}
+
+	addrs := make([]string, len(gates))
+	for i, gate := range gates {
+		addrs[i] = gate.Hostname
+	}
+
+	return addrs, nil
+}
+
 // DiscoverVTGates is part of the Discovery interface.
 func (d *JSONDiscovery) DiscoverVTGates(ctx context.Context, tags []string) ([]*vtadminpb.VTGate, error) {
 	span, ctx := trace.NewSpan(ctx, "JSONDiscovery.DiscoverVTGates")
@@ -228,6 +247,24 @@ func (d *JSONDiscovery) DiscoverVtctldAddr(ctx context.Context, tags []string) (
 	return vtctld.Hostname, nil
 }
 
+// DiscoverVtctldAddrs is part of the Discovery interface.
+func (d *JSONDiscovery) DiscoverVtctldAddrs(ctx context.Context, tags []string) ([]string, error) {
+	span, ctx := trace.NewSpan(ctx, "JSONDiscovery.DiscoverVtctldAddrs")
+	defer span.Finish()
+
+	vtctlds, err := d.discoverVtctlds(ctx, tags)
+	if err != nil {
+		return nil, err
+	}
+
+	addrs := make([]string, len(vtctlds))
+	for i, vtctld := range vtctlds {
+		addrs[i] = vtctld.Hostname
+	}
+
+	return addrs, nil
+}
+
 // DiscoverVtctlds is part of the Discovery interface.
 func (d *JSONDiscovery) DiscoverVtctlds(ctx context.Context, tags []string) ([]*vtadminpb.Vtctld, error) {
 	span, ctx := trace.NewSpan(ctx, "JSONDiscovery.DiscoverVtctlds")

diff --git a/go/vt/vtadmin/cluster/discovery/fakediscovery/discovery.go b/go/vt/vtadmin/cluster/discovery/fakediscovery/discovery.go
@@ -181,6 +181,21 @@ func (d *Fake) DiscoverVTGateAddr(ctx context.Context, tags []string) (string, e
 	return gate.Hostname, nil
 }
 
+// DiscoverVTGateAddrs is part of the discovery.Discovery interface.
+func (d *Fake) DiscoverVTGateAddrs(ctx context.Context, tags []string) ([]string, error) {
+	gates, err := d.DiscoverVTGates(ctx, tags)
+	if err != nil {
+		return nil, err
+	}
+
+	addrs := make([]string, len(gates))
+	for i, gate := range gates {
+		addrs[i] = gate.Hostname
+	}
+
+	return addrs, nil
+}
+
 // DiscoverVtctlds is part of the discover.Discovery interface.
 func (d *Fake) DiscoverVtctlds(ctx context.Context, tags []string) ([]*vtadminpb.Vtctld, error) {
 	if d.vtctlds.shouldErr {
@@ -234,6 +249,21 @@ func (d *Fake) DiscoverVtctldAddr(ctx context.Context, tags []string) (string, e
 	return vtctld.Hostname, nil
 }
 
+// DiscoverVtctldAddrs is part of the discovery.Discovery interface.
+func (d *Fake) DiscoverVtctldAddrs(ctx context.Context, tags []string) ([]string, error) {
+	vtctlds, err := d.DiscoverVtctlds(ctx, tags)
+	if err != nil {
+		return nil, err
+	}
+
+	addrs := make([]string, len(vtctlds))
+	for i, vtctld := range vtctlds {
+		addrs[i] = vtctld.Hostname
+	}
+
+	return addrs, nil
+}
+
 // DiscoverVtctld is part of the discover.Discovery interface.
 func (d *Fake) DiscoverVtctld(ctx context.Context, tags []string) (*vtadminpb.Vtctld, error) {
 	vtctlds, err := d.DiscoverVtctlds(ctx, tags)