s3vectors: integrate bucket metrics, audit events, and request limits (#12514)

* s3vectors: attribute request and traffic metrics to authorized buckets

* s3vectors: emit resource and identity audit events

* s3vectors: apply existing S3 request limits and preserve JSON errors

* s3vectors: select limit resources by the dispatched operation

* s3vectors: count decoded chunked request bodies against byte limits

weed/command/s3.go

@@ -764,6 +764,7 @@ func (s3opt *S3Options) vectorsRouter(s3ApiServer *s3api.S3ApiServer, grpcDialOp
 	vectorsServer.SetAccountID(s3tables.DefaultAccountID)
 	vectorsServer.SetEncryptionChecker(s3api.GetSSES3KeyManager().IsConfigured)
 	vectorsServer.SetDescribe(describe)
+	vectorsServer.SetLimiter(s3ApiServer.Limit)
 	vectorsServer.RegisterRoutes(vectorsRouter)
 	return vectorsRouter
 }

weed/s3api/s3api_circuit_breaker.go

@@ -260,3 +260,11 @@ func (cb *CircuitBreaker) loadCounterAndCompare(key string, inc int64, errCode s
 	}
 	return
 }
+
+func (s3a *S3ApiServer) Limit(handler http.HandlerFunc, action string) http.HandlerFunc {
+	if s3a.cb == nil {
+		return handler
+	}
+	limited, _ := s3a.cb.Limit(handler, action)
+	return limited
+}

weed/s3api/s3err/audit_fluent.go

@@ -145,7 +145,16 @@ func getResourceType(object string, query_key string, method string) (string, bo
 	}
 }
 
+type accessLogOperationKey struct{}
+
+func WithAccessLogOperation(r *http.Request, operation string) *http.Request {
+	return r.WithContext(context.WithValue(r.Context(), accessLogOperationKey{}, operation))
+}
+
 func getOperation(object string, r *http.Request) string {
+	if operation, ok := r.Context().Value(accessLogOperationKey{}).(string); ok {
+		return operation
+	}
 	queries := r.URL.Query()
 	var operation string
 	var queryFound bool
@@ -210,7 +219,7 @@ func PostLog(r *http.Request, HTTPStatusCode int, errorCode ErrorCode) {
 }
 
 func PostAccessLog(log AccessLog) {
-	if Logger == nil || len(log.Key) == 0 {
+	if Logger == nil {
 		return
 	}
 	if err := Logger.Post(tag, log); err != nil {

weed/s3api/s3vectors/audit_test.go

@@ -0,0 +1,50 @@
+package s3vectors
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
+)
+
+func TestVectorAuditRecordsIdentityResourceAndErrors(t *testing.T) {
+	for _, status := range []int{http.StatusOK, http.StatusForbidden, http.StatusServiceUnavailable} {
+		t.Run(http.StatusText(status), func(t *testing.T) {
+			r := s3err.EnsureAuditTracking(httptest.NewRequest(http.MethodPost, "/QueryVectors", strings.NewReader(`{}`)))
+			r = s3_constants.EnsureIdentityHolder(r)
+			s := &Server{}
+			var got *s3err.AccessLog
+			s.instrument(func(w http.ResponseWriter, request *http.Request) {
+				request = request.WithContext(s3_constants.SetIdentityNameInContext(request.Context(), "alice"))
+				setRequestTarget(request, "embeddings", "documents")
+				w.WriteHeader(status)
+				if status >= http.StatusBadRequest {
+					_, _ = w.Write([]byte(`{"code":"AccessDenied","message":"no secret data"}`))
+				}
+				info := request.Context().Value(requestInfoKey{}).(*requestInfo)
+				got = accessLog(request, status, "s3vectors.QueryVectors", info)
+			})(httptest.NewRecorder(), r)
+			if !s3err.AuditAlreadyLogged(r) {
+				t.Fatal("vector response did not emit an audit event")
+			}
+			if got.Requester != "alice" || got.Bucket != "embeddings" || got.Key != "documents" || got.Operation != "s3vectors.QueryVectors" || got.HTTPStatus != status {
+				t.Fatalf("audit event: %+v", got)
+			}
+			if status >= http.StatusBadRequest && got.ErrorCode != "AccessDenied" {
+				t.Fatalf("error code = %q", got.ErrorCode)
+			}
+		})
+	}
+}
+
+func TestVectorBucketAuditHasNoObjectKey(t *testing.T) {
+	r := httptest.NewRequest(http.MethodPost, "/CreateVectorBucket", nil).WithContext(context.Background())
+	log := accessLog(r, http.StatusOK, "s3vectors.CreateVectorBucket", &requestInfo{bucket: "embeddings"})
+	if log.Bucket != "embeddings" || log.Key != "" {
+		t.Fatalf("bucket audit: %+v", log)
+	}
+}

weed/s3api/s3vectors/handlers.go

@@ -273,6 +273,7 @@ func (s *Server) handleCreateVectorBucket(w http.ResponseWriter, r *http.Request
 		writeError(w, http.StatusBadRequest, "ValidationException", err.Error())
 		return
 	}
+	setRequestTarget(r, req.VectorBucketName, "")
 	if req.VectorBucketName == "" {
 		writeError(w, http.StatusBadRequest, "ValidationException", "vectorBucketName is required")
 		return
@@ -314,6 +315,7 @@ func (s *Server) handleCreateVectorBucket(w http.ResponseWriter, r *http.Request
 		catalogErr(w, err)
 		return
 	}
+	authorizeRequest(r, req.VectorBucketName, "")
 	writeJSON(w, http.StatusOK, &CreateVectorBucketResponse{VectorBucketArn: s.vectorBucketArn(req.VectorBucketName)})
 }
 
@@ -909,6 +911,7 @@ var errVectorBucketReplaced = &s3tables.S3TablesError{Type: s3tables.ErrCodeConf
 // compare-and-swap, so a grant revoked between reads never authorizes the
 // pending write, and the entry must still be the bucket record was read from.
 func (s *Server) checkVectorPolicyDoc(r *http.Request, action, bucket string, record *s3tables.GetTableBucketResponse, ext map[string][]byte) error {
+	setRequestTarget(r, bucket, "")
 	var meta struct {
 		CreatedAt time.Time `json:"createdAt"`
 	}
@@ -924,6 +927,7 @@ func (s *Server) checkVectorPolicyDoc(r *http.Request, action, bucket string, re
 		vectorPolicyFromExt(ext), s.vectorBucketArn(bucket), s.policyContext(r, bucket, tags, tags, tagRequest{})) {
 		return errVectorDenied
 	}
+	authorizeRequest(r, bucket, "")
 	return nil
 }
 
@@ -940,6 +944,7 @@ func (s *Server) authorizeVector(ctx context.Context, r *http.Request, action, b
 // authorizeVectorTags is authorizeVector for a request that sets or removes
 // tags.
 func (s *Server) authorizeVectorTags(ctx context.Context, r *http.Request, action, bucket, index string, req tagRequest) (*s3tables.GetTableBucketResponse, error) {
+	setRequestTarget(r, bucket, index)
 	record, err := s.vectorBucket(ctx, bucket)
 	if err != nil {
 		return nil, err
@@ -966,6 +971,7 @@ func (s *Server) authorizeVectorTags(ctx context.Context, r *http.Request, actio
 		return nil, &s3tables.S3TablesError{Type: s3tables.ErrCodeNoSuchBucket,
 			Message: fmt.Sprintf("vector bucket %s not found", bucket)}
 	}
+	authorizeRequest(r, bucket, index)
 	return record, nil
 }
 

weed/s3api/s3vectors/limits.go

@@ -0,0 +1,118 @@
+package s3vectors
+
+import (
+	"bytes"
+	"encoding/xml"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+
+	"github.com/gorilla/mux"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+)
+
+func (s *Server) SetLimiter(limit func(http.HandlerFunc, string) http.HandlerFunc) {
+	s.limit = limit
+}
+
+func (s *Server) limitRequest(handler http.HandlerFunc, w http.ResponseWriter, r *http.Request) {
+
+	op := operationFromRequest(r)
+	var target interface{} = &struct{}{}
+	resolve := func() (vectorBucketARNs, error) { return vectorBucketARNs{}, nil }
+	switch op {
+	case "TagResource", "UntagResource", "ListTagsForResource":
+		req := &struct {
+			ResourceArn string `json:"resourceArn"`
+		}{}
+		target = req
+		resolve = func() (vectorBucketARNs, error) { arns, _, err := s.tagTarget(r, req.ResourceArn); return arns, err }
+	case "CreateVectorBucket":
+		req := &struct {
+			VectorBucketName string `json:"vectorBucketName"`
+		}{}
+		target = req
+		resolve = func() (vectorBucketARNs, error) { return vectorBucketARNs{bucket: req.VectorBucketName}, nil }
+	case "GetVectorBucket", "DeleteVectorBucket", "PutVectorBucketPolicy", "GetVectorBucketPolicy", "DeleteVectorBucketPolicy", "ListIndexes":
+		req := &GetVectorBucketRequest{}
+		target = req
+		resolve = func() (vectorBucketARNs, error) { return s.bucketARNs(req.VectorBucketName, req.VectorBucketArn) }
+	case "CreateIndex", "GetIndex", "DeleteIndex", "PutVectors", "GetVectors", "DeleteVectors", "ListVectors", "QueryVectors":
+		req := &GetIndexRequest{}
+		target = req
+		resolve = func() (vectorBucketARNs, error) {
+			return s.resolveIndexARNs(req.VectorBucketName, req.VectorBucketArn, req.IndexName, req.IndexArn)
+		}
+	}
+	if r.Method == http.MethodPost {
+		if err := readJSON(r, target); err != nil {
+			writeError(w, http.StatusBadRequest, "ValidationException", err.Error())
+			return
+		}
+	}
+	if info, ok := r.Context().Value(requestInfoKey{}).(*requestInfo); ok && r.Method == http.MethodPost {
+		r = r.Clone(r.Context())
+		r.ContentLength = int64(len(info.body))
+		r.Body = io.NopCloser(bytes.NewReader(info.body))
+	}
+	arns, err := resolve()
+	if err != nil {
+		writeError(w, http.StatusBadRequest, "ValidationException", err.Error())
+		return
+	}
+
+	setRequestTarget(r, arns.bucket, arns.index)
+	vars := make(map[string]string, len(mux.Vars(r))+2)
+	for k, v := range mux.Vars(r) {
+		vars[k] = v
+	}
+	vars["bucket"], vars["object"] = arns.bucket, arns.index
+	r = mux.SetURLVars(r, vars)
+	action := s3_constants.ACTION_WRITE
+	if strings.HasPrefix(op, "Get") || strings.HasPrefix(op, "List") || op == "QueryVectors" {
+		action = s3_constants.ACTION_READ
+	}
+	response := &limitResponseWriter{ResponseWriter: w, denied: httptest.NewRecorder()}
+	s.limit(func(w http.ResponseWriter, r *http.Request) {
+		response.allowed = true
+		handler(w, r)
+	}, action)(response, r)
+	if !response.allowed {
+		w.Header().Del("Content-Length")
+		var failure struct{ Code, Message string }
+		if xml.Unmarshal(response.denied.Body.Bytes(), &failure) != nil || failure.Code == "" {
+			failure.Code, failure.Message = "ThrottlingException", "request limit exceeded"
+		}
+		writeError(w, response.denied.Code, failure.Code, failure.Message)
+	}
+}
+
+type limitResponseWriter struct {
+	http.ResponseWriter
+	denied  *httptest.ResponseRecorder
+	allowed bool
+}
+
+func (w *limitResponseWriter) WriteHeader(status int) {
+	if w.allowed {
+		w.ResponseWriter.WriteHeader(status)
+	} else {
+		w.denied.WriteHeader(status)
+	}
+}
+
+func (w *limitResponseWriter) Write(body []byte) (int, error) {
+	if w.allowed {
+		return w.ResponseWriter.Write(body)
+	}
+	return w.denied.Write(body)
+}
+
+func (w *limitResponseWriter) Flush() {
+	if w.allowed {
+		if f, ok := w.ResponseWriter.(http.Flusher); ok {
+			f.Flush()
+		}
+	}
+}

weed/s3api/s3vectors/limits_test.go

@@ -0,0 +1,125 @@
+package s3vectors
+
+import (
+	"bytes"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/gorilla/mux"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
+)
+
+func TestVectorLimitsResolveResourcesAndPreserveBodies(t *testing.T) {
+	s := NewServer(nil, headerAuth{}, nil)
+	s.SetRegion("r")
+	s.SetAccountID("a")
+	for _, tc := range []struct {
+		op, action string
+		body       interface{}
+	}{
+		{"PutVectors", s3_constants.ACTION_WRITE, PutVectorsRequest{VectorBucketName: "embeddings", IndexName: "documents", Vectors: []*PutInputVector{{Key: "a", Data: &VectorData{Float32: []float32{1, 2}}}}}},
+		{"QueryVectors", s3_constants.ACTION_READ, QueryVectorsRequest{IndexArn: s.indexArn("embeddings", "documents"), TopK: 1}},
+		{"GetVectorBucket", s3_constants.ACTION_READ, GetVectorBucketRequest{VectorBucketArn: s.vectorBucketArn("embeddings")}},
+	} {
+		t.Run(tc.op, func(t *testing.T) {
+			payload, err := json.Marshal(tc.body)
+			if err != nil {
+				t.Fatal(err)
+			}
+			r := httptest.NewRequest(http.MethodPost, "/"+tc.op, bytes.NewReader(payload))
+			var called bool
+			s.SetLimiter(func(next http.HandlerFunc, action string) http.HandlerFunc {
+				return func(w http.ResponseWriter, r *http.Request) {
+					if mux.Vars(r)["bucket"] != "embeddings" || action != tc.action {
+						t.Fatalf("limit target %v, action %s", mux.Vars(r), action)
+					}
+					next(w, r)
+				}
+			})
+			s.instrument(func(w http.ResponseWriter, r *http.Request) {
+				s.limitRequest(func(w http.ResponseWriter, r *http.Request) {
+					called = true
+					var got json.RawMessage
+					if err := readJSON(r, &got); err != nil || !bytes.Equal(got, payload) {
+						t.Fatalf("handler body %s: %v", got, err)
+					}
+				}, w, r)
+			})(httptest.NewRecorder(), r)
+			if !called {
+				t.Fatal("limited handler was not called")
+			}
+		})
+	}
+}
+
+func TestVectorLimitsReturnJSONWithoutCallingHandler(t *testing.T) {
+	s := NewServer(nil, headerAuth{}, nil)
+	s.SetLimiter(func(next http.HandlerFunc, action string) http.HandlerFunc {
+		return func(w http.ResponseWriter, r *http.Request) {
+			s3err.WriteErrorResponse(w, r, s3err.ErrTooManyRequest)
+		}
+	})
+	r := httptest.NewRequest(http.MethodPost, "/PutVectors", bytes.NewBufferString(`{"vectorBucketName":"embeddings","indexName":"documents"}`))
+	w := httptest.NewRecorder()
+	s.instrument(func(w http.ResponseWriter, r *http.Request) {
+		s.limitRequest(func(http.ResponseWriter, *http.Request) { t.Fatal("rejected handler ran") }, w, r)
+	})(w, r)
+	var failure Error
+	if json.Unmarshal(w.Body.Bytes(), &failure) != nil || failure.Code == "" || w.Code < http.StatusBadRequest || w.Header().Get("Content-Type") != "application/x-amz-json-1.1" {
+		t.Fatalf("limit response: %d %v %s", w.Code, w.Header(), w.Body)
+	}
+}
+
+func TestVectorLimitsIgnoreUnrelatedResourceFields(t *testing.T) {
+	s := NewServer(nil, headerAuth{}, nil)
+	for _, tc := range []struct{ op, body, bucket string }{
+		{"PutVectors", `{"vectorBucketName":"bucket-a","indexName":"idx","resourceArn":"arn:aws:s3vectors:r:a:bucket/bucket-b"}`, "bucket-a"},
+		{"PutVectors", `{"vectorBucketName":"bucket-a","indexName":"idx","resourceArn":{"ignored":true}}`, "bucket-a"},
+		{"GetVectorBucket", `{"vectorBucketName":"bucket-a","indexArn":"invalid","indexName":42}`, "bucket-a"},
+		{"CreateIndex", `{"vectorBucketName":"bucket-a","indexName":"idx","resourceArn":"ignored"}`, "bucket-a"},
+		{"ListVectorBuckets", `{"vectorBucketName":"ignored","resourceArn":"ignored"}`, ""},
+	} {
+		t.Run(tc.op+tc.body, func(t *testing.T) {
+			called := false
+			s.SetLimiter(func(next http.HandlerFunc, action string) http.HandlerFunc {
+				return func(w http.ResponseWriter, r *http.Request) {
+					called = true
+					if got := mux.Vars(r)["bucket"]; got != tc.bucket {
+						t.Fatalf("limited bucket %q, want %q", got, tc.bucket)
+					}
+					next(w, r)
+				}
+			})
+			r := httptest.NewRequest(http.MethodPost, "/"+tc.op, bytes.NewBufferString(tc.body))
+			s.instrument(func(w http.ResponseWriter, r *http.Request) {
+				s.limitRequest(func(http.ResponseWriter, *http.Request) {}, w, r)
+			})(httptest.NewRecorder(), r)
+			if !called {
+				t.Fatal("limiter was not invoked")
+			}
+		})
+	}
+}
+
+func TestVectorLimitsCountChunkedBodyBytes(t *testing.T) {
+	s := NewServer(nil, headerAuth{}, nil)
+	body := `{"vectorBucketName":"embeddings","indexName":"documents"}`
+	var got int64
+	s.SetLimiter(func(next http.HandlerFunc, action string) http.HandlerFunc {
+		return func(w http.ResponseWriter, r *http.Request) { got = r.ContentLength; next(w, r) }
+	})
+	r := httptest.NewRequest(http.MethodPost, "/PutVectors", bytes.NewBufferString(body))
+	r.ContentLength = -1
+	s.instrument(func(w http.ResponseWriter, r *http.Request) {
+		s.limitRequest(func(http.ResponseWriter, *http.Request) {}, w, r)
+	})(httptest.NewRecorder(), r)
+	if got != int64(len(body)) {
+		t.Fatalf("limited body length = %d, want %d", got, len(body))
+	}
+	if r.ContentLength != -1 {
+		t.Fatal("mutated original request")
+	}
+}

weed/s3api/s3vectors/metrics_test.go

@@ -0,0 +1,76 @@
+package s3vectors
+
+import (
+	"bytes"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/gorilla/mux"
+	"github.com/prometheus/client_golang/prometheus/testutil"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables/s3tablestest"
+	"github.com/seaweedfs/seaweedfs/weed/stats"
+)
+
+func TestVectorMetricsAttributeAuthorizedBuckets(t *testing.T) {
+	fs := s3tablestest.Start(t)
+	s := NewServer(s3tables.NewManagerClient(fs.Client), headerAuth{}, newFakeVectors(0))
+	var limitCalls int
+	s.SetLimiter(func(next http.HandlerFunc, action string) http.HandlerFunc {
+		return func(w http.ResponseWriter, r *http.Request) { limitCalls++; next(w, r) }
+	})
+	router := mux.NewRouter()
+	s.RegisterRoutes(router)
+	request := func(identity, operation string, payload interface{}) *httptest.ResponseRecorder {
+		t.Helper()
+		body, err := json.Marshal(payload)
+		if err != nil {
+			t.Fatal(err)
+		}
+		r := httptest.NewRequest(http.MethodPost, "/"+operation, bytes.NewReader(body))
+		r.Header.Set("X-Test-Identity", identity)
+		w := httptest.NewRecorder()
+		router.ServeHTTP(w, r)
+		return w
+	}
+	const bucket = "metrics-vector-bucket"
+	if w := request("", "CreateVectorBucket", CreateVectorBucketRequest{VectorBucketName: bucket}); w.Code != http.StatusOK {
+		t.Fatalf("create: %d %s", w.Code, w.Body)
+	}
+	count := stats.S3RequestCounter.WithLabelValues("s3vectors.GetVectorBucket", "200", bucket)
+	before := testutil.ToFloat64(count)
+	if w := request("", "GetVectorBucket", GetVectorBucketRequest{VectorBucketArn: s.vectorBucketArn(bucket)}); w.Code != http.StatusOK {
+		t.Fatalf("get by ARN: %d %s", w.Code, w.Body)
+	}
+	if got := testutil.ToFloat64(count) - before; got != 1 {
+		t.Fatalf("bucket requests = %v, want 1", got)
+	}
+	denied := stats.S3RequestCounter.WithLabelValues("s3vectors.GetVectorBucket", "403", "")
+	before = testutil.ToFloat64(denied)
+	if w := request("stranger", "GetVectorBucket", GetVectorBucketRequest{VectorBucketName: bucket}); w.Code != http.StatusForbidden {
+		t.Fatalf("unauthorized get: %d %s", w.Code, w.Body)
+	}
+	if got := testutil.ToFloat64(denied) - before; got != 1 {
+		t.Fatalf("unattributed denied requests = %v, want 1", got)
+	}
+	if got := testutil.ToFloat64(stats.S3BucketTrafficReceivedBytesCounter.WithLabelValues(bucket)); got <= 0 {
+		t.Fatalf("received bytes = %v", got)
+	}
+	if got := testutil.ToFloat64(stats.S3BucketTrafficSentBytesCounter.WithLabelValues(bucket)); got <= 0 {
+		t.Fatalf("sent bytes = %v", got)
+	}
+	if got := testutil.ToFloat64(stats.S3InFlightRequestsGauge.WithLabelValues("s3vectors.GetVectorBucket")); got != 0 {
+		t.Fatalf("in-flight requests = %v after responses", got)
+	}
+	r := httptest.NewRequest(http.MethodGet, tagsPathPrefix+s.vectorBucketArn(bucket), nil)
+	w := httptest.NewRecorder()
+	router.ServeHTTP(w, r)
+	if w.Code != http.StatusOK || testutil.ToFloat64(stats.S3RequestCounter.WithLabelValues("s3vectors.ListTagsForResource", "200", bucket)) == 0 {
+		t.Fatalf("tag metrics: %d %s", w.Code, w.Body)
+	}
+	if limitCalls != 4 {
+		t.Fatalf("limiter calls = %d, want 4", limitCalls)
+	}
+}

weed/s3api/s3vectors/server.go

@@ -3,6 +3,8 @@
 package s3vectors
 
 import (
+	"context"
+	"encoding/json"
 	"net/http"
 	"strconv"
 	"strings"
@@ -37,6 +39,7 @@ type Server struct {
 	authenticator S3Authenticator
 	region        string
 	accountID     string
+	limit         func(http.HandlerFunc, string) http.HandlerFunc
 }
 
 // NewServer creates an S3 Vectors server over the given filer and catalog.
@@ -111,13 +114,29 @@ func (s *Server) instrument(next http.HandlerFunc) http.HandlerFunc {
 		inFlight := stats_collect.S3InFlightRequestsGauge.WithLabelValues(label)
 		inFlight.Inc()
 		defer inFlight.Dec()
-		recorder := stats_collect.NewStatusResponseWriter(w)
+		r = s3err.WithAccessLogOperation(r, label)
+		info := &requestInfo{}
+		r = s3_constants.EnsureIdentityHolder(s3err.EnsureAuditTracking(r.WithContext(context.WithValue(r.Context(), requestInfoKey{}, info))))
+		recorder := &responseWriter{StatusRecorder: stats_collect.NewStatusResponseWriter(w), info: info}
 		start := time.Now()
 		next(recorder, r)
 		elapsed := time.Since(start)
-		stats_collect.S3RequestHistogram.WithLabelValues(label, "").Observe(elapsed.Seconds())
-		stats_collect.S3RequestCounter.WithLabelValues(label, strconv.Itoa(recorder.Status), "").Inc()
-		stats_collect.RecordBucketActiveTime("")
+		bucket := ""
+		if info.authorized && recorder.Status != http.StatusForbidden {
+			bucket = info.bucket
+		}
+		log := accessLog(r, recorder.Status, label, info)
+		if !s3err.AuditAlreadyLogged(r) {
+			s3err.PostAccessLog(*log)
+			s3err.MarkAuditLogged(r)
+		}
+		stats_collect.S3RequestHistogram.WithLabelValues(label, bucket).Observe(elapsed.Seconds())
+		stats_collect.S3RequestCounter.WithLabelValues(label, strconv.Itoa(recorder.Status), bucket).Inc()
+		if bucket != "" {
+			stats_collect.RecordBucketActiveTime(bucket)
+			stats_collect.S3BucketTrafficReceivedBytesCounter.WithLabelValues(bucket).Add(float64(info.received))
+			stats_collect.S3BucketTrafficSentBytesCounter.WithLabelValues(bucket).Add(float64(info.sent))
+		}
 	}
 }
 
@@ -130,6 +149,10 @@ func (s *Server) dispatch(w http.ResponseWriter, r *http.Request) {
 		writeError(w, http.StatusBadRequest, "ValidationException", "unknown operation "+op)
 		return
 	}
+	if s.limit != nil {
+		s.limitRequest(func(w http.ResponseWriter, r *http.Request) { handler(s, w, r) }, w, r)
+		return
+	}
 	handler(s, w, r)
 }
 
@@ -156,3 +179,49 @@ func (s *Server) Auth(handler http.HandlerFunc) http.HandlerFunc {
 		handler(w, r)
 	}
 }
+
+type requestInfoKey struct{}
+
+type requestInfo struct {
+	bucket, index  string
+	authorized     bool
+	received, sent int64
+	errorCode      string
+	body           []byte
+}
+
+func authorizeRequest(r *http.Request, bucket, index string) {
+	if info, ok := r.Context().Value(requestInfoKey{}).(*requestInfo); ok {
+		info.bucket, info.index, info.authorized = bucket, index, true
+	}
+}
+
+type responseWriter struct {
+	*stats_collect.StatusRecorder
+	info *requestInfo
+}
+
+func (w *responseWriter) Write(body []byte) (int, error) {
+	n, err := w.StatusRecorder.Write(body)
+	w.info.sent += int64(n)
+	if w.Status >= http.StatusBadRequest {
+		var response Error
+		if json.Unmarshal(body, &response) == nil {
+			w.info.errorCode = response.Code
+		}
+	}
+	return n, err
+}
+
+func setRequestTarget(r *http.Request, bucket, index string) {
+	if info, ok := r.Context().Value(requestInfoKey{}).(*requestInfo); ok {
+		info.bucket, info.index = bucket, index
+	}
+}
+
+func accessLog(r *http.Request, status int, operation string, info *requestInfo) *s3err.AccessLog {
+	log := s3err.GetAccessLog(r, status, s3err.ErrNone)
+	log.Operation, log.Bucket, log.Key = operation, info.bucket, info.index
+	log.ErrorCode = info.errorCode
+	return log
+}

weed/s3api/s3vectors/types.go

@@ -42,12 +42,24 @@ const maxRequestBodyBytes = s3_constants.S3VectorsMaxRequestBodyBytes
 var errRequestBodyTooLarge = fmt.Errorf("request body exceeds %d MiB", maxRequestBodyBytes>>20)
 
 func readJSON(r *http.Request, req interface{}) error {
-	body, err := io.ReadAll(io.LimitReader(r.Body, maxRequestBodyBytes+1))
-	if err != nil {
-		return err
+	info, _ := r.Context().Value(requestInfoKey{}).(*requestInfo)
+	var body []byte
+	if info != nil {
+		body = info.body
 	}
-	if len(body) > maxRequestBodyBytes {
-		return errRequestBodyTooLarge
+	if body == nil {
+		var err error
+		body, err = io.ReadAll(io.LimitReader(r.Body, maxRequestBodyBytes+1))
+		if err != nil {
+			return err
+		}
+		if len(body) > maxRequestBodyBytes {
+			return errRequestBodyTooLarge
+		}
+		if info != nil {
+			info.received += int64(len(body))
+			info.body = body
+		}
 	}
 	return json.NewDecoder(bytes.NewReader(body)).Decode(req)
 }