Transport: Add connection pool

This patch adds support for the "connection pool" feature, standard in official Elasticsearch clients, which maintains state for node connections passed in configuration, and removes connections with failures from the list of live connections. It periodically adds them back to the list, with an exponentional backoff delay.

The main parts of the implementation are contained in the connection.go file. The implementation short-circuits the connection pooling when only a single URL has been passed in configuration — common when using Elastic Cloud or a proxy/load balancer in front of Elasticsearch. When multiple URLs are passed, they are added to the list of live connections, and returned in a round-robin fashion from the Next() method. When the connection fails to respond to a request, it is removed from the list of live connections, put into the list of dead connections, and scheduled for a "resurrect" attempt. In case no live connection is available, a dead connection is forcibly resurrected. A sync.Mutex is used for synchronization.

For easier debugging, the implementation allows to use the EnableDebugLogger option to print debugging statements related to connection management to STDOUT.

This patch also adds support for collecting and getting client metrics: number of requests and failures, statistics for specific response codes, and the contents of the live/dead connection lists. An example for usage with the expvar package is provided in _examples/instrumentation/expvar.go. Integration with APM and other instrumentation tools will be researched and documented in future.

Closes #95

(cherry picked from commit 365fc2458d4135cdd9613cb6bcb772d41e81522f)

Author: karmi

_examples/instrumentation/Makefile

@@ -1,4 +1,5 @@
 test:
+	go build -o /dev/null expvar.go
 	go build -o /dev/null opencensus.go
 	go build -o /dev/null apmelasticsearch.go
 

_examples/instrumentation/expvar.go

@@ -0,0 +1,132 @@
+// Licensed to Elasticsearch B.V. under one or more agreements.
+// Elasticsearch B.V. licenses this file to you under the Apache 2.0 License.
+// See the LICENSE file in the project root for more information.
+
+package main
+
+import (
+	"expvar"
+	"fmt"
+	"io"
+	"log"
+	"os"
+	"os/signal"
+	"runtime"
+	"strings"
+	"syscall"
+	"time"
+
+	// Import the "expvar" and "pprof" package >>>>>>>>>>
+
+	"net/http"
+	_ "net/http/pprof"
+	// <<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<
+
+	"golang.org/x/crypto/ssh/terminal"
+
+	"github.com/elastic/go-elasticsearch/v8"
+	"github.com/elastic/go-elasticsearch/v8/estransport"
+)
+
+var (
+	tWidth, _, _ = terminal.GetSize(int(os.Stdout.Fd()))
+)
+
+func init() {
+	runtime.SetMutexProfileFraction(10)
+}
+
+func main() {
+	log.SetFlags(0)
+
+	ticker := time.NewTicker(time.Second)
+	defer ticker.Stop()
+
+	aborted := make(chan os.Signal)
+	signal.Notify(aborted, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
+
+	go func() {
+		<-aborted
+
+		log.Println("\nDone!\n")
+		os.Exit(0)
+	}()
+
+	log.Println(strings.Repeat("─", tWidth))
+	log.Println("Open <http://localhost:6060/debug/vars> to see all exported variables.")
+	log.Println(strings.Repeat("─", tWidth))
+
+	// Start the debug server >>>>>>>>>>>>>>>>>>>>>>>>>>>
+	go func() { log.Fatalln(http.ListenAndServe("localhost:6060", nil)) }()
+	// <<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<
+
+	for i := 1; i <= 2; i++ {
+		go func(i int) {
+			log.Printf("==> Starting server on <localhost:1000%d>", i)
+			if err := http.ListenAndServe(
+				fmt.Sprintf("localhost:1000%d", i),
+				http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { io.WriteString(w, "OK\n") }),
+			); err != nil && err != http.ErrServerClosed {
+				log.Fatalf("Unable to start server: %s", err)
+			}
+		}(i)
+	}
+
+	es, err := elasticsearch.NewClient(
+		elasticsearch.Config{
+			Addresses: []string{
+				"http://localhost:10001",
+				"http://localhost:10002",
+				"http://localhost:10003",
+			},
+
+			Logger:            &estransport.ColorLogger{Output: os.Stdout},
+			DisableRetry:      true,
+			EnableDebugLogger: true,
+
+			// Enable metric collection >>>>>>>>>>>>>>>>>>>>>>>>>
+			EnableMetrics: true,
+			// <<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<
+		})
+	if err != nil {
+		log.Fatal("ERROR: %s", err)
+	}
+
+	// Publish client metrics to expvar >>>>>>>>>>>>>>>>>
+	expvar.Publish("go-elasticsearch", expvar.Func(func() interface{} { m, _ := es.Metrics(); return m }))
+	// <<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<
+
+	for {
+		select {
+		case t := <-ticker.C:
+
+			go func() {
+				if res, _ := es.Info(); res != nil {
+					res.Body.Close()
+				}
+			}()
+
+			go func() {
+				if res, _ := es.Cluster.Health(); res != nil {
+					res.Body.Close()
+				}
+			}()
+
+			if t.Second()%5 == 0 {
+				m, err := es.Metrics()
+				if err != nil {
+					log.Printf("\x1b[31;1mUnable to get metrics: %s\x1b[0m", err)
+					continue
+				}
+				log.Println("███", fmt.Sprintf("\x1b[1m%s\x1b[0m", "Metrics"), strings.Repeat("█", tWidth-12))
+				log.Printf(
+					""+
+						"    \x1b[2mRequests:  \x1b[0m %d\n"+
+						"    \x1b[2mFailures:  \x1b[0m %d\n"+
+						"    \x1b[2mLive nodes:\x1b[0m %s",
+					m.Requests, m.Failures, m.Live)
+				log.Println(strings.Repeat("─", tWidth))
+			}
+		}
+	}
+}

elasticsearch.go

@@ -42,6 +42,9 @@ type Config struct {
 	EnableRetryOnTimeout bool  // Default: false.
 	MaxRetries           int   // Default: 3.
 
+	EnableMetrics     bool // Enable the metrics collection.
+	EnableDebugLogger bool // Enable the debug logging.
+
 	RetryBackoff func(attempt int) time.Duration // Optional backoff duration. Default: nil.
 
 	Transport http.RoundTripper  // The HTTP transport object.
@@ -130,6 +133,9 @@ func NewClient(cfg Config) (*Client, error) {
 		MaxRetries:           cfg.MaxRetries,
 		RetryBackoff:         cfg.RetryBackoff,
 
+		EnableMetrics:     cfg.EnableMetrics,
+		EnableDebugLogger: cfg.EnableDebugLogger,
+
 		Transport: cfg.Transport,
 		Logger:    cfg.Logger,
 	})
@@ -143,6 +149,15 @@ func (c *Client) Perform(req *http.Request) (*http.Response, error) {
 	return c.Transport.Perform(req)
 }
 
+// Metrics returns the client metrics.
+//
+func (c *Client) Metrics() (estransport.Metrics, error) {
+	if mt, ok := c.Transport.(estransport.Measurable); ok {
+		return mt.Metrics()
+	}
+	return estransport.Metrics{}, errors.New("transport is missing method Metrics()")
+}
+
 // addrsFromEnvironment returns a list of addresses by splitting
 // the ELASTICSEARCH_URL environment variable with comma, or an empty list.
 //

elasticsearch_internal_test.go

@@ -279,3 +279,16 @@ func TestVersion(t *testing.T) {
 		t.Error("Version is empty")
 	}
 }
+
+func TestClientMetrics(t *testing.T) {
+	c, _ := NewClient(Config{EnableMetrics: true})
+
+	m, err := c.Metrics()
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	if m.Requests != 0 {
+		t.Errorf("Unexpected output: %s", m)
+	}
+}