From 04f340ab8461ffcedbbcad89fff162acca101133 Mon Sep 17 00:00:00 2001 From: ZverGuy Date: Mon, 13 Apr 2026 15:03:37 +0300 Subject: [PATCH] feat(controller): add physical storage size metric for S3 buckets Query SeaweedFS_s3_bucket_physical_size_bytes alongside the logical size metric. Physical size includes all replicas and reflects actual disk usage, while logical size reflects what the user stored. Refactor queryBucketSizeBytes into generic queryPrometheusMetric to reuse for both metrics. Co-Authored-By: Claude Signed-off-by: ZverGuy --- .../controller/workloadmonitor_controller.go | 22 +++++++++------ .../workloadmonitor_controller_test.go | 28 ++++++++++++++----- 2 files changed, 34 insertions(+), 16 deletions(-) diff --git a/internal/controller/workloadmonitor_controller.go b/internal/controller/workloadmonitor_controller.go index 960be5c4..d7514552 100644 --- a/internal/controller/workloadmonitor_controller.go +++ b/internal/controller/workloadmonitor_controller.go @@ -115,21 +115,20 @@ func updateOwnerReferences(obj metav1.Object, monitor client.Object) { obj.SetOwnerReferences(owners) } -// queryBucketSizeBytes queries Prometheus for the logical size of a SeaweedFS bucket. +// queryPrometheusMetric queries a Prometheus-compatible API for a single instant value. // Returns 0 if the metric is not available or PrometheusURL is not configured. -func (r *WorkloadMonitorReconciler) queryBucketSizeBytes(ctx context.Context, seaweedBucketName string) int64 { - if r.PrometheusURL == "" || seaweedBucketName == "" { +func (r *WorkloadMonitorReconciler) queryPrometheusMetric(ctx context.Context, promQL string) int64 { + if r.PrometheusURL == "" { return 0 } logger := log.FromContext(ctx) - query := fmt.Sprintf(`SeaweedFS_s3_bucket_size_bytes{bucket="%s"}`, seaweedBucketName) u, err := url.Parse(r.PrometheusURL + "/api/v1/query") if err != nil { logger.Error(err, "Failed to parse Prometheus URL") return 0 } - u.RawQuery = url.Values{"query": {query}}.Encode() + u.RawQuery = url.Values{"query": {promQL}}.Encode() httpCtx, cancel := context.WithTimeout(ctx, 5*time.Second) defer cancel() @@ -142,7 +141,7 @@ func (r *WorkloadMonitorReconciler) queryBucketSizeBytes(ctx context.Context, se resp, err := http.DefaultClient.Do(req) if err != nil { - logger.V(1).Info("Failed to query Prometheus for bucket size", "bucket", seaweedBucketName, "error", err) + logger.V(1).Info("Failed to query Prometheus", "query", promQL, "error", err) return 0 } defer resp.Body.Close() @@ -203,11 +202,16 @@ func (r *WorkloadMonitorReconciler) reconcileBucketClaimForMonitor( resources := make(map[string]resource.Quantity) resources["s3-buckets"] = resource.MustParse("1") - // Query actual bucket size from SeaweedFS metrics via Prometheus. + // Query actual bucket sizes from SeaweedFS metrics via Prometheus. // bc.Status.BucketName is the COSI Bucket name, which the COSI driver // uses directly as the SeaweedFS bucket name. - if sizeBytes := r.queryBucketSizeBytes(ctx, bc.Status.BucketName); sizeBytes > 0 { - resources["s3-storage-bytes"] = *resource.NewQuantity(sizeBytes, resource.BinarySI) + if bn := bc.Status.BucketName; bn != "" { + if v := r.queryPrometheusMetric(ctx, fmt.Sprintf(`SeaweedFS_s3_bucket_size_bytes{bucket="%s"}`, bn)); v > 0 { + resources["s3-storage-bytes"] = *resource.NewQuantity(v, resource.BinarySI) + } + if v := r.queryPrometheusMetric(ctx, fmt.Sprintf(`SeaweedFS_s3_bucket_physical_size_bytes{bucket="%s"}`, bn)); v > 0 { + resources["s3-physical-storage-bytes"] = *resource.NewQuantity(v, resource.BinarySI) + } } _, err := ctrl.CreateOrUpdate(ctx, r.Client, workload, func() error { diff --git a/internal/controller/workloadmonitor_controller_test.go b/internal/controller/workloadmonitor_controller_test.go index fa7074cb..de9b78e5 100644 --- a/internal/controller/workloadmonitor_controller_test.go +++ b/internal/controller/workloadmonitor_controller_test.go @@ -386,35 +386,35 @@ func TestReconcileNoBucketClaimSkips(t *testing.T) { } } -func TestQueryBucketSizeBytes(t *testing.T) { +func TestQueryPrometheusMetric(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"]}]}}`) })) defer srv.Close() reconciler := &WorkloadMonitorReconciler{PrometheusURL: srv.URL} - size := reconciler.queryBucketSizeBytes(context.TODO(), "cosi-abc123") + size := reconciler.queryPrometheusMetric(context.TODO(), `SeaweedFS_s3_bucket_size_bytes{bucket="cosi-abc123"}`) if size != 5368709120 { t.Errorf("expected 5368709120, got %d", size) } } -func TestQueryBucketSizeBytesEmpty(t *testing.T) { +func TestQueryPrometheusMetricEmpty(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{PrometheusURL: srv.URL} - size := reconciler.queryBucketSizeBytes(context.TODO(), "nonexistent") + size := reconciler.queryPrometheusMetric(context.TODO(), `SeaweedFS_s3_bucket_size_bytes{bucket="nonexistent"}`) if size != 0 { t.Errorf("expected 0 for empty result, got %d", size) } } -func TestQueryBucketSizeBytesNoPrometheus(t *testing.T) { +func TestQueryPrometheusMetricNoURL(t *testing.T) { reconciler := &WorkloadMonitorReconciler{PrometheusURL: ""} - size := reconciler.queryBucketSizeBytes(context.TODO(), "cosi-abc123") + size := reconciler.queryPrometheusMetric(context.TODO(), `anything`) if size != 0 { t.Errorf("expected 0 when PrometheusURL is empty, got %d", size) } @@ -422,7 +422,13 @@ func TestQueryBucketSizeBytesNoPrometheus(t *testing.T) { func TestReconcileBucketClaimWithPrometheus(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,"1073741824"]}]}}`) + query := r.URL.Query().Get("query") + switch { + case len(query) > 0 && query[0:len("SeaweedFS_s3_bucket_physical")] == "SeaweedFS_s3_bucket_physical": + fmt.Fprint(w, `{"status":"success","data":{"resultType":"vector","result":[{"metric":{"bucket":"cosi-abc123"},"value":[1713000000,"2147483648"]}]}}`) + default: + fmt.Fprint(w, `{"status":"success","data":{"resultType":"vector","result":[{"metric":{"bucket":"cosi-abc123"},"value":[1713000000,"1073741824"]}]}}`) + } })) defer srv.Close() @@ -509,4 +515,12 @@ func TestReconcileBucketClaimWithPrometheus(t *testing.T) { if bucketsQty.Cmp(resource.MustParse("1")) != 0 { t.Errorf("expected s3-buckets=1, got %s", bucketsQty.String()) } + + physQty, ok := workload.Status.Resources["s3-physical-storage-bytes"] + if !ok { + t.Fatal("expected s3-physical-storage-bytes resource to be set") + } + if physQty.Value() != 2147483648 { + t.Errorf("expected s3-physical-storage-bytes=2147483648 (2 GiB), got %d", physQty.Value()) + } }