From 84cd3c621be4720cbb43b83d3ec602dcb93d3671 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 9 Nov 2026 11:37:06 -0800 Subject: [PATCH] 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 | 1 + weed/s3api/s3api_circuit_breaker.go | 8 ++ weed/s3api/s3err/audit_fluent.go | 11 ++- weed/s3api/s3vectors/audit_test.go | 50 +++++++++++ weed/s3api/s3vectors/handlers.go | 6 ++ weed/s3api/s3vectors/limits.go | 118 +++++++++++++++++++++++++ weed/s3api/s3vectors/limits_test.go | 125 +++++++++++++++++++++++++++ weed/s3api/s3vectors/metrics_test.go | 76 ++++++++++++++++ weed/s3api/s3vectors/server.go | 77 ++++++++++++++++- weed/s3api/s3vectors/types.go | 22 +++-- 10 files changed, 484 insertions(+), 10 deletions(-) create mode 100644 weed/s3api/s3vectors/audit_test.go create mode 100644 weed/s3api/s3vectors/limits.go create mode 100644 weed/s3api/s3vectors/limits_test.go create mode 100644 weed/s3api/s3vectors/metrics_test.go diff --git a/weed/command/s3.go b/weed/command/s3.go index 12b293d1de..753cfeacf2 100644 --- a/weed/command/s3.go +++ b/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 } diff --git a/weed/s3api/s3api_circuit_breaker.go b/weed/s3api/s3api_circuit_breaker.go index 786542d757..5bc9de126c 100644 --- a/weed/s3api/s3api_circuit_breaker.go +++ b/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 +} diff --git a/weed/s3api/s3err/audit_fluent.go b/weed/s3api/s3err/audit_fluent.go index 077018f892..a47966ac7d 100644 --- a/weed/s3api/s3err/audit_fluent.go +++ b/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 { diff --git a/weed/s3api/s3vectors/audit_test.go b/weed/s3api/s3vectors/audit_test.go new file mode 100644 index 0000000000..4521e058c1 --- /dev/null +++ b/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) + } +} diff --git a/weed/s3api/s3vectors/handlers.go b/weed/s3api/s3vectors/handlers.go index 56a48f6828..f9f5b75cea 100644 --- a/weed/s3api/s3vectors/handlers.go +++ b/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 } diff --git a/weed/s3api/s3vectors/limits.go b/weed/s3api/s3vectors/limits.go new file mode 100644 index 0000000000..d88a356a7d --- /dev/null +++ b/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() + } + } +} diff --git a/weed/s3api/s3vectors/limits_test.go b/weed/s3api/s3vectors/limits_test.go new file mode 100644 index 0000000000..2f86755b43 --- /dev/null +++ b/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") + } +} diff --git a/weed/s3api/s3vectors/metrics_test.go b/weed/s3api/s3vectors/metrics_test.go new file mode 100644 index 0000000000..351b52a814 --- /dev/null +++ b/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) + } +} diff --git a/weed/s3api/s3vectors/server.go b/weed/s3api/s3vectors/server.go index b901881eaf..65ad351748 100644 --- a/weed/s3api/s3vectors/server.go +++ b/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 +} diff --git a/weed/s3api/s3vectors/types.go b/weed/s3api/s3vectors/types.go index 4f5f1496ec..9139683a7e 100644 --- a/weed/s3api/s3vectors/types.go +++ b/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) }