[apps] feat(bucket): add WorkloadMonitor and BucketClaim tracking to controller (#2391)
## What this PR does
Adds BucketClaim support to WorkloadMonitorReconciler, bringing S3
buckets in line with how Pods, PVCs, and Services are already tracked as
Workload resources.
- Add WorkloadMonitor CR to the bucket Helm chart with
`app.kubernetes.io/instance` label on BucketClaim for selector matching
- Add `reconcileBucketClaimForMonitor()` following the same pattern as
`reconcilePVCForMonitor()` — watches COSI `BucketClaim` objects and
creates corresponding `Workload` CRDs
- Query SeaweedFS bucket size metrics (logical + physical) from
VictoriaMetrics, with the monitoring endpoint resolved automatically
from the `namespace.cozystack.io/monitoring` namespace label
- COSI API types dependency (`container-object-storage-interface-api`)
for typed BucketClaim access
No configuration flags needed — the controller discovers the monitoring
stack for each tenant namespace automatically.
### Result
When a BucketClaim is matched by a WorkloadMonitor, the controller
creates a Workload with S3 storage metrics:
```yaml
apiVersion: cozystack.io/v1alpha1
kind: Workload
metadata:
name: bucket-bucket-test-billing
namespace: tenant-testing
labels:
workloads.cozystack.io/monitor: bucket-test-billing
ownerReferences:
- apiVersion: objectstorage.k8s.io/v1alpha1
kind: BucketClaim
name: bucket-test-billing
status:
kind: bucket
type: s3
operational: true
resources:
s3-buckets: "1"
s3-storage-bytes: "10485864" # 10 MB logical
s3-physical-storage-bytes: "20971728" # 20 MB physical (replication factor 2)
```
- `s3-storage-bytes` — logical size (what the user stored), from
`SeaweedFS_s3_bucket_size_bytes`
- `s3-physical-storage-bytes` — physical size (with replicas), from
`SeaweedFS_s3_bucket_physical_size_bytes`
- When monitoring is not configured for the namespace, only `s3-buckets:
1` is tracked
- Sizes refresh every 60 seconds via `RequeueAfter`
Tested on a live dev cluster with SeaweedFS deployed in `tenant-root`
and a bucket created in `tenant-testing`.
### Release note
```release-note
[apps] Add WorkloadMonitor to bucket application. BucketClaims are now tracked as Workload resources with S3 storage size metrics resolved automatically from the tenant monitoring stack.
```
<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit
* **New Features**
* Monitor COSI bucket claims: creates/updates Workload records
reflecting bucket readiness and requeues periodically when buckets
exist.
* Prometheus-backed bucket storage metrics, with SeaweedFS-aware sizing
when available.
* Helm chart additions: WorkloadMonitor resource and instance label on
BucketClaim templates.
* **Tests**
* Unit tests covering bucket-monitor flows, metric querying, URL
resolution, and requeue behavior.
* **Chores**
* Updated module dependencies to enable COSI integration.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->
This commit is contained in:
Timofei Larkin
GitHub
commit
4053fad9ed
7 changed files with 665 additions and 4 deletions
|
|
@ -42,6 +42,7 @@ import (
|
|||
"github.com/cozystack/cozystack/internal/telemetry"
|
||||
|
||||
helmv2 "github.com/fluxcd/helm-controller/api/v2"
|
||||
cosiv1alpha1 "sigs.k8s.io/container-object-storage-interface-api/apis/objectstorage/v1alpha1"
|
||||
// +kubebuilder:scaffold:imports
|
||||
)
|
||||
|
||||
|
|
@ -56,6 +57,7 @@ func init() {
|
|||
utilruntime.Must(cozystackiov1alpha1.AddToScheme(scheme))
|
||||
utilruntime.Must(dashboard.AddToScheme(scheme))
|
||||
utilruntime.Must(helmv2.AddToScheme(scheme))
|
||||
utilruntime.Must(cosiv1alpha1.AddToScheme(scheme))
|
||||
// +kubebuilder:scaffold:scheme
|
||||
}
|
||||
|
||||
|
|
|
|||
6
go.mod
6
go.mod
|
|
@ -5,6 +5,7 @@ module github.com/cozystack/cozystack
|
|||
go 1.25.0
|
||||
|
||||
require (
|
||||
github.com/cozystack/cozystack-scheduler/pkg/apis v0.1.1
|
||||
github.com/emicklei/dot v1.10.0
|
||||
github.com/fluxcd/helm-controller/api v1.4.3
|
||||
github.com/fluxcd/source-controller/api v1.7.4
|
||||
|
|
@ -28,8 +29,10 @@ require (
|
|||
k8s.io/klog/v2 v2.130.1
|
||||
k8s.io/kube-openapi v0.0.0-20250710124328-f3f2b991d03b
|
||||
k8s.io/utils v0.0.0-20250820121507-0af2bda4dd1d
|
||||
sigs.k8s.io/container-object-storage-interface-api v0.1.0
|
||||
sigs.k8s.io/controller-runtime v0.22.4
|
||||
sigs.k8s.io/structured-merge-diff/v6 v6.3.0
|
||||
sigs.k8s.io/yaml v1.6.0
|
||||
)
|
||||
|
||||
require (
|
||||
|
|
@ -42,10 +45,8 @@ require (
|
|||
github.com/cespare/xxhash/v2 v2.3.0 // indirect
|
||||
github.com/coreos/go-semver v0.3.1 // indirect
|
||||
github.com/coreos/go-systemd/v22 v22.5.0 // indirect
|
||||
github.com/cozystack/cozystack-scheduler/pkg/apis v0.1.1 // indirect
|
||||
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
|
||||
github.com/emicklei/go-restful/v3 v3.12.2 // indirect
|
||||
github.com/evanphx/json-patch v4.12.0+incompatible // indirect
|
||||
github.com/evanphx/json-patch/v5 v5.9.11 // indirect
|
||||
github.com/felixge/httpsnoop v1.0.4 // indirect
|
||||
github.com/fluxcd/pkg/apis/acl v0.9.0 // indirect
|
||||
|
|
@ -126,7 +127,6 @@ require (
|
|||
sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.31.2 // indirect
|
||||
sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730 // indirect
|
||||
sigs.k8s.io/randfill v1.0.0 // indirect
|
||||
sigs.k8s.io/yaml v1.6.0 // indirect
|
||||
)
|
||||
|
||||
// See: issues.k8s.io/135537
|
||||
|
|
|
|||
2
go.sum
2
go.sum
|
|
@ -321,6 +321,8 @@ k8s.io/utils v0.0.0-20250820121507-0af2bda4dd1d h1:wAhiDyZ4Tdtt7e46e9M5ZSAJ/MnPG
|
|||
k8s.io/utils v0.0.0-20250820121507-0af2bda4dd1d/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=
|
||||
sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.31.2 h1:jpcvIRr3GLoUoEKRkHKSmGjxb6lWwrBlJsXc+eUYQHM=
|
||||
sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.31.2/go.mod h1:Ve9uj1L+deCXFrPOk1LpFXqTg7LCFzFso6PA48q/XZw=
|
||||
sigs.k8s.io/container-object-storage-interface-api v0.1.0 h1:8tB6JFQhbQIC1hwGQ+q4+tmSSNfjKemb7bFI6C0CK/4=
|
||||
sigs.k8s.io/container-object-storage-interface-api v0.1.0/go.mod h1:YiB+i/UGkzqgODDhRG3u7jkbWkQcoUeLEJ7hwOT/2Qk=
|
||||
sigs.k8s.io/controller-runtime v0.22.4 h1:GEjV7KV3TY8e+tJ2LCTxUTanW4z/FmNB7l327UfMq9A=
|
||||
sigs.k8s.io/controller-runtime v0.22.4/go.mod h1:+QX1XUpTXN4mLoblf4tqr5CQcyHPAki2HLXqQMY6vh8=
|
||||
sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730 h1:IpInykpT6ceI+QxKBbEflcR5EXP7sU1kvOlxwZh5txg=
|
||||
|
|
|
|||
|
|
@ -4,7 +4,13 @@ import (
|
|||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
|
|
@ -22,8 +28,24 @@ import (
|
|||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
|
||||
cozyv1alpha1 "github.com/cozystack/cozystack/api/v1alpha1"
|
||||
cosiv1alpha1 "sigs.k8s.io/container-object-storage-interface-api/apis/objectstorage/v1alpha1"
|
||||
)
|
||||
|
||||
const (
|
||||
// namespaceMonitoringLabel is the namespace label that indicates which tenant
|
||||
// namespace hosts the monitoring stack (VictoriaMetrics/Prometheus).
|
||||
namespaceMonitoringLabel = "namespace.cozystack.io/monitoring"
|
||||
// vmSelectService is the well-known service name for VictoriaMetrics vmselect
|
||||
// within a monitoring namespace. Port 8481, path /select/0/prometheus.
|
||||
vmSelectService = "vmselect-shortterm"
|
||||
vmSelectPort = "8481"
|
||||
vmSelectPath = "/select/0/prometheus"
|
||||
)
|
||||
|
||||
// prometheusHTTPClient is a dedicated HTTP client for Prometheus queries,
|
||||
// avoiding the shared http.DefaultClient global.
|
||||
var prometheusHTTPClient = &http.Client{Timeout: 10 * time.Second}
|
||||
|
||||
// WorkloadMonitorReconciler reconciles a WorkloadMonitor object
|
||||
type WorkloadMonitorReconciler struct {
|
||||
client.Client
|
||||
|
|
@ -36,6 +58,13 @@ type WorkloadMonitorReconciler struct {
|
|||
// +kubebuilder:rbac:groups=cozystack.io,resources=workloads/status,verbs=get;update;patch
|
||||
// +kubebuilder:rbac:groups=core,resources=pods,verbs=get;list;watch
|
||||
// +kubebuilder:rbac:groups=core,resources=persistentvolumeclaims,verbs=get;list;watch
|
||||
// +kubebuilder:rbac:groups=core,resources=namespaces,verbs=get
|
||||
// +kubebuilder:rbac:groups=objectstorage.k8s.io,resources=bucketclaims,verbs=get;list;watch
|
||||
|
||||
// isBucketClaimReady checks if the BucketClaim has been provisioned.
|
||||
func (r *WorkloadMonitorReconciler) isBucketClaimReady(bc *cosiv1alpha1.BucketClaim) bool {
|
||||
return bc.Status.BucketReady
|
||||
}
|
||||
|
||||
// isServiceReady checks if the service has an external IP bound
|
||||
func (r *WorkloadMonitorReconciler) isServiceReady(svc *corev1.Service) bool {
|
||||
|
|
@ -101,6 +130,185 @@ func updateOwnerReferences(obj metav1.Object, monitor client.Object) {
|
|||
obj.SetOwnerReferences(owners)
|
||||
}
|
||||
|
||||
// resolvePrometheusURL returns the Prometheus-compatible API base URL for the given namespace.
|
||||
// It reads the namespace.cozystack.io/monitoring label to find the monitoring namespace,
|
||||
// then constructs the vmselect URL. Returns empty string if monitoring is not configured.
|
||||
func (r *WorkloadMonitorReconciler) resolvePrometheusURL(ctx context.Context, namespace string) string {
|
||||
logger := log.FromContext(ctx)
|
||||
ns := &corev1.Namespace{}
|
||||
if err := r.Get(ctx, types.NamespacedName{Name: namespace}, ns); err != nil {
|
||||
logger.V(1).Info("Failed to read namespace for monitoring resolution", "namespace", namespace, "error", err)
|
||||
return ""
|
||||
}
|
||||
monitoringNS := ns.Labels[namespaceMonitoringLabel]
|
||||
if monitoringNS == "" {
|
||||
return ""
|
||||
}
|
||||
return fmt.Sprintf("http://%s.%s.svc:%s%s", vmSelectService, monitoringNS, vmSelectPort, vmSelectPath)
|
||||
}
|
||||
|
||||
// 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 SeaweedFS bucket size metrics for the given
|
||||
// bucket names in a single Prometheus query and returns them keyed by bucket
|
||||
// name. The query is scoped to only the requested buckets to avoid fetching
|
||||
// metrics for buckets belonging to other WorkloadMonitors.
|
||||
func (r *WorkloadMonitorReconciler) queryAllBucketMetrics(ctx context.Context, prometheusBaseURL string, bucketNames []string) map[string]*bucketMetrics {
|
||||
result := make(map[string]*bucketMetrics)
|
||||
if prometheusBaseURL == "" || len(bucketNames) == 0 {
|
||||
return result
|
||||
}
|
||||
logger := log.FromContext(ctx)
|
||||
|
||||
query := fmt.Sprintf(`{__name__=~"SeaweedFS_s3_bucket_(size|physical_size)_bytes",bucket=~"%s"}`, strings.Join(bucketNames, "|"))
|
||||
u, err := url.Parse(strings.TrimRight(prometheusBaseURL, "/") + "/api/v1/query")
|
||||
if err != nil {
|
||||
logger.Error(err, "Failed to parse Prometheus URL")
|
||||
return result
|
||||
}
|
||||
u.RawQuery = url.Values{"query": {query}}.Encode()
|
||||
|
||||
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 result
|
||||
}
|
||||
|
||||
resp, err := prometheusHTTPClient.Do(req)
|
||||
if err != nil {
|
||||
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 for bucket metrics", "status", resp.StatusCode)
|
||||
return result
|
||||
}
|
||||
|
||||
body, err := io.ReadAll(io.LimitReader(resp.Body, 4<<20))
|
||||
if err != nil {
|
||||
logger.Error(err, "Failed to read Prometheus response")
|
||||
return result
|
||||
}
|
||||
|
||||
var promResp struct {
|
||||
Status string `json:"status"`
|
||||
Data struct {
|
||||
Result []struct {
|
||||
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 result
|
||||
}
|
||||
if promResp.Status != "success" {
|
||||
return result
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
}
|
||||
|
||||
return result
|
||||
}
|
||||
|
||||
// reconcileBucketClaimForMonitor creates or updates a Workload object for the given BucketClaim and WorkloadMonitor.
|
||||
func (r *WorkloadMonitorReconciler) reconcileBucketClaimForMonitor(
|
||||
ctx context.Context,
|
||||
monitor *cozyv1alpha1.WorkloadMonitor,
|
||||
bc cosiv1alpha1.BucketClaim,
|
||||
allMetrics map[string]*bucketMetrics,
|
||||
) error {
|
||||
logger := log.FromContext(ctx)
|
||||
workload := &cozyv1alpha1.Workload{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: fmt.Sprintf("bucket-%s", bc.Name),
|
||||
Namespace: bc.Namespace,
|
||||
Labels: make(map[string]string, len(bc.Labels)),
|
||||
},
|
||||
}
|
||||
|
||||
resources := make(map[string]resource.Quantity)
|
||||
|
||||
// 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 bm, ok := allMetrics[bc.Status.BucketName]; ok {
|
||||
if bm.HasLogical {
|
||||
resources["s3-storage-bytes"] = *resource.NewQuantity(bm.LogicalSize, resource.BinarySI)
|
||||
}
|
||||
if bm.HasPhysical {
|
||||
resources["s3-physical-storage-bytes"] = *resource.NewQuantity(bm.PhysicalSize, resource.BinarySI)
|
||||
}
|
||||
}
|
||||
|
||||
_, err := ctrl.CreateOrUpdate(ctx, r.Client, workload, func() error {
|
||||
updateOwnerReferences(workload.GetObjectMeta(), &bc)
|
||||
|
||||
if workload.Labels == nil {
|
||||
workload.Labels = make(map[string]string)
|
||||
}
|
||||
for k, v := range bc.Labels {
|
||||
workload.Labels[k] = v
|
||||
}
|
||||
workload.Labels["workloads.cozystack.io/monitor"] = monitor.Name
|
||||
|
||||
workload.Status.Kind = monitor.Spec.Kind
|
||||
workload.Status.Type = monitor.Spec.Type
|
||||
workload.Status.Resources = resources
|
||||
workload.Status.Operational = r.isBucketClaimReady(&bc)
|
||||
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
logger.Error(err, "Failed to CreateOrUpdate Workload", "workload", workload.Name)
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// reconcileServiceForMonitor creates or updates a Workload object for the given Service and WorkloadMonitor.
|
||||
func (r *WorkloadMonitorReconciler) reconcileServiceForMonitor(
|
||||
ctx context.Context,
|
||||
|
|
@ -375,6 +583,34 @@ func (r *WorkloadMonitorReconciler) Reconcile(ctx context.Context, req ctrl.Requ
|
|||
}
|
||||
}
|
||||
|
||||
bucketClaimList := &cosiv1alpha1.BucketClaimList{}
|
||||
if err := r.List(
|
||||
ctx,
|
||||
bucketClaimList,
|
||||
client.InNamespace(monitor.Namespace),
|
||||
client.MatchingLabels(monitor.Spec.Selector),
|
||||
); err != nil {
|
||||
logger.Error(err, "Unable to list BucketClaims for WorkloadMonitor", "monitor", monitor.Name)
|
||||
return ctrl.Result{}, err
|
||||
}
|
||||
|
||||
if len(bucketClaimList.Items) > 0 {
|
||||
bucketPromURL := r.resolvePrometheusURL(ctx, monitor.Namespace)
|
||||
var bucketNames []string
|
||||
for _, bc := range bucketClaimList.Items {
|
||||
if bc.Status.BucketName != "" {
|
||||
bucketNames = append(bucketNames, bc.Status.BucketName)
|
||||
}
|
||||
}
|
||||
allBucketMetrics := r.queryAllBucketMetrics(ctx, bucketPromURL, bucketNames)
|
||||
for _, bc := range bucketClaimList.Items {
|
||||
if err := r.reconcileBucketClaimForMonitor(ctx, monitor, bc, allBucketMetrics); err != nil {
|
||||
logger.Error(err, "Failed to reconcile Workload for BucketClaim", "BucketClaim", bc.Name)
|
||||
continue
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Update WorkloadMonitor status based on observed pods
|
||||
monitor.Status.ObservedReplicas = observedReplicas
|
||||
monitor.Status.AvailableReplicas = availableReplicas
|
||||
|
|
@ -402,7 +638,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 {
|
||||
return ctrl.Result{RequeueAfter: 60 * time.Second}, nil
|
||||
}
|
||||
|
||||
return ctrl.Result{}, nil
|
||||
}
|
||||
|
||||
|
|
@ -421,6 +662,11 @@ func (r *WorkloadMonitorReconciler) SetupWithManager(mgr ctrl.Manager) error {
|
|||
&corev1.PersistentVolumeClaim{},
|
||||
handler.EnqueueRequestsFromMapFunc(mapObjectToMonitor(&corev1.PersistentVolumeClaim{}, r.Client)),
|
||||
).
|
||||
// Watch BucketClaims for S3 bucket billing
|
||||
Watches(
|
||||
&cosiv1alpha1.BucketClaim{},
|
||||
handler.EnqueueRequestsFromMapFunc(mapObjectToMonitor(&cosiv1alpha1.BucketClaim{}, r.Client)),
|
||||
).
|
||||
// Watch for changes to Workload objects we create (owned by WorkloadMonitor)
|
||||
Owns(&cozyv1alpha1.Workload{}).
|
||||
Complete(r)
|
||||
|
|
|
|||
|
|
@ -2,6 +2,9 @@ package controller
|
|||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
cozyv1alpha1 "github.com/cozystack/cozystack/api/v1alpha1"
|
||||
|
|
@ -9,6 +12,7 @@ import (
|
|||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apimachinery/pkg/types"
|
||||
cosiv1alpha1 "sigs.k8s.io/container-object-storage-interface-api/apis/objectstorage/v1alpha1"
|
||||
"sigs.k8s.io/controller-runtime/pkg/client/fake"
|
||||
"sigs.k8s.io/controller-runtime/pkg/reconcile"
|
||||
)
|
||||
|
|
@ -17,6 +21,7 @@ func TestReconcile_OperationalStatusPersisted(t *testing.T) {
|
|||
scheme := runtime.NewScheme()
|
||||
_ = cozyv1alpha1.AddToScheme(scheme)
|
||||
_ = corev1.AddToScheme(scheme)
|
||||
_ = cosiv1alpha1.AddToScheme(scheme)
|
||||
|
||||
minReplicas := int32(2)
|
||||
monitor := &cozyv1alpha1.WorkloadMonitor{
|
||||
|
|
@ -82,6 +87,7 @@ func TestReconcile_OperationalTrue_WhenEnoughReplicas(t *testing.T) {
|
|||
scheme := runtime.NewScheme()
|
||||
_ = cozyv1alpha1.AddToScheme(scheme)
|
||||
_ = corev1.AddToScheme(scheme)
|
||||
_ = cosiv1alpha1.AddToScheme(scheme)
|
||||
|
||||
minReplicas := int32(1)
|
||||
monitor := &cozyv1alpha1.WorkloadMonitor{
|
||||
|
|
@ -139,6 +145,7 @@ func TestReconcile_OperationalTrue_WhenNoMinReplicas(t *testing.T) {
|
|||
scheme := runtime.NewScheme()
|
||||
_ = cozyv1alpha1.AddToScheme(scheme)
|
||||
_ = corev1.AddToScheme(scheme)
|
||||
_ = cosiv1alpha1.AddToScheme(scheme)
|
||||
|
||||
monitor := &cozyv1alpha1.WorkloadMonitor{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
|
|
@ -178,3 +185,392 @@ func TestReconcile_OperationalTrue_WhenNoMinReplicas(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
func newTestScheme() *runtime.Scheme {
|
||||
s := runtime.NewScheme()
|
||||
_ = cozyv1alpha1.AddToScheme(s)
|
||||
_ = corev1.AddToScheme(s)
|
||||
_ = cosiv1alpha1.AddToScheme(s)
|
||||
return s
|
||||
}
|
||||
|
||||
func TestReconcileBucketClaimCreatesWorkload(t *testing.T) {
|
||||
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}
|
||||
req := reconcile.Request{NamespacedName: types.NamespacedName{
|
||||
Name: "my-bucket",
|
||||
Namespace: "tenant-demo",
|
||||
}}
|
||||
|
||||
_, err := reconciler.Reconcile(context.TODO(), req)
|
||||
if err != nil {
|
||||
t.Fatalf("Reconcile returned error: %v", err)
|
||||
}
|
||||
|
||||
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)
|
||||
}
|
||||
|
||||
if workload.Status.Kind != "bucket" {
|
||||
t.Errorf("expected Kind=bucket, got %q", workload.Status.Kind)
|
||||
}
|
||||
if workload.Status.Type != "s3" {
|
||||
t.Errorf("expected Type=s3, got %q", workload.Status.Type)
|
||||
}
|
||||
if !workload.Status.Operational {
|
||||
t.Error("expected Operational=true for ready BucketClaim")
|
||||
}
|
||||
}
|
||||
|
||||
func TestReconcileBucketClaimNotReady(t *testing.T) {
|
||||
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: false,
|
||||
},
|
||||
}
|
||||
|
||||
fakeClient := fake.NewClientBuilder().
|
||||
WithScheme(s).
|
||||
WithObjects(monitor, bc).
|
||||
WithStatusSubresource(monitor).
|
||||
Build()
|
||||
|
||||
reconciler := &WorkloadMonitorReconciler{Client: fakeClient, Scheme: s}
|
||||
req := reconcile.Request{NamespacedName: types.NamespacedName{
|
||||
Name: "my-bucket",
|
||||
Namespace: "tenant-demo",
|
||||
}}
|
||||
|
||||
_, err := reconciler.Reconcile(context.TODO(), req)
|
||||
if err != nil {
|
||||
t.Fatalf("Reconcile returned error: %v", err)
|
||||
}
|
||||
|
||||
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)
|
||||
}
|
||||
|
||||
if workload.Status.Operational {
|
||||
t.Error("expected Operational=false for not-ready BucketClaim")
|
||||
}
|
||||
}
|
||||
|
||||
func TestReconcileNoBucketClaimSkips(t *testing.T) {
|
||||
s := newTestScheme()
|
||||
|
||||
monitor := &cozyv1alpha1.WorkloadMonitor{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: "my-postgres",
|
||||
Namespace: "tenant-demo",
|
||||
},
|
||||
Spec: cozyv1alpha1.WorkloadMonitorSpec{
|
||||
Kind: "postgres",
|
||||
Type: "postgres",
|
||||
Selector: map[string]string{
|
||||
"app.kubernetes.io/instance": "my-postgres",
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
fakeClient := fake.NewClientBuilder().
|
||||
WithScheme(s).
|
||||
WithObjects(monitor).
|
||||
WithStatusSubresource(monitor).
|
||||
Build()
|
||||
|
||||
reconciler := &WorkloadMonitorReconciler{Client: fakeClient, Scheme: s}
|
||||
req := reconcile.Request{NamespacedName: types.NamespacedName{
|
||||
Name: "my-postgres",
|
||||
Namespace: "tenant-demo",
|
||||
}}
|
||||
|
||||
_, err := reconciler.Reconcile(context.TODO(), req)
|
||||
if err != nil {
|
||||
t.Fatalf("Reconcile returned error: %v", err)
|
||||
}
|
||||
|
||||
workloadList := &cozyv1alpha1.WorkloadList{}
|
||||
err = fakeClient.List(context.TODO(), workloadList)
|
||||
if err != nil {
|
||||
t.Fatalf("List returned error: %v", err)
|
||||
}
|
||||
|
||||
for _, w := range workloadList.Items {
|
||||
if w.Status.Kind == "bucket" {
|
||||
t.Error("expected no bucket workloads to be created for postgres monitor")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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":{"__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{}
|
||||
metrics := reconciler.queryAllBucketMetrics(context.TODO(), srv.URL, []string{"bucket-aaa", "bucket-bbb"})
|
||||
|
||||
bm, ok := metrics["bucket-aaa"]
|
||||
if !ok {
|
||||
t.Fatal("expected bucket-aaa in metrics")
|
||||
}
|
||||
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 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{}
|
||||
metrics := reconciler.queryAllBucketMetrics(context.TODO(), srv.URL, []string{"bucket-aaa", "bucket-bbb"})
|
||||
if len(metrics) != 0 {
|
||||
t.Errorf("expected empty metrics, got %d", len(metrics))
|
||||
}
|
||||
}
|
||||
|
||||
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{}
|
||||
metrics := reconciler.queryAllBucketMetrics(context.TODO(), srv.URL, []string{"bucket-aaa", "bucket-bbb"})
|
||||
if len(metrics) != 0 {
|
||||
t.Errorf("expected empty metrics on error, got %d", len(metrics))
|
||||
}
|
||||
}
|
||||
|
||||
func TestQueryAllBucketMetricsNoURL(t *testing.T) {
|
||||
reconciler := &WorkloadMonitorReconciler{}
|
||||
metrics := reconciler.queryAllBucketMetrics(context.TODO(), "", nil)
|
||||
if len(metrics) != 0 {
|
||||
t.Errorf("expected empty metrics when URL is empty, got %d", len(metrics))
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolvePrometheusURL(t *testing.T) {
|
||||
s := newTestScheme()
|
||||
|
||||
ns := &corev1.Namespace{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: "tenant-demo",
|
||||
Labels: map[string]string{
|
||||
"namespace.cozystack.io/monitoring": "tenant-root",
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
fakeClient := fake.NewClientBuilder().
|
||||
WithScheme(s).
|
||||
WithObjects(ns).
|
||||
Build()
|
||||
|
||||
reconciler := &WorkloadMonitorReconciler{Client: fakeClient, Scheme: s}
|
||||
url := reconciler.resolvePrometheusURL(context.TODO(), "tenant-demo")
|
||||
|
||||
expected := "http://vmselect-shortterm.tenant-root.svc:8481/select/0/prometheus"
|
||||
if url != expected {
|
||||
t.Errorf("expected %q, got %q", expected, url)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolvePrometheusURLNoLabel(t *testing.T) {
|
||||
s := newTestScheme()
|
||||
|
||||
ns := &corev1.Namespace{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: "tenant-demo",
|
||||
},
|
||||
}
|
||||
|
||||
fakeClient := fake.NewClientBuilder().
|
||||
WithScheme(s).
|
||||
WithObjects(ns).
|
||||
Build()
|
||||
|
||||
reconciler := &WorkloadMonitorReconciler{Client: fakeClient, Scheme: s}
|
||||
url := reconciler.resolvePrometheusURL(context.TODO(), "tenant-demo")
|
||||
|
||||
if url != "" {
|
||||
t.Errorf("expected empty URL when no monitoring label, got %q", url)
|
||||
}
|
||||
}
|
||||
|
||||
func TestReconcileBucketClaimRequeuesWhenBucketsExist(t *testing.T) {
|
||||
s := newTestScheme()
|
||||
|
||||
ns := &corev1.Namespace{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: "tenant-demo",
|
||||
},
|
||||
}
|
||||
|
||||
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(ns, monitor, bc).
|
||||
WithStatusSubresource(monitor).
|
||||
Build()
|
||||
|
||||
reconciler := &WorkloadMonitorReconciler{Client: fakeClient, Scheme: s}
|
||||
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 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)
|
||||
}
|
||||
|
||||
// Without monitoring label on namespace, no size metrics should be set
|
||||
if _, ok := workload.Status.Resources["s3-storage-bytes"]; ok {
|
||||
t.Error("expected no s3-storage-bytes when monitoring is not configured")
|
||||
}
|
||||
if len(workload.Status.Resources) != 0 {
|
||||
t.Errorf("expected empty resources without monitoring, got %v", workload.Status.Resources)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -4,6 +4,8 @@ apiVersion: objectstorage.k8s.io/v1alpha1
|
|||
kind: BucketClaim
|
||||
metadata:
|
||||
name: {{ .Release.Name }}
|
||||
labels:
|
||||
app.kubernetes.io/instance: {{ .Release.Name }}
|
||||
spec:
|
||||
bucketClassName: {{ $seaweedfs }}{{- if $pool }}-{{ $pool }}{{- end }}{{- if .Values.locking }}-lock{{- end }}
|
||||
protocols:
|
||||
|
|
|
|||
13
packages/apps/bucket/templates/workloadmonitor.yaml
Normal file
13
packages/apps/bucket/templates/workloadmonitor.yaml
Normal file
|
|
@ -0,0 +1,13 @@
|
|||
---
|
||||
apiVersion: cozystack.io/v1alpha1
|
||||
kind: WorkloadMonitor
|
||||
metadata:
|
||||
name: {{ $.Release.Name }}
|
||||
spec:
|
||||
replicas: 0
|
||||
minReplicas: 0
|
||||
kind: bucket
|
||||
type: s3
|
||||
selector:
|
||||
app.kubernetes.io/instance: {{ $.Release.Name }}
|
||||
version: {{ $.Chart.Version }}
|
||||
Loading…
Add table
Add a link
Reference in a new issue