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 <noreply@anthropic.com>
Signed-off-by: ZverGuy <maximbel2003@gmail.com>

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)

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
 }
 

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())
+	}
+}