impr: BKTCLT-34 add Prometheus metrics

Add metrics when requests are processed, in order for them to be
exposed by the API user and scraped by Prometheus.

go/bucketclient.go

@@ -2,8 +2,9 @@ package bucketclient
 
 type BucketClient struct {
 	Endpoint string
+	Metrics  *BucketClientMetrics
 }
 
 func New(bucketdEndpoint string) *BucketClient {
-	return &BucketClient{bucketdEndpoint}
+	return &BucketClient{Endpoint: bucketdEndpoint}
 }

go/bucketclientmetrics.go

@@ -0,0 +1,94 @@
+package bucketclient
+
+import (
+	"sync"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+type BucketClientMetrics struct {
+	RequestsTotal              *prometheus.CounterVec
+	RequestDurationSeconds     *prometheus.SummaryVec
+	RequestBytesSentTotal      *prometheus.CounterVec
+	ResponseBytesReceivedTotal *prometheus.CounterVec
+}
+
+var metricsLabels = []string{
+	"endpoint",
+	"method",
+	"action",
+	"code",
+}
+
+var globalMetrics *BucketClientMetrics
+var globalMetricsLock sync.Mutex
+
+// EnableMetrics enables Prometheus metrics gathering for the provided client and registers
+// them in the provided registerer.
+//
+// Metrics implemented:
+//   - `s3_metadata_bucketclient_requests_total`:
+//     Number of requests processed (counter)
+//   - `s3_metadata_bucketclient_request_duration_seconds`:
+//     Time elapsed processing requests to bucketd, in seconds (summary)
+//   - `s3_metadata_bucketclient_request_bytes_sent_total`:
+//     Number of request body bytes sent to bucketd (counter)
+//   - `s3_metadata_bucketclient_response_bytes_received_total`:
+//     Number of response body bytes received from bucketd (counter)
+//
+// Metrics have the following labels attached:
+//   - `endpoint`:
+//     bucketd endpoint such as `http://localhost:9000`
+//   - `method`:
+//     HTTP method
+//   - `action`:
+//     name of the API action, such as `CreateBucket`. Admin actions are prefixed with `Admin`.
+//   - `code`:
+//     HTTP status code returned, or "0" for generic network or protocol errors
+func (client *BucketClient) EnableMetrics(registerer prometheus.Registerer) error {
+	globalMetricsLock.Lock()
+	defer globalMetricsLock.Unlock()
+
+	if globalMetrics == nil {
+		globalMetrics = &BucketClientMetrics{
+			RequestsTotal: prometheus.NewCounterVec(prometheus.CounterOpts{
+				Namespace: MetricsNamespace,
+				Name:      "requests_total",
+				Help:      "Number of requests processed",
+			}, metricsLabels),
+
+			RequestDurationSeconds: prometheus.NewSummaryVec(prometheus.SummaryOpts{
+				Namespace:  MetricsNamespace,
+				Name:       "request_duration_seconds",
+				Help:       "Time elapsed processing requests to bucketd, in seconds",
+				Objectives: MetricsSummaryDefaultObjectives,
+			}, metricsLabels),
+
+			RequestBytesSentTotal: prometheus.NewCounterVec(prometheus.CounterOpts{
+				Namespace: MetricsNamespace,
+				Name:      "request_bytes_sent_total",
+				Help:      "Number of request body bytes sent to bucketd",
+			}, metricsLabels),
+
+			ResponseBytesReceivedTotal: prometheus.NewCounterVec(prometheus.CounterOpts{
+				Namespace: MetricsNamespace,
+				Name:      "response_bytes_received_total",
+				Help:      "Number of response body bytes received from bucketd",
+			}, metricsLabels),
+		}
+		if err := registerer.Register(globalMetrics.RequestsTotal); err != nil {
+			return err
+		}
+		if err := registerer.Register(globalMetrics.RequestDurationSeconds); err != nil {
+			return err
+		}
+		if err := registerer.Register(globalMetrics.RequestBytesSentTotal); err != nil {
+			return err
+		}
+		if err := registerer.Register(globalMetrics.ResponseBytesReceivedTotal); err != nil {
+			return err
+		}
+	}
+	client.Metrics = globalMetrics
+	return nil
+}

go/bucketclientmetrics_test.go

@@ -0,0 +1,76 @@
+package bucketclient_test
+
+import (
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+
+	"io"
+	"net/http"
+	"strings"
+
+	"github.com/jarcoal/httpmock"
+	"github.com/prometheus/client_golang/prometheus"
+	promTestutil "github.com/prometheus/client_golang/prometheus/testutil"
+
+	"github.com/scality/bucketclient/go"
+)
+
+var _ = Describe("BucketClientMetrics", func() {
+	Describe("RegisterMetrics()", func() {
+		It("enables metrics collection for all requests of all clients with metrics enabled", func(ctx SpecContext) {
+			client1 := bucketclient.New("http://localhost:9000")
+			Expect(client1).ToNot(BeNil())
+			client2 := bucketclient.New("http://localhost:9001")
+			Expect(client2).ToNot(BeNil())
+
+			registry := prometheus.NewPedanticRegistry()
+			Expect(client1.EnableMetrics(registry)).To(Succeed())
+			Expect(client2.EnableMetrics(registry)).To(Succeed())
+
+			mockAttributes := `{"foo":"bar"}`
+			httpmock.RegisterResponder(
+				"GET", "http://localhost:9000/default/attributes/my-bucket",
+				httpmock.NewStringResponder(200, mockAttributes),
+			)
+			expectedBatch := `{"batch":[{"key":"foo","value":"{}"}]}`
+			batchErrorResponse := "OOPS!"
+			httpmock.RegisterResponder(
+				"POST", "http://localhost:9001/default/batch/my-bucket",
+				func(req *http.Request) (*http.Response, error) {
+					defer req.Body.Close()
+					Expect(io.ReadAll(req.Body)).To(Equal([]byte(expectedBatch)))
+					return httpmock.NewStringResponse(500, batchErrorResponse), nil
+				},
+			)
+
+			_, err := client1.GetBucketAttributes(ctx, "my-bucket")
+			Expect(err).ToNot(HaveOccurred())
+
+			err = client2.PostBatch(ctx, "my-bucket", []bucketclient.PostBatchEntry{
+				{Key: "foo", Value: "{}"},
+			})
+			Expect(err).To(HaveOccurred())
+
+			Expect(promTestutil.GatherAndCompare(registry, strings.NewReader(`
+# HELP s3_metadata_bucketclient_requests_total Number of requests processed
+# TYPE s3_metadata_bucketclient_requests_total counter
+s3_metadata_bucketclient_requests_total{action="GetBucketAttributes",code="200",endpoint="http://localhost:9000",method="GET"} 1
+s3_metadata_bucketclient_requests_total{action="PostBatch",code="500",endpoint="http://localhost:9001",method="POST"} 1
+
+# HELP s3_metadata_bucketclient_request_bytes_sent_total Number of request body bytes sent to bucketd
+# TYPE s3_metadata_bucketclient_request_bytes_sent_total counter
+s3_metadata_bucketclient_request_bytes_sent_total{action="GetBucketAttributes",code="200",endpoint="http://localhost:9000",method="GET"} 0
+s3_metadata_bucketclient_request_bytes_sent_total{action="PostBatch",code="500",endpoint="http://localhost:9001",method="POST"} 38
+
+# HELP s3_metadata_bucketclient_response_bytes_received_total Number of response body bytes received from bucketd
+# TYPE s3_metadata_bucketclient_response_bytes_received_total counter
+s3_metadata_bucketclient_response_bytes_received_total{action="GetBucketAttributes",code="200",endpoint="http://localhost:9000",method="GET"} 13
+s3_metadata_bucketclient_response_bytes_received_total{action="PostBatch",code="500",endpoint="http://localhost:9001",method="POST"} 5
+`),
+				"s3_metadata_bucketclient_requests_total",
+				"s3_metadata_bucketclient_request_bytes_sent_total",
+				"s3_metadata_bucketclient_response_bytes_received_total",
+			)).To(Succeed())
+		})
+	})
+})

go/bucketclientrequest.go

@@ -6,7 +6,11 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"strconv"
 	"strings"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
 )
 
 type requestOptionSet struct {
@@ -45,11 +49,44 @@ func parseRequestOptions(opts ...RequestOption) (requestOptionSet, error) {
 	return parsedOpts, nil
 }
 
+// updateMetrics is a local helper to update Prometheus metrics in a generic way before
+// returning a response or an error to the API caller.
+func (client *BucketClient) updateMetrics(apiMethod, httpMethod string,
+	startTime time.Time, requestBody []byte,
+	httpCode int, responseBody []byte) {
+	if client.Metrics == nil {
+		return
+	}
+	labels := prometheus.Labels{
+		"endpoint": client.Endpoint,
+		"method":   httpMethod,
+		"action":   apiMethod,
+		"code":     strconv.Itoa(httpCode),
+	}
+	elapsedSeconds := time.Since(startTime).Seconds()
+
+	client.Metrics.RequestsTotal.With(labels).Inc()
+	client.Metrics.RequestDurationSeconds.With(labels).Observe(elapsedSeconds)
+
+	var requestBodyLength int
+	if requestBody != nil {
+		requestBodyLength = len(requestBody)
+	}
+	client.Metrics.RequestBytesSentTotal.With(labels).Add(float64(requestBodyLength))
+
+	var responseBodyLength int
+	if responseBody != nil {
+		responseBodyLength = len(responseBody)
+	}
+	client.Metrics.ResponseBytesReceivedTotal.With(labels).Add(float64(responseBodyLength))
+}
+
 func (client *BucketClient) Request(ctx context.Context,
 	apiMethod string, httpMethod string, resource string, opts ...RequestOption) ([]byte, error) {
 	var response *http.Response
 	var err error
 
+	startTime := time.Now()
 	options, err := parseRequestOptions(opts...)
 	if err == nil {
 		url := fmt.Sprintf("%s%s", client.Endpoint, resource)
@@ -71,13 +108,24 @@ func (client *BucketClient) Request(ctx context.Context,
 		}
 	}
 	if err != nil {
+		client.updateMetrics(apiMethod, httpMethod, startTime, options.requestBody, 0, nil)
 		return nil, &BucketClientError{
 			apiMethod, httpMethod, client.Endpoint, resource, 0, "", err,
 		}
 	}
-	if response.Body != nil {
-		defer response.Body.Close()
+	defer response.Body.Close()
+	responseBody, err := io.ReadAll(response.Body)
+	if err != nil {
+		// We have a HTTP status code but we couldn't read the whole response body,
+		// so use "0" as status code for a generic transport error
+		client.updateMetrics(apiMethod, httpMethod, startTime, options.requestBody, 0, nil)
+		return nil, &BucketClientError{
+			apiMethod, httpMethod, client.Endpoint, resource, 0, "",
+			fmt.Errorf("error reading response body: %w", err),
+		}
 	}
+	client.updateMetrics(apiMethod, httpMethod, startTime, options.requestBody,
+		response.StatusCode, responseBody)
 
 	if response.StatusCode/100 != 2 {
 		splitStatus := strings.Split(response.Status, " ")
@@ -90,12 +138,5 @@ func (client *BucketClient) Request(ctx context.Context,
 			response.StatusCode, errorType, nil,
 		}
 	}
-	responseBody, err := io.ReadAll(response.Body)
-	if err != nil {
-		return nil, &BucketClientError{
-			apiMethod, httpMethod, client.Endpoint, resource, 0, "",
-			fmt.Errorf("error reading response body: %w", err),
-		}
-	}
 	return responseBody, nil
 }

go/constants.go

@@ -14,3 +14,9 @@ const (
 	DBMethodBatch         DBMethodType = 8
 	DBMethodNoop          DBMethodType = 9
 )
+
+const (
+	MetricsNamespace = "s3_metadata_bucketclient"
+)
+
+var MetricsSummaryDefaultObjectives = map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001}

go/go.mod

@@ -6,16 +6,26 @@ require (
 	github.com/jarcoal/httpmock v1.3.1
 	github.com/onsi/ginkgo/v2 v2.20.2
 	github.com/onsi/gomega v1.34.2
+	github.com/prometheus/client_golang v1.20.5
+	github.com/prometheus/client_model v0.6.1
 )
 
 require (
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-task/slim-sprig/v3 v3.0.0 // indirect
 	github.com/google/go-cmp v0.6.0 // indirect
 	github.com/google/pprof v0.0.0-20240827171923-fa2c70bbbfe5 // indirect
+	github.com/kr/text v0.2.0 // indirect
+	github.com/kylelemons/godebug v1.1.0 // indirect
+	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
+	github.com/prometheus/common v0.55.0 // indirect
+	github.com/prometheus/procfs v0.15.1 // indirect
 	golang.org/x/net v0.28.0 // indirect
 	golang.org/x/sys v0.24.0 // indirect
 	golang.org/x/text v0.17.0 // indirect
 	golang.org/x/tools v0.24.0 // indirect
+	google.golang.org/protobuf v1.34.2 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )