From 475d24b0296ccee0a0cd9b03553ab29df17d8d97 Mon Sep 17 00:00:00 2001 From: ZverGuy Date: Tue, 14 Apr 2026 14:47:15 +0300 Subject: [PATCH] perf(controller): batch all bucket metrics into single Prometheus query MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace 2×N per-bucket HTTP requests with a single query that fetches all SeaweedFS bucket size metrics at once: {__name__=~"SeaweedFS_s3_bucket_(size|physical_size)_bytes"} Results are keyed by bucket name in memory, then looked up per BucketClaim. This reduces HTTP round-trips from 2N+1 to 2 (one namespace lookup + one Prometheus query) regardless of bucket count. Co-Authored-By: Claude Signed-off-by: ZverGuy --- .../controller/workloadmonitor_controller.go | 113 +++++++++++------- .../workloadmonitor_controller_test.go | 72 +++++------ 2 files changed, 108 insertions(+), 77 deletions(-) diff --git a/internal/controller/workloadmonitor_controller.go b/internal/controller/workloadmonitor_controller.go index cb070e0c..63ca02cb 100644 --- a/internal/controller/workloadmonitor_controller.go +++ b/internal/controller/workloadmonitor_controller.go @@ -143,80 +143,109 @@ func (r *WorkloadMonitorReconciler) resolvePrometheusURL(ctx context.Context, na return fmt.Sprintf("http://%s.%s.svc:%s%s", vmSelectService, monitoringNS, vmSelectPort, vmSelectPath) } -// queryPrometheusMetric queries a Prometheus-compatible API for a single instant value. -// Returns the value and true if the metric was found, or 0 and false if the query -// failed or no metric exists. This distinction allows callers to emit a resource -// with value 0 for empty buckets vs omitting it when monitoring is unavailable. -func (r *WorkloadMonitorReconciler) queryPrometheusMetric(ctx context.Context, prometheusBaseURL, promQL string) (int64, bool) { +// bucketMetrics holds size metrics for a single bucket, keyed by metric name. +type bucketMetrics struct { + LogicalSize int64 + PhysicalSize int64 + HasLogical bool + HasPhysical bool +} + +// queryAllBucketMetrics fetches all SeaweedFS bucket size metrics in a single +// Prometheus query and returns them keyed by bucket name. This avoids 2×N HTTP +// round-trips when there are N buckets. +func (r *WorkloadMonitorReconciler) queryAllBucketMetrics(ctx context.Context, prometheusBaseURL string) map[string]*bucketMetrics { + result := make(map[string]*bucketMetrics) if prometheusBaseURL == "" { - return 0, false + return result } logger := log.FromContext(ctx) + query := `{__name__=~"SeaweedFS_s3_bucket_(size|physical_size)_bytes"}` u, err := url.Parse(strings.TrimRight(prometheusBaseURL, "/") + "/api/v1/query") if err != nil { logger.Error(err, "Failed to parse Prometheus URL") - return 0, false + return result } - u.RawQuery = url.Values{"query": {promQL}}.Encode() + u.RawQuery = url.Values{"query": {query}}.Encode() - httpCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + httpCtx, cancel := context.WithTimeout(ctx, 10*time.Second) defer cancel() req, err := http.NewRequestWithContext(httpCtx, http.MethodGet, u.String(), nil) if err != nil { logger.Error(err, "Failed to create Prometheus request") - return 0, false + return result } resp, err := http.DefaultClient.Do(req) if err != nil { - logger.V(1).Info("Failed to query Prometheus", "query", promQL, "error", err) - return 0, false + logger.V(1).Info("Failed to query Prometheus for bucket metrics", "error", err) + return result } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { - logger.V(1).Info("Prometheus returned non-OK status", "query", promQL, "status", resp.StatusCode) - return 0, false + logger.V(1).Info("Prometheus returned non-OK status for bucket metrics", "status", resp.StatusCode) + return result } - body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) + body, err := io.ReadAll(io.LimitReader(resp.Body, 4<<20)) if err != nil { logger.Error(err, "Failed to read Prometheus response") - return 0, false + return result } - // Parse Prometheus instant query response: - // {"status":"success","data":{"resultType":"vector","result":[{"metric":{...},"value":[timestamp,"value"]}]}} var promResp struct { Status string `json:"status"` Data struct { Result []struct { - Value [2]json.RawMessage `json:"value"` + Metric map[string]string `json:"metric"` + Value [2]json.RawMessage `json:"value"` } `json:"result"` } `json:"data"` } if err := json.Unmarshal(body, &promResp); err != nil { logger.Error(err, "Failed to parse Prometheus response") - return 0, false + return result } - if promResp.Status != "success" || len(promResp.Data.Result) == 0 { - return 0, false + if promResp.Status != "success" { + return result } - var valueStr string - if err := json.Unmarshal(promResp.Data.Result[0].Value[1], &valueStr); err != nil { - logger.Error(err, "Failed to parse Prometheus metric value") - return 0, false + for _, r := range promResp.Data.Result { + bucket := r.Metric["bucket"] + metricName := r.Metric["__name__"] + if bucket == "" || metricName == "" { + continue + } + + var valueStr string + if err := json.Unmarshal(r.Value[1], &valueStr); err != nil { + continue + } + val, err := strconv.ParseFloat(valueStr, 64) + if err != nil { + continue + } + + bm, ok := result[bucket] + if !ok { + bm = &bucketMetrics{} + result[bucket] = bm + } + + switch metricName { + case "SeaweedFS_s3_bucket_size_bytes": + bm.LogicalSize = int64(val) + bm.HasLogical = true + case "SeaweedFS_s3_bucket_physical_size_bytes": + bm.PhysicalSize = int64(val) + bm.HasPhysical = true + } } - val, err := strconv.ParseFloat(valueStr, 64) - if err != nil { - logger.Error(err, "Failed to parse metric value", "value", valueStr) - return 0, false - } - return int64(val), true + return result } // reconcileBucketClaimForMonitor creates or updates a Workload object for the given BucketClaim and WorkloadMonitor. @@ -224,7 +253,7 @@ func (r *WorkloadMonitorReconciler) reconcileBucketClaimForMonitor( ctx context.Context, monitor *cozyv1alpha1.WorkloadMonitor, bc cosiv1alpha1.BucketClaim, - promURL string, + allMetrics map[string]*bucketMetrics, ) error { logger := log.FromContext(ctx) workload := &cozyv1alpha1.Workload{ @@ -237,18 +266,15 @@ func (r *WorkloadMonitorReconciler) reconcileBucketClaimForMonitor( resources := make(map[string]resource.Quantity) - // Query actual bucket sizes from SeaweedFS metrics via Prometheus. - // The monitoring endpoint is resolved from the namespace label - // namespace.cozystack.io/monitoring, which points to the tenant - // namespace hosting VictoriaMetrics. + // Look up pre-fetched bucket metrics by the SeaweedFS bucket name. // bc.Status.BucketName is the COSI Bucket name, which the COSI driver // uses directly as the SeaweedFS bucket name. - if bn := bc.Status.BucketName; bn != "" { - if v, ok := r.queryPrometheusMetric(ctx, promURL, fmt.Sprintf(`SeaweedFS_s3_bucket_size_bytes{bucket="%s"}`, bn)); ok { - resources["s3-storage-bytes"] = *resource.NewQuantity(v, resource.BinarySI) + if bm, ok := allMetrics[bc.Status.BucketName]; ok { + if bm.HasLogical { + resources["s3-storage-bytes"] = *resource.NewQuantity(bm.LogicalSize, resource.BinarySI) } - if v, ok := r.queryPrometheusMetric(ctx, promURL, fmt.Sprintf(`SeaweedFS_s3_bucket_physical_size_bytes{bucket="%s"}`, bn)); ok { - resources["s3-physical-storage-bytes"] = *resource.NewQuantity(v, resource.BinarySI) + if bm.HasPhysical { + resources["s3-physical-storage-bytes"] = *resource.NewQuantity(bm.PhysicalSize, resource.BinarySI) } } @@ -564,8 +590,9 @@ func (r *WorkloadMonitorReconciler) Reconcile(ctx context.Context, req ctrl.Requ } bucketPromURL := r.resolvePrometheusURL(ctx, monitor.Namespace) + allBucketMetrics := r.queryAllBucketMetrics(ctx, bucketPromURL) for _, bc := range bucketClaimList.Items { - if err := r.reconcileBucketClaimForMonitor(ctx, monitor, bc, bucketPromURL); err != nil { + if err := r.reconcileBucketClaimForMonitor(ctx, monitor, bc, allBucketMetrics); err != nil { logger.Error(err, "Failed to reconcile Workload for BucketClaim", "BucketClaim", bc.Name) continue } diff --git a/internal/controller/workloadmonitor_controller_test.go b/internal/controller/workloadmonitor_controller_test.go index f702c978..5bb3397e 100644 --- a/internal/controller/workloadmonitor_controller_test.go +++ b/internal/controller/workloadmonitor_controller_test.go @@ -377,69 +377,73 @@ func TestReconcileNoBucketClaimSkips(t *testing.T) { } } -func TestQueryPrometheusMetric(t *testing.T) { +func TestQueryAllBucketMetrics(t *testing.T) { srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - fmt.Fprint(w, `{"status":"success","data":{"resultType":"vector","result":[{"metric":{"bucket":"cosi-abc123"},"value":[1713000000,"5368709120"]}]}}`) + fmt.Fprint(w, `{"status":"success","data":{"resultType":"vector","result":[ + {"metric":{"__name__":"SeaweedFS_s3_bucket_size_bytes","bucket":"bucket-aaa"},"value":[1713000000,"10485864"]}, + {"metric":{"__name__":"SeaweedFS_s3_bucket_physical_size_bytes","bucket":"bucket-aaa"},"value":[1713000000,"20971728"]}, + {"metric":{"__name__":"SeaweedFS_s3_bucket_size_bytes","bucket":"bucket-bbb"},"value":[1713000000,"0"]} + ]}}`) })) defer srv.Close() reconciler := &WorkloadMonitorReconciler{} - size, ok := reconciler.queryPrometheusMetric(context.TODO(), srv.URL, `SeaweedFS_s3_bucket_size_bytes{bucket="cosi-abc123"}`) + metrics := reconciler.queryAllBucketMetrics(context.TODO(), srv.URL) + + bm, ok := metrics["bucket-aaa"] if !ok { - t.Fatal("expected ok=true for valid metric") + t.Fatal("expected bucket-aaa in metrics") } - if size != 5368709120 { - t.Errorf("expected 5368709120, got %d", size) + if !bm.HasLogical || bm.LogicalSize != 10485864 { + t.Errorf("expected logical=10485864, got %d", bm.LogicalSize) + } + if !bm.HasPhysical || bm.PhysicalSize != 20971728 { + t.Errorf("expected physical=20971728, got %d", bm.PhysicalSize) + } + + bm2, ok := metrics["bucket-bbb"] + if !ok { + t.Fatal("expected bucket-bbb in metrics") + } + if !bm2.HasLogical || bm2.LogicalSize != 0 { + t.Errorf("expected logical=0 for empty bucket, got %d", bm2.LogicalSize) + } + if bm2.HasPhysical { + t.Error("expected no physical size for bucket-bbb") } } -func TestQueryPrometheusMetricZeroValue(t *testing.T) { - srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - fmt.Fprint(w, `{"status":"success","data":{"resultType":"vector","result":[{"metric":{"bucket":"empty"},"value":[1713000000,"0"]}]}}`) - })) - defer srv.Close() - - reconciler := &WorkloadMonitorReconciler{} - size, ok := reconciler.queryPrometheusMetric(context.TODO(), srv.URL, `SeaweedFS_s3_bucket_size_bytes{bucket="empty"}`) - if !ok { - t.Fatal("expected ok=true for zero-value metric (empty bucket)") - } - if size != 0 { - t.Errorf("expected 0, got %d", size) - } -} - -func TestQueryPrometheusMetricEmpty(t *testing.T) { +func TestQueryAllBucketMetricsEmpty(t *testing.T) { srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { fmt.Fprint(w, `{"status":"success","data":{"resultType":"vector","result":[]}}`) })) defer srv.Close() reconciler := &WorkloadMonitorReconciler{} - _, ok := reconciler.queryPrometheusMetric(context.TODO(), srv.URL, `SeaweedFS_s3_bucket_size_bytes{bucket="nonexistent"}`) - if ok { - t.Error("expected ok=false for empty result") + metrics := reconciler.queryAllBucketMetrics(context.TODO(), srv.URL) + if len(metrics) != 0 { + t.Errorf("expected empty metrics, got %d", len(metrics)) } } -func TestQueryPrometheusMetricServerError(t *testing.T) { +func TestQueryAllBucketMetricsServerError(t *testing.T) { srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusInternalServerError) })) defer srv.Close() reconciler := &WorkloadMonitorReconciler{} - _, ok := reconciler.queryPrometheusMetric(context.TODO(), srv.URL, `SeaweedFS_s3_bucket_size_bytes{bucket="test"}`) - if ok { - t.Error("expected ok=false for server error") + metrics := reconciler.queryAllBucketMetrics(context.TODO(), srv.URL) + if len(metrics) != 0 { + t.Errorf("expected empty metrics on error, got %d", len(metrics)) } } -func TestQueryPrometheusMetricNoURL(t *testing.T) { +func TestQueryAllBucketMetricsNoURL(t *testing.T) { reconciler := &WorkloadMonitorReconciler{} - _, ok := reconciler.queryPrometheusMetric(context.TODO(), "", `anything`) - if ok { - t.Error("expected ok=false when PrometheusURL is empty") + metrics := reconciler.queryAllBucketMetrics(context.TODO(), "") + if len(metrics) != 0 { + t.Errorf("expected empty metrics when URL is empty, got %d", len(metrics)) } }