feat: singleflight to deduplicate GET requests (#418)

Co-authored-by: Andrey Karpov <[email protected]>

Author: byashimov

README.md

@@ -21,6 +21,8 @@ go get github.com/aiven/go-client-codegen
 | `AIVEN_USER_AGENT` | `string` | User Agent                     |
 | `AIVEN_DEBUG`      | `bool`   | Debug Output Flag (stderr)     |
 
+See all configuration options in [`client.go`](client.go).
+
 #### Via Constructor Options
 
 ```go

Taskfile.yml

@@ -40,4 +40,4 @@ tasks:
       - task: fmt-imports
   test:
     cmds:
-      - go test -v
+      - go test -race -count=1 -v ./...

client.go

@@ -16,8 +16,8 @@ import (
 	"time"
 
 	"golang.org/x/exp/slices"
+	"golang.org/x/sync/singleflight"
 
-	"github.com/hashicorp/go-multierror"
 	"github.com/hashicorp/go-retryablehttp"
 	"github.com/kelseyhightower/envconfig"
 	"github.com/rs/zerolog"
@@ -105,23 +105,28 @@ func NewClient(opts ...Option) (Client, error) {
 // Default values: 26 seconds
 // Changed values: 67 seconds
 type aivenClient struct {
-	Host         string        `envconfig:"AIVEN_WEB_URL" default:"https://api.aiven.io"`
-	UserAgent    string        `envconfig:"AIVEN_USER_AGENT" default:"aiven-go-client/v3"`
-	Token        string        `envconfig:"AIVEN_TOKEN"`
-	Debug        bool          `envconfig:"AIVEN_DEBUG"`
-	RetryMax     int           `envconfig:"AIVEN_CLIENT_RETRY_MAX" default:"6"`
-	RetryWaitMin time.Duration `envconfig:"AIVEN_CLIENT_RETRY_WAIT_MIN" default:"2s"`
-	RetryWaitMax time.Duration `envconfig:"AIVEN_CLIENT_RETRY_WAIT_MAX" default:"15s"`
-	logger       zerolog.Logger
-	doer         Doer
+	Host               string        `envconfig:"AIVEN_WEB_URL" default:"https://api.aiven.io"`
+	UserAgent          string        `envconfig:"AIVEN_USER_AGENT" default:"aiven-go-client/v3"`
+	Token              string        `envconfig:"AIVEN_TOKEN"`
+	Debug              bool          `envconfig:"AIVEN_DEBUG"`
+	RetryMax           int           `envconfig:"AIVEN_CLIENT_RETRY_MAX" default:"6"`
+	RetryWaitMin       time.Duration `envconfig:"AIVEN_CLIENT_RETRY_WAIT_MIN" default:"2s"`
+	RetryWaitMax       time.Duration `envconfig:"AIVEN_CLIENT_RETRY_WAIT_MAX" default:"15s"`
+	EnableSingleFlight bool          `envconfig:"AIVEN_CLIENT_ENABLE_SINGLE_FLIGHT" default:"true"`
+	logger             zerolog.Logger
+	doer               Doer
+	singleflight       singleflight.Group
 }
 
 // OperationIDKey is the key used to store the operation ID in the context.
 type OperationIDKey struct{}
 
 func (d *aivenClient) Do(ctx context.Context, operationID, method, path string, in any, query ...[2]string) ([]byte, error) {
 	ctx = context.WithValue(ctx, OperationIDKey{}, operationID)
-	var rsp *http.Response
+	queryString := fmtQuery(operationID, query...)
+
+	var statusCode int
+	var shared bool
 	var err error
 
 	if d.Debug {
@@ -133,54 +138,72 @@ func (d *aivenClient) Do(ctx context.Context, operationID, method, path string,
 			if err != nil {
 				event = d.logger.Error().Err(err)
 			} else {
-				event = d.logger.Info().Str("status", rsp.Status)
+				event = d.logger.Info()
 			}
 
 			event.Ctx(ctx).
 				Stringer("duration", end).
 				Str("operationID", operationID).
 				Str("method", method).
 				Str("path", path).
-				Str("query", fmtQuery(operationID, query...)).
+				Str("query", queryString).
+				Int("status_code", statusCode).
+				Bool("shared", shared).
 				Send()
 		}()
 	}
 
-	rsp, err = d.do(ctx, operationID, method, path, in, query...)
+	var body []byte
+	if d.EnableSingleFlight && (method == http.MethodGet || method == http.MethodHead || method == http.MethodOptions || method == http.MethodTrace) {
+		type result struct {
+			statusCode int
+			body       []byte
+		}
+		key := strings.Join([]string{method, d.Host, path, queryString}, "|")
+		v, serr, sh := d.singleflight.Do(key, func() (any, error) {
+			statusCode, body, err := d.do(ctx, method, path, in, queryString)
+			return result{statusCode: statusCode, body: body}, err
+		})
+		res := v.(result)
+		statusCode, body, err, shared = res.statusCode, res.body, serr, sh
+	} else {
+		statusCode, body, err = d.do(ctx, method, path, in, queryString)
+	}
 	if err != nil {
 		return nil, err
 	}
 
-	defer func() {
-		err = multierror.Append(rsp.Body.Close()).ErrorOrNil()
-	}()
-
-	return fromResponse(operationID, rsp)
+	return fromBytes(operationID, statusCode, body)
 }
 
-func (d *aivenClient) do(ctx context.Context, operationID, method, path string, in any, query ...[2]string) (*http.Response, error) {
+func (d *aivenClient) do(ctx context.Context, method, path string, in any, queryString string) (int, []byte, error) {
 	var body io.Reader
 
 	if !(in == nil || isEmpty(in)) {
 		b, err := json.Marshal(in)
 		if err != nil {
-			return nil, err
+			return 0, nil, err
 		}
 
 		body = bytes.NewBuffer(b)
 	}
 
 	req, err := http.NewRequestWithContext(ctx, method, d.Host+path, body)
 	if err != nil {
-		return nil, err
+		return 0, nil, err
 	}
 
 	req.Header.Set("Content-Type", "application/json")
 	req.Header.Set("User-Agent", d.UserAgent)
 	req.Header.Set("Authorization", "aivenv1 "+d.Token)
-	req.URL.RawQuery = fmtQuery(operationID, query...)
+	req.URL.RawQuery = queryString
 
-	return d.doer.Do(req)
+	rsp, err := d.doer.Do(req)
+	if err != nil {
+		return 0, nil, err
+	}
+	respBody, err := io.ReadAll(rsp.Body)
+	return rsp.StatusCode, respBody, errors.Join(err, rsp.Body.Close())
 }
 
 func isEmpty(a any) bool {