Autodiscovery

go.mod

@@ -27,30 +27,64 @@ require (
 	golang.org/x/net v0.8.0
 	gopkg.in/natefinch/lumberjack.v2 v2.2.1
 	gopkg.in/yaml.v2 v2.4.0
+	k8s.io/api v0.27.2
+	k8s.io/apimachinery v0.27.2
+	k8s.io/client-go v0.27.2
+	k8s.io/utils v0.0.0-20230209194617-a36077c30491
 )
 
 require (
 	github.com/AndreasBriese/bbloom v0.0.0-20190825152654-46b345b51c96 // indirect
 	github.com/alicebob/gopher-json v0.0.0-20230218143504-906a9b012302 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/dgraph-io/ristretto v0.1.1 // indirect
 	github.com/dustin/go-humanize v1.0.1 // indirect
+	github.com/emicklei/go-restful/v3 v3.9.0 // indirect
+	github.com/evanphx/json-patch v4.12.0+incompatible // indirect
 	github.com/go-logfmt/logfmt v0.6.0 // indirect
 	github.com/go-logr/logr v1.2.3 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/go-openapi/jsonpointer v0.19.6 // indirect
+	github.com/go-openapi/jsonreference v0.20.1 // indirect
+	github.com/go-openapi/swag v0.22.3 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/glog v1.1.1 // indirect
 	github.com/golang/protobuf v1.5.3 // indirect
 	github.com/gomodule/redigo v1.8.9 // indirect
+	github.com/google/gnostic v0.5.7-v3refs // indirect
+	github.com/google/go-cmp v0.5.9 // indirect
+	github.com/google/gofuzz v1.1.0 // indirect
+	github.com/google/uuid v1.3.0 // indirect
+	github.com/imdario/mergo v0.3.6 // indirect
+	github.com/josharian/intern v1.0.0 // indirect
+	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/mailru/easyjson v0.7.7 // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
+	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
+	github.com/modern-go/reflect2 v1.0.2 // indirect
+	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/openzipkin/zipkin-go v0.4.1 // indirect
 	github.com/philhofer/fwd v1.1.2 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_model v0.3.0 // indirect
 	github.com/prometheus/common v0.42.0 // indirect
 	github.com/prometheus/procfs v0.9.0 // indirect
+	github.com/spf13/pflag v1.0.5 // indirect
 	github.com/yuin/gopher-lua v1.1.0 // indirect
+	golang.org/x/oauth2 v0.5.0 // indirect
 	golang.org/x/sys v0.6.0 // indirect
+	golang.org/x/term v0.6.0 // indirect
+	golang.org/x/text v0.8.0 // indirect
+	golang.org/x/time v0.0.0-20220210224613-90d013bbcef8 // indirect
+	google.golang.org/appengine v1.6.7 // indirect
 	google.golang.org/protobuf v1.30.0 // indirect
+	gopkg.in/inf.v0 v0.9.1 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
+	k8s.io/klog/v2 v2.90.1 // indirect
+	k8s.io/kube-openapi v0.0.0-20230501164219-8b0f38b5fd1f // indirect
+	sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect
+	sigs.k8s.io/structured-merge-diff/v4 v4.2.3 // indirect
+	sigs.k8s.io/yaml v1.3.0 // indirect
 )

pkg/backends/alb/client.go

@@ -117,7 +117,7 @@ func StartALBPools(clients backends.Backends, hcs healthcheck.StatusLookup) erro
 			if err != nil {
 				return err
 			}
-		}
+		} // else if template
 	}
 	return nil
 }
@@ -167,13 +167,15 @@ func (c *Client) ValidateAndStartPool(clients backends.Backends, hcs healthcheck
 		return fmt.Errorf("invalid mechanism name [%s] in backend [%s]", o.MechanismName, c.Name())
 	}
 	targets := make([]*pool.Target, 0, len(o.Pool))
-	for _, n := range o.Pool {
-		tc, ok := clients[n]
-		if !ok {
-			return fmt.Errorf("invalid pool member name [%s] in backend [%s]", n, c.Name())
+	if len(o.Pool) > 0 {
+		for _, n := range o.Pool {
+			tc, ok := clients[n]
+			if !ok {
+				return fmt.Errorf("invalid pool member name [%s] in backend [%s]", n, c.Name())
+			}
+			hc := hcs[n]
+			targets = append(targets, pool.NewTarget(tc.Router(), hc))
 		}
-		hc, _ := hcs[n]
-		targets = append(targets, pool.NewTarget(tc.Router(), hc))
 	}
 	c.pool = pool.New(m, targets, o.HealthyFloor)
 	return nil

pkg/backends/alb/discovery/clients/clients.go

@@ -0,0 +1,16 @@
+package clients
+
+import (
+	"errors"
+
+	"github.com/trickstercache/trickster/v2/pkg/backends/alb/discovery"
+	"github.com/trickstercache/trickster/v2/pkg/backends/alb/discovery/clients/kubernetes"
+)
+
+func New(provider string) (discovery.Client, error) {
+	switch provider {
+	case "kubernetes":
+		return &kubernetes.KubeClient{}, nil
+	}
+	return nil, errors.New("unrecognized provider " + provider)
+}

pkg/backends/alb/discovery/clients/kubernetes/kubernetes.go

@@ -0,0 +1,82 @@
+package kubernetes
+
+import (
+	"context"
+	"errors"
+
+	"github.com/trickstercache/trickster/v2/pkg/backends/alb/discovery"
+	do "github.com/trickstercache/trickster/v2/pkg/backends/alb/discovery/options"
+	"github.com/trickstercache/trickster/v2/pkg/kube"
+)
+
+type KubeClient struct{}
+
+func doInternal(ctx context.Context, kc *kube.Client, opts *do.Options) ([]discovery.Result, error) {
+	pods, err := kube.Pods(ctx, kc, opts.Selector)
+	if err != nil {
+		return nil, err
+	}
+	ress := make([]discovery.Result, len(pods))
+	for i, pod := range pods {
+		ress[i] = discovery.Result{
+			Name: pod.Meta.Name,
+			URL:  pod.Address,
+		}
+	}
+	return ress, nil
+}
+
+func doExternal(ctx context.Context, kc *kube.Client, opts *do.Options) ([]discovery.Result, error) {
+	svcs, err := kube.Services(ctx, kc, opts.Selector)
+	if err != nil {
+		return nil, err
+	}
+	ings, err := kube.Ingresses(ctx, kc, &kube.Selector{})
+	if err != nil {
+		return nil, err
+	}
+	// Find every ingress path that matches a service result
+	out := make([]discovery.Result, 0)
+	// Iterate services
+	for _, svc := range svcs {
+		host := "localhost"
+		// Iterate ingresses
+		for _, ing := range ings {
+			// Iterate rules
+			for _, rule := range ing.Rules {
+				scheme := rule.Scheme
+				// Iterate paths. We can go deeper!
+				for _, path := range rule.Paths {
+					if path.Backend == nil {
+						continue
+					} else if path.Backend.Name == svc.Meta.Name {
+						if rule.Host != "" {
+							host = rule.Host
+						}
+						out = append(out, discovery.Result{
+							Name:   svc.Meta.Name,
+							Scheme: scheme,
+							URL:    host + path.Path,
+						})
+					}
+				}
+			}
+		}
+	}
+	return out, nil
+}
+
+func (c *KubeClient) Execute(ctx context.Context, opts *do.Options) ([]discovery.Result, error) {
+	if opts.Provider != "kubernetes" || opts.Kubernetes == nil {
+		return nil, errors.New("KubeClient must be provided a set of options for Kubernetes service discovery")
+	}
+	kc, err := kube.New(opts.Kubernetes)
+	if err != nil {
+		return nil, err
+	}
+	if opts.Kubernetes.InCluster {
+		return doInternal(ctx, kc, opts)
+	} else {
+		return doExternal(ctx, kc, opts)
+	}
+}

pkg/backends/alb/discovery/discovery.go

@@ -0,0 +1,78 @@
+package discovery
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	do "github.com/trickstercache/trickster/v2/pkg/backends/alb/discovery/options"
+	bo "github.com/trickstercache/trickster/v2/pkg/backends/options"
+)
+
+type Result struct {
+	Name   string
+	Scheme string
+	URL    string
+}
+
+// Clients act on a set of options to fetch valid backend origins.
+type Client interface {
+	Execute(ctx context.Context, opts *do.Options) ([]Result, error)
+}
+
+// Discover services using the provided client.
+// This function takes a few steps given the provided inputs:
+//   - Ensure that the discovery options match with the provided client
+//   - Use the client to find a set of named origin URLs
+//   - Use the template options provided to instantiate more options based on targets
+//   - Return the resulting instantiated options
+func DiscoverServices(ctx context.Context, c Client, opts *do.Options, bs map[string]*bo.Options) ([]*bo.Options, error) {
+	// Arrange template mapping
+	templates := make(map[string]*bo.Options)
+	for _, b := range bs {
+		if !b.IsTemplate {
+			continue
+		}
+		for tFrom, tTo := range opts.Targets {
+			if b.Name == tTo {
+				templates[tFrom] = b.Clone()
+			}
+		}
+	}
+	ress, err := c.Execute(ctx, opts)
+	if err != nil {
+		return nil, err
+	}
+	out := make([]*bo.Options, len(ress))
+	for i, res := range ress {
+		t, ok := templates[res.Name]
+		if !ok {
+			t, ok = templates["default"]
+			if !ok {
+				return nil, fmt.Errorf("resolved autodiscovery but could not find template %s", res.Name)
+			}
+		}
+		t = t.Clone()
+		t.IsTemplate = false
+		if res.Name != "" {
+			t.Name = res.Name
+		} else {
+			t.Name = fmt.Sprintf("%s_%d", t.Name, i)
+		}
+		// scheme
+		t.Scheme = res.Scheme
+		t.OriginURL = res.URL
+		hostEnd := strings.Index(res.URL, "/")
+		if hostEnd == -1 {
+			hostEnd = len(res.URL)
+		} else {
+			t.PathPrefix = res.URL[hostEnd:]
+		}
+		t.Host = res.URL[:hostEnd]
+		if err != nil {
+			return nil, err
+		}
+		out[i] = t
+	}
+	return out, nil
+}

pkg/backends/alb/discovery/options/options.go

@@ -0,0 +1,13 @@
+package options
+
+import (
+	"github.com/trickstercache/trickster/v2/pkg/kube"
+	ko "github.com/trickstercache/trickster/v2/pkg/kube/options"
+)
+
+type Options struct {
+	Provider   string            `yaml:"provider"`
+	Kubernetes *ko.Options       `yaml:"kubernetes"`
+	Selector   *kube.Selector    `yaml:"selector"`
+	Targets    map[string]string `yaml:"targets"`
+}

pkg/backends/alb/discovery/options/options_test.go

@@ -0,0 +1,45 @@
+package options
+
+import (
+	"testing"
+
+	"gopkg.in/yaml.v3"
+)
+
+var testYaml = `
+provider: kubernetes
+kubernetes:
+  in_cluster: false
+  config_path: ~/.kubeconfig
+selector:
+  namespace: default
+  matchLabels:
+    app: prometheus
+targets:
+  prometheus: prom_mock
+`
+
+func TestKubernetes(t *testing.T) {
+	var opts *Options
+	err := yaml.Unmarshal([]byte(testYaml), &opts)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if opts.Provider != "kubernetes" {
+		t.Error("expected kubernetes")
+	}
+	if k8s := opts.Kubernetes; k8s == nil {
+		t.Error("expected kubernetes client")
+	} else if k8s.InCluster {
+		t.Error("expected out-of-cluster")
+	} else if k8s.ConfigPath != "~/.kubeconfig" {
+		t.Error("wrong configpath")
+	}
+	if sel := opts.Selector; sel == nil {
+		t.Error("expected selector")
+	} else if sel.Namespace != "default" {
+		t.Error("expected default")
+	} else if len(sel.MatchLabels) != 1 {
+		t.Errorf("expected populated matchLabels, got %v", sel.MatchLabels)
+	}
+}

pkg/backends/alb/options/options.go

@@ -20,6 +20,7 @@ import (
 	"errors"
 	"strings"
 
+	do "github.com/trickstercache/trickster/v2/pkg/backends/alb/discovery/options"
 	"github.com/trickstercache/trickster/v2/pkg/backends/providers"
 	"github.com/trickstercache/trickster/v2/pkg/util/copiers"
 	"github.com/trickstercache/trickster/v2/pkg/util/yamlx"
@@ -29,6 +30,8 @@ import (
 type Options struct {
 	// MechanismName indicates the name of the load balancing mechanism
 	MechanismName string `yaml:"mechanism,omitempty"`
+	// Discovery provides a mechanism for service discovery
+	Discovery *do.Options `yaml:"discovery,omitempty"`
 	// Pool provides the list of backend names to be used by the load balancer
 	Pool []string `yaml:"pool,omitempty"`
 	// HealthyFloor is the minimum health check status value to be considered Available in the pool
@@ -102,6 +105,10 @@ func SetDefaults(name string, options *Options, metadata yamlx.KeyLookup) (*Opti
 		o.Pool = options.Pool
 	}
 
+	if metadata.IsDefined("backends", name, "alb", "discovery") {
+		o.Discovery = options.Discovery
+	}
+
 	if metadata.IsDefined("backends", name, "alb", "mechanism") && options.MechanismName != "" {
 		o.MechanismName = options.MechanismName
 	}

pkg/backends/alb/pool/health.go

@@ -29,7 +29,10 @@ func (p *pool) checkHealth() {
 			p.mtx.Lock()
 			h := make([]http.Handler, 0, len(p.targets))
 			for _, t := range p.targets {
-				if t.hcStatus.Get() >= p.healthyFloor {
+				// Assume unchecked backends are healthy
+				if t.hcStatus == nil {
+					h = append(h, t.handler)
+				} else if t.hcStatus.Get() >= p.healthyFloor {
 					h = append(h, t.handler)
 				}
 			}

pkg/backends/alb/pool/pool.go

@@ -56,7 +56,9 @@ func New(mechanism Mechanism, targets []*Target, healthyFloor int) Pool {
 	p.ch <- true
 
 	for _, t := range targets {
-		t.hcStatus.RegisterSubscriber(p.ch)
+		if t.hcStatus != nil {
+			t.hcStatus.RegisterSubscriber(p.ch)
+		}
 	}
 
 	go p.checkHealth()