From 0aab19f50003149d2bfc63137a09696b4777eb48 Mon Sep 17 00:00:00 2001 From: ZverGuy Date: Mon, 13 Apr 2026 14:45:06 +0300 Subject: [PATCH] feat(controller): add SeaweedFS bucket size metrics via Prometheus Query SeaweedFS_s3_bucket_size_bytes from a Prometheus-compatible API to populate s3-storage-bytes resource on bucket Workloads. The Prometheus URL is configurable via --prometheus-url flag. When set, bucket WorkloadMonitors are requeued every 60s to keep sizes current. When Prometheus is not configured, buckets still get tracked with s3-buckets=1 for existence-based billing. Co-Authored-By: Claude Signed-off-by: ZverGuy --- cmd/cozystack-controller/main.go | 9 +- .../controller/workloadmonitor_controller.go | 91 ++++++++++++- .../workloadmonitor_controller_test.go | 128 ++++++++++++++++++ 3 files changed, 225 insertions(+), 3 deletions(-) diff --git a/cmd/cozystack-controller/main.go b/cmd/cozystack-controller/main.go index bcaca9c9..1175e2b4 100644 --- a/cmd/cozystack-controller/main.go +++ b/cmd/cozystack-controller/main.go @@ -70,6 +70,7 @@ func main() { var disableTelemetry bool var telemetryEndpoint string var telemetryInterval string + var prometheusURL string var tlsOpts []func(*tls.Config) flag.StringVar(&metricsAddr, "metrics-bind-address", "0", "The address the metrics endpoint binds to. "+ "Use :8443 for HTTPS or :8080 for HTTP, or leave as 0 to disable the metrics service.") @@ -87,6 +88,9 @@ func main() { "Endpoint for sending telemetry data") flag.StringVar(&telemetryInterval, "telemetry-interval", "15m", "Interval between telemetry data collection (e.g. 15m, 1h)") + flag.StringVar(&prometheusURL, "prometheus-url", "", + "Prometheus-compatible API URL for querying SeaweedFS bucket metrics (e.g. http://vmselect:8481). "+ + "If empty, S3 bucket size metrics are not collected.") opts := zap.Options{ Development: false, } @@ -180,8 +184,9 @@ func main() { } if err = (&controller.WorkloadMonitorReconciler{ - Client: mgr.GetClient(), - Scheme: mgr.GetScheme(), + Client: mgr.GetClient(), + Scheme: mgr.GetScheme(), + PrometheusURL: prometheusURL, }).SetupWithManager(mgr); err != nil { setupLog.Error(err, "unable to create controller", "controller", "WorkloadMonitor") os.Exit(1) diff --git a/internal/controller/workloadmonitor_controller.go b/internal/controller/workloadmonitor_controller.go index 3b69d0fb..960be5c4 100644 --- a/internal/controller/workloadmonitor_controller.go +++ b/internal/controller/workloadmonitor_controller.go @@ -4,7 +4,11 @@ import ( "context" "encoding/json" "fmt" + "io" + "net/http" + "net/url" "sort" + "time" apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/runtime" @@ -29,6 +33,9 @@ import ( type WorkloadMonitorReconciler struct { client.Client Scheme *runtime.Scheme + // PrometheusURL is the base URL of a Prometheus-compatible API for querying + // SeaweedFS bucket metrics. If empty, bucket size metrics are not collected. + PrometheusURL string } // +kubebuilder:rbac:groups=cozystack.io,resources=workloadmonitors,verbs=get;list;watch;create;update;patch;delete @@ -108,6 +115,76 @@ func updateOwnerReferences(obj metav1.Object, monitor client.Object) { obj.SetOwnerReferences(owners) } +// queryBucketSizeBytes queries Prometheus for the logical size of a SeaweedFS bucket. +// 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 == "" { + 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() + + httpCtx, cancel := context.WithTimeout(ctx, 5*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 + } + + resp, err := http.DefaultClient.Do(req) + if err != nil { + logger.V(1).Info("Failed to query Prometheus for bucket size", "bucket", seaweedBucketName, "error", err) + return 0 + } + defer resp.Body.Close() + + body, err := io.ReadAll(resp.Body) + if err != nil { + logger.Error(err, "Failed to read Prometheus response") + return 0 + } + + // 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"` + } `json:"result"` + } `json:"data"` + } + if err := json.Unmarshal(body, &promResp); err != nil { + logger.Error(err, "Failed to parse Prometheus response") + return 0 + } + if promResp.Status != "success" || len(promResp.Data.Result) == 0 { + return 0 + } + + 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 + } + + qty, err := resource.ParseQuantity(valueStr) + if err != nil { + logger.Error(err, "Failed to parse metric value as quantity", "value", valueStr) + return 0 + } + return qty.Value() +} + // reconcileBucketClaimForMonitor creates or updates a Workload object for the given BucketClaim and WorkloadMonitor. func (r *WorkloadMonitorReconciler) reconcileBucketClaimForMonitor( ctx context.Context, @@ -126,6 +203,13 @@ 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. + // 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) + } + _, err := ctrl.CreateOrUpdate(ctx, r.Client, workload, func() error { updateOwnerReferences(workload.GetObjectMeta(), &bc) @@ -470,7 +554,12 @@ func (r *WorkloadMonitorReconciler) Reconcile(ctx context.Context, req ctrl.Requ return ctrl.Result{}, err } - // Return without requeue if we want purely event-driven reconciliations + // Requeue periodically if there are BucketClaims to keep sizes up to date. + // Bucket sizes come from Prometheus metrics that update every 60s. + if len(bucketClaimList.Items) > 0 && r.PrometheusURL != "" { + return ctrl.Result{RequeueAfter: 60 * time.Second}, nil + } + return ctrl.Result{}, nil } diff --git a/internal/controller/workloadmonitor_controller_test.go b/internal/controller/workloadmonitor_controller_test.go index 9df61cf3..fa7074cb 100644 --- a/internal/controller/workloadmonitor_controller_test.go +++ b/internal/controller/workloadmonitor_controller_test.go @@ -2,6 +2,9 @@ package controller import ( "context" + "fmt" + "net/http" + "net/http/httptest" "testing" cozyv1alpha1 "github.com/cozystack/cozystack/api/v1alpha1" @@ -382,3 +385,128 @@ func TestReconcileNoBucketClaimSkips(t *testing.T) { } } } + +func TestQueryBucketSizeBytes(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") + if size != 5368709120 { + t.Errorf("expected 5368709120, got %d", size) + } +} + +func TestQueryBucketSizeBytesEmpty(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") + if size != 0 { + t.Errorf("expected 0 for empty result, got %d", size) + } +} + +func TestQueryBucketSizeBytesNoPrometheus(t *testing.T) { + reconciler := &WorkloadMonitorReconciler{PrometheusURL: ""} + size := reconciler.queryBucketSizeBytes(context.TODO(), "cosi-abc123") + if size != 0 { + t.Errorf("expected 0 when PrometheusURL is empty, got %d", size) + } +} + +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"]}]}}`) + })) + defer srv.Close() + + s := newTestScheme() + + monitor := &cozyv1alpha1.WorkloadMonitor{ + ObjectMeta: metav1.ObjectMeta{ + Name: "my-bucket", + Namespace: "tenant-demo", + }, + Spec: cozyv1alpha1.WorkloadMonitorSpec{ + Kind: "bucket", + Type: "s3", + Selector: map[string]string{ + "app.kubernetes.io/instance": "my-bucket", + }, + }, + } + + bc := &cosiv1alpha1.BucketClaim{ + ObjectMeta: metav1.ObjectMeta{ + Name: "my-bucket", + Namespace: "tenant-demo", + Labels: map[string]string{ + "app.kubernetes.io/instance": "my-bucket", + }, + }, + Spec: cosiv1alpha1.BucketClaimSpec{ + BucketClassName: "seaweedfs", + Protocols: []cosiv1alpha1.Protocol{cosiv1alpha1.ProtocolS3}, + }, + Status: cosiv1alpha1.BucketClaimStatus{ + BucketReady: true, + BucketName: "cosi-abc123", + }, + } + + fakeClient := fake.NewClientBuilder(). + WithScheme(s). + WithObjects(monitor, bc). + WithStatusSubresource(monitor). + Build() + + reconciler := &WorkloadMonitorReconciler{ + Client: fakeClient, + Scheme: s, + PrometheusURL: srv.URL, + } + req := reconcile.Request{NamespacedName: types.NamespacedName{ + Name: "my-bucket", + Namespace: "tenant-demo", + }} + + result, err := reconciler.Reconcile(context.TODO(), req) + if err != nil { + t.Fatalf("Reconcile returned error: %v", err) + } + + if result.RequeueAfter == 0 { + t.Error("expected RequeueAfter > 0 when PrometheusURL is set and buckets exist") + } + + workload := &cozyv1alpha1.Workload{} + err = fakeClient.Get(context.TODO(), types.NamespacedName{ + Name: "bucket-my-bucket", + Namespace: "tenant-demo", + }, workload) + if err != nil { + t.Fatalf("expected Workload to be created, got error: %v", err) + } + + sizeQty, ok := workload.Status.Resources["s3-storage-bytes"] + if !ok { + t.Fatal("expected s3-storage-bytes resource to be set") + } + if sizeQty.Value() != 1073741824 { + t.Errorf("expected s3-storage-bytes=1073741824 (1 GiB), got %d", sizeQty.Value()) + } + + bucketsQty, ok := workload.Status.Resources["s3-buckets"] + if !ok { + t.Fatal("expected s3-buckets resource to be set") + } + if bucketsQty.Cmp(resource.MustParse("1")) != 0 { + t.Errorf("expected s3-buckets=1, got %s", bucketsQty.String()) + } +}