vrr/cozystack
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

perf(controller): batch all bucket metrics into single Prometheus query

Browse source 
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 <noreply@anthropic.com>
Signed-off-by: ZverGuy <maximbel2003@gmail.com>
This commit is contained in:
ZverGuy 2026-04-14 14:47:15 +03:00
parent ed9808fad6
commit 475d24b029
2 changed files with 108 additions and 77 deletions
Download patch file Download diff file Expand all files Collapse all files

113
internal/controller/workloadmonitor_controller.go
View file

@ -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
}

72
internal/controller/workloadmonitor_controller_test.go
View file

@ -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))
}
}