[DRAFT] Add support for retrying requests

[karmi]

This patch adds a draft of support for retrying the requests automatically by the client.

[axw]

This won't work, you're just returning the same body over and over. The first attempt will consume it, and then it'll be empty on retries. Instead you'd need to modify req.Body so that it reads into a buffer, and then have GetBody use that buffer. Basically, move the dupReqBodyForLog logic here.

[axw]

if err != nil, this will panic. You should just do the usual if err != nil { return nil, err } here.

[karmi]

Fixed in c78709c.

[axw]

the idea behind using GetBody was to get rid of this

[axw]

if i := 0; i < maxRetries; i++ { ?

[axw]

here you would do something like:

if req.Body != nil && req.Body != http.NoBody {
    body, err := req.GetBody()
    if err != nil {
        return nil, err
    }
    req.Body = body
}

[axw]

You can use req.GetBody here:

if req.Body != nil && req.Body != http.NoBody {
    body, err := req.GetBody()
    if err != nil {
        return nil, err
    }
    dupReqBodyForLog = body
}

Or you could just let logRoundTrip do that.

[karmi]

Executable example:

//+build ignore

package main

import (
	"context"
	"fmt"
	"log"
	"math"
	"net"
	"net/http"
	"os"
	"os/signal"
	"strings"
	"time"

	"github.com/elastic/go-elasticsearch/v8"
	"github.com/elastic/go-elasticsearch/v8/esapi"
	"github.com/elastic/go-elasticsearch/v8/estransport"

	"github.com/cenkalti/backoff"
)

var (
	_ = fmt.Print
	_ = context.WithTimeout
	_ = math.Exp
	_ = strings.NewReader
	_ = http.DefaultClient
)

func main() {
	log.SetFlags(0)

	ticker := time.NewTicker(time.Second)
	defer ticker.Stop()

	aborted := make(chan os.Signal)
	signal.Notify(aborted, os.Interrupt)

	retryBackoff := backoff.NewExponentialBackOff()
	retryBackoff.InitialInterval = time.Second
	_ = retryBackoff

	cfg := elasticsearch.Config{
		Addresses: []string{
			"http://localhost:9200",
			// "http://localhost:1000",
			// "http://localhost:2000",
			// "http://localhost:3000",
			"http://localhost:4000",
			"http://localhost:5000",
			// "http://localhost:6000",
			// "http://localhost:7000",
			// "http://localhost:8000",
			// "http://localhost:9200",
			// "http://localhost:9000",
			// "http://localhost:9201",
			// "http://localhost:9202",
		},

		Logger: &estransport.ColorLogger{
			Output:             os.Stdout,
			EnableRequestBody:  true,
			EnableResponseBody: true,
		},

		// RetryOnStatus:        []int{404},
		// EnableRetryOnTimeout: true,
		// MaxRetries: 10,

		// RetryBackoff: func(i int) time.Duration {
		// 	d := time.Duration(i) * time.Second
		// 	fmt.Printf("Attempt: %d | Sleeping for %s...\n", i, d)
		// 	return d
		// },

		// RetryBackoff: func(i int) time.Duration {
		// 	d := time.Duration(math.Exp2(float64(i))) * time.Second
		// 	fmt.Printf("Attempt: %d | Sleeping for %s...\n", i, d)
		// 	return d
		// },

		RetryBackoff: func(i int) time.Duration {
			if i == 1 {
				retryBackoff.Reset()
			}
			d := retryBackoff.NextBackOff()
			fmt.Printf("Attempt: %d | Sleeping for %s...\n", i, d)
			return d
		},

		// Transport: &http.Transport{
		// 	ResponseHeaderTimeout: time.Millisecond,
		// },
	}

	es, err := elasticsearch.NewClient(cfg)
	if err != nil {
		log.Fatalf("Error creating the client: %s", err)
	}
	log.Println("Client ready with URLs:", es.Transport.(*estransport.Client).URLs())

	for {
		select {
		case <-aborted:
			log.Println("\nDone!")
			return
		case <-ticker.C:
			var (
				res *esapi.Response
				err error
			)
			// res, err = es.Info()

			// ctx, cancel := context.WithTimeout(context.Background(), time.Nanosecond)
			// defer cancel()
			// res, err = es.Info(es.Info.WithContext(ctx))

			// es.Search(es.Search.WithTimeout(time.Nanosecond))

			// res, err = es.Get("test", "MISSING")

			res, err = es.Index("test", strings.NewReader(`{"foo":"bar"}`))

			if err != nil {
				if e, ok := err.(net.Error); ok {
					log.Fatalf("Error getting response: [%T]: %s (timeout:%v)", e, e, e.Timeout())
				} else {
					log.Fatalf("Error getting response: [%T]: %s", err, err)
				}
			}
			if res.IsError() {
				log.Fatalf("Error response: %s", res.Status())
			}
		}
	}
}