Update clientutil.Backend interface

proxy/proxy.go

 package proxy
 
 import (
+	"context"
 	"crypto/tls"
 	"errors"
 	"fmt"
 	"io/ioutil"
+	"math/rand"
+	"net"
 	"net/http"
 	"net/http/httputil"
 	"net/url"
+	"time"
 
 	"git.autistici.org/ai3/go-common/clientutil"
 	"github.com/gorilla/mux"
@@ -15,6 +19,9 @@ import (
 	"git.autistici.org/id/go-sso/httpsso"
 )
 
+// RNG for the random backend selector.
+var rnd = rand.New(rand.NewSource(time.Now().UnixNano()))
+
 // Backend defines a single-host HTTP proxy to a set of upstream
 // backends.
 type Backend struct {
@@ -54,7 +61,16 @@ func (b *Backend) newHandler(ssow *httpsso.SSOWrapper) (http.Handler, error) {
 		// regardless of the HTTP request details.
 		tlsConfig.ServerName = b.ServerName
 	}
-	proxy.Transport = clientutil.NewTransport(b.Upstream, tlsConfig, nil)
+	proxy.Transport = &http.Transport{
+		TLSClientConfig: tlsConfig,
+		DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
+			// The simplest possible backend
+			// load-balancer: we just pick a random
+			// upstream target to connect to every time.
+			var d net.Dialer
+			return d.DialContext(ctx, network, b.Upstream[rnd.Intn(len(b.Upstream))])
+		},
+	}
 
 	h := ssow.Wrap(proxy, b.Host+"/", b.AllowedGroups)
 	return h, nil

vendor/git.autistici.org/ai3/go-common/clientutil/backend.go

 package clientutil
 
 import (
-	"crypto/tls"
-	"fmt"
-	"net/http"
-	"net/url"
-	"sync"
-	"time"
+	"context"
 )
 
-// BackendConfig specifies the configuration to access a service.
+// BackendConfig specifies the configuration of a service backend.
 //
 // Services with multiple backends can be replicated or partitioned,
 // depending on a configuration switch, making it a deployment-time
@@ -18,102 +13,30 @@ import (
 // 'shard' parameter on their APIs.
 type BackendConfig struct {
 	URL       string           `yaml:"url"`
-	Sharded   bool             `yaml:"sharded"`
 	TLSConfig *TLSClientConfig `yaml:"tls_config"`
+	Sharded   bool             `yaml:"sharded"`
+	Debug     bool             `yaml:"debug"`
 }
 
 // Backend is a runtime class that provides http Clients for use with
 // a specific service backend. If the service can't be partitioned,
-// pass an empty string to the Client method.
+// pass an empty string to the Call method.
 type Backend interface {
-	// URL for the service for a specific shard.
-	URL(string) string
+	// Call a remote method. The sharding behavior is the following:
+	//
+	// Services that support sharding (partitioning) should always
+	// include the shard ID in their Call() requests. Users can
+	// then configure backends to be sharded or not in their
+	// Config. When invoking Call with a shard ID on a non-sharded
+	// service, the shard ID is simply ignored. Invoking Call
+	// *without* a shard ID on a sharded service is an error.
+	Call(context.Context, string, string, interface{}, interface{}) error
 
-	// Client that can be used to make a request to the service.
-	Client(string) *http.Client
+	// Close all resources associated with the backend.
+	Close()
 }
 
 // NewBackend returns a new Backend with the given config.
 func NewBackend(config *BackendConfig) (Backend, error) {
-	u, err := url.Parse(config.URL)
-	if err != nil {
-		return nil, err
-	}
-
-	var tlsConfig *tls.Config
-	if config.TLSConfig != nil {
-		tlsConfig, err = config.TLSConfig.TLSConfig()
-		if err != nil {
-			return nil, err
-		}
-	}
-
-	if config.Sharded {
-		return &replicatedClient{
-			u: u,
-			c: newHTTPClient(u, tlsConfig),
-		}, nil
-	}
-	return &shardedClient{
-		baseURL:   u,
-		tlsConfig: tlsConfig,
-		urls:      make(map[string]*url.URL),
-		shards:    make(map[string]*http.Client),
-	}, nil
-}
-
-type replicatedClient struct {
-	c *http.Client
-	u *url.URL
-}
-
-func (r *replicatedClient) Client(_ string) *http.Client { return r.c }
-func (r *replicatedClient) URL(_ string) string          { return r.u.String() }
-
-type shardedClient struct {
-	baseURL   *url.URL
-	tlsConfig *tls.Config
-	mx        sync.Mutex
-	urls      map[string]*url.URL
-	shards    map[string]*http.Client
-}
-
-func (s *shardedClient) getShardURL(shard string) *url.URL {
-	if shard == "" {
-		return s.baseURL
-	}
-	u, ok := s.urls[shard]
-	if !ok {
-		var tmp = *s.baseURL
-		tmp.Host = fmt.Sprintf("%s.%s", shard, tmp.Host)
-		u = &tmp
-		s.urls[shard] = u
-	}
-	return u
-}
-
-func (s *shardedClient) URL(shard string) string {
-	s.mx.Lock()
-	defer s.mx.Unlock()
-	return s.getShardURL(shard).String()
-}
-
-func (s *shardedClient) Client(shard string) *http.Client {
-	s.mx.Lock()
-	defer s.mx.Unlock()
-
-	client, ok := s.shards[shard]
-	if !ok {
-		u := s.getShardURL(shard)
-		client = newHTTPClient(u, s.tlsConfig)
-		s.shards[shard] = client
-	}
-	return client
-}
-
-func newHTTPClient(u *url.URL, tlsConfig *tls.Config) *http.Client {
-	return &http.Client{
-		Transport: NewTransport([]string{u.Host}, tlsConfig, nil),
-		Timeout:   30 * time.Second,
-	}
+	return newBalancedBackend(config, defaultResolver)
 }

vendor/git.autistici.org/ai3/go-common/clientutil/balancer.go

+package clientutil
+
+import (
+	"bytes"
+	"context"
+	"crypto/tls"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log"
+	"math/rand"
+	"net/http"
+	"net/url"
+	"os"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/cenkalti/backoff"
+)
+
+// Our own narrow logger interface.
+type logger interface {
+	Printf(string, ...interface{})
+}
+
+// A nilLogger is used when Config.Debug is false.
+type nilLogger struct{}
+
+func (l nilLogger) Printf(_ string, _ ...interface{}) {}
+
+// Parameters that define the exponential backoff algorithm used.
+var (
+	ExponentialBackOffInitialInterval = 100 * time.Millisecond
+	ExponentialBackOffMultiplier      = 1.4142
+)
+
+// newExponentialBackOff creates a backoff.ExponentialBackOff object
+// with our own default values.
+func newExponentialBackOff() *backoff.ExponentialBackOff {
+	b := backoff.NewExponentialBackOff()
+	b.InitialInterval = ExponentialBackOffInitialInterval
+	b.Multiplier = ExponentialBackOffMultiplier
+
+	// Set MaxElapsedTime to 0 because we expect the overall
+	// timeout to be dictated by the request Context.
+	b.MaxElapsedTime = 0
+
+	return b
+}
+
+// Balancer for HTTP connections. It will round-robin across available
+// backends, trying to avoid ones that are erroring out, until one
+// succeeds or returns a permanent error.
+//
+// This object should not be used for load balancing of individual
+// HTTP requests: it doesn't do anything smart beyond trying to avoid
+// broken targets. It's meant to provide a *reliable* connection to a
+// set of equivalent services for HA purposes.
+type balancedBackend struct {
+	*backendTracker
+	*transportCache
+	baseURI  *url.URL
+	sharded  bool
+	resolver resolver
+	log      logger
+}
+
+func newBalancedBackend(config *BackendConfig, resolver resolver) (*balancedBackend, error) {
+	u, err := url.Parse(config.URL)
+	if err != nil {
+		return nil, err
+	}
+
+	var tlsConfig *tls.Config
+	if config.TLSConfig != nil {
+		tlsConfig, err = config.TLSConfig.TLSConfig()
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	var logger logger = &nilLogger{}
+	if config.Debug {
+		logger = log.New(os.Stderr, fmt.Sprintf("backend %s: ", u.Host), 0)
+	}
+	return &balancedBackend{
+		backendTracker: newBackendTracker(u.Host, resolver, logger),
+		transportCache: newTransportCache(tlsConfig),
+		sharded:        config.Sharded,
+		baseURI:        u,
+		resolver:       resolver,
+		log:            logger,
+	}, nil
+}
+
+// Call the backend. Makes an HTTP POST request to the specified uri,
+// with a JSON-encoded request body. It will attempt to decode the
+// response body as JSON.
+func (b *balancedBackend) Call(ctx context.Context, shard, path string, req, resp interface{}) error {
+	data, err := json.Marshal(req)
+	if err != nil {
+		return err
+	}
+
+	var tg targetGenerator = b.backendTracker
+	if b.sharded && shard != "" {
+		tg = newShardedGenerator(shard, b.baseURI.Host, b.resolver)
+	}
+	seq := newSequence(tg)
+	b.log.Printf("%016x: initialized", seq.ID())
+
+	var httpResp *http.Response
+	err = backoff.Retry(func() error {
+		req, rerr := b.newJSONRequest(path, shard, data)
+		if rerr != nil {
+			return rerr
+		}
+		httpResp, rerr = b.do(ctx, seq, req)
+		return rerr
+	}, backoff.WithContext(newExponentialBackOff(), ctx))
+	if err != nil {
+		return err
+	}
+	defer httpResp.Body.Close() // nolint
+
+	if httpResp.Header.Get("Content-Type") != "application/json" {
+		return errors.New("not a JSON response")
+	}
+
+	if resp == nil {
+		return nil
+	}
+	return json.NewDecoder(httpResp.Body).Decode(resp)
+}
+
+// Return the URI to be used for the request. This is used both in the
+// Host HTTP header and as the TLS server name used to pick a server
+// certificate (if using TLS).
+func (b *balancedBackend) getURIForRequest(shard, path string) string {
+	u := *b.baseURI
+	if b.sharded && shard != "" {
+		u.Host = fmt.Sprintf("%s.%s", shard, u.Host)
+	}
+	u.Path = appendPath(u.Path, path)
+	return u.String()
+}
+
+// Build a http.Request object.
+func (b *balancedBackend) newJSONRequest(path, shard string, data []byte) (*http.Request, error) {
+	req, err := http.NewRequest("POST", b.getURIForRequest(shard, path), bytes.NewReader(data))
+	if err != nil {
+		return nil, err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Content-Length", strconv.FormatInt(int64(len(data)), 10))
+	return req, nil
+}
+
+// Select a new target from the given sequence and send the request to
+// it. Wrap HTTP errors in a RemoteError object.
+func (b *balancedBackend) do(ctx context.Context, seq *sequence, req *http.Request) (resp *http.Response, err error) {
+	target, terr := seq.Next()
+	if terr != nil {
+		return
+	}
+
+	b.log.Printf("sequence %016x: connecting to %s", seq.ID(), target)
+	client := &http.Client{
+		Transport: b.transportCache.getTransport(target),
+	}
+	resp, err = client.Do(req.WithContext(ctx))
+	if err == nil && resp.StatusCode != 200 {
+		err = remoteErrorFromResponse(resp)
+		if !isStatusTemporary(resp.StatusCode) {
+			err = backoff.Permanent(err)
+		}
+		resp.Body.Close() // nolint
+		resp = nil
+	}
+
+	seq.Done(target, err)
+	return
+}
+
+var errNoTargets = errors.New("no available backends")
+
+type targetGenerator interface {
+	getTargets() []string
+	setStatus(string, bool)
+}
+
+// A replicatedSequence repeatedly iterates over available backends in order of
+// preference. Once in a while it refreshes its list of available
+// targets.
+type sequence struct {
+	id      uint64
+	tg      targetGenerator
+	targets []string
+	pos     int
+}
+
+func newSequence(tg targetGenerator) *sequence {
+	return &sequence{
+		id:      rand.Uint64(),
+		tg:      tg,
+		targets: tg.getTargets(),
+	}
+}
+
+func (s *sequence) ID() uint64 { return s.id }
+
+func (s *sequence) reloadTargets() {
+	targets := s.tg.getTargets()
+	if len(targets) > 0 {
+		s.targets = targets
+		s.pos = 0
+	}
+}
+
+// Next returns the next target.
+func (s *sequence) Next() (t string, err error) {
+	if s.pos >= len(s.targets) {
+		s.reloadTargets()
+		if len(s.targets) == 0 {
+			err = errNoTargets
+			return
+		}
+	}
+	t = s.targets[s.pos]
+	s.pos++
+	return
+}
+
+func (s *sequence) Done(t string, err error) {
+	s.tg.setStatus(t, err == nil)
+}
+
+// A shardedGenerator returns a single sharded target to a sequence.
+type shardedGenerator struct {
+	id    uint64
+	addrs []string
+}
+
+func newShardedGenerator(shard, base string, resolver resolver) *shardedGenerator {
+	return &shardedGenerator{
+		id:    rand.Uint64(),
+		addrs: resolver.ResolveIP(fmt.Sprintf("%s.%s", shard, base)),
+	}
+}
+
+func (g *shardedGenerator) getTargets() []string       { return g.addrs }
+func (g *shardedGenerator) setStatus(_ string, _ bool) {}
+
+// Concatenate two URI paths.
+func appendPath(a, b string) string {
+	if strings.HasSuffix(a, "/") && strings.HasPrefix(b, "/") {
+		return a + b[1:]
+	}
+	return a + b
+}
+
+// Some HTTP status codes are treated are temporary errors.
+func isStatusTemporary(code int) bool {
+	switch code {
+	case http.StatusTooManyRequests, http.StatusBadGateway, http.StatusServiceUnavailable, http.StatusGatewayTimeout:
+		return true
+	default:
+		return false
+	}
+}

vendor/git.autistici.org/ai3/go-common/clientutil/dns.go

+package clientutil
+
+import (
+	"log"
+	"net"
+	"sync"
+	"time"
+
+	"golang.org/x/sync/singleflight"
+)
+
+type resolver interface {
+	ResolveIP(string) []string
+}
+
+type dnsResolver struct{}
+
+func (r *dnsResolver) ResolveIP(hostport string) []string {
+	var resolved []string
+	host, port, err := net.SplitHostPort(hostport)
+	if err != nil {
+		log.Printf("error parsing %s: %v", hostport, err)
+		return nil
+	}
+	hostIPs, err := net.LookupIP(host)
+	if err != nil {
+		log.Printf("error resolving %s: %v", host, err)
+		return nil
+	}
+	for _, ip := range hostIPs {
+		resolved = append(resolved, net.JoinHostPort(ip.String(), port))
+	}
+	return resolved
+}
+
+var defaultResolver = newDNSCache(&dnsResolver{})
+
+type cacheDatum struct {
+	addrs    []string
+	deadline time.Time
+}
+
+type dnsCache struct {