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

feat(controller): add BucketClaim support to WorkloadMonitorReconciler

Browse source 
Add reconcileBucketClaimForMonitor() that watches COSI BucketClaim
objects and creates Workload CRDs with s3-buckets resource, following
the same pattern as PVC and Service reconcilers. This enables the
billing pipeline to discover and track S3 buckets per tenant.

Changes:
- Add COSI API types dependency (container-object-storage-interface-api)
- Register cosiv1alpha1 scheme in controller main
- Add BucketClaim watch in SetupWithManager
- Add BucketClaim list + reconcile in Reconcile loop
- Add RBAC annotation for objectstorage.k8s.io/bucketclaims
- Add unit tests for BucketClaim reconciliation

Co-Authored-By: Claude <noreply@anthropic.com>
Signed-off-by: ZverGuy <maximbel2003@gmail.com>
This commit is contained in:
ZverGuy 2026-04-13 14:41:10 +03:00
parent ee4e71b139
commit 92286213ab
5 changed files with 284 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

2
cmd/cozystack-controller/main.go
View file

@ -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
View file

@ -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
View file

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

73
internal/controller/workloadmonitor_controller.go
View file

@ -22,6 +22,7 @@ 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"
)
// WorkloadMonitorReconciler reconciles a WorkloadMonitor object
@ -36,6 +37,12 @@ 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=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 +108,47 @@ func updateOwnerReferences(obj metav1.Object, monitor client.Object) {
obj.SetOwnerReferences(owners)
}
// 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,
) 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)
resources["s3-buckets"] = resource.MustParse("1")
_, err := ctrl.CreateOrUpdate(ctx, r.Client, workload, func() error {
updateOwnerReferences(workload.GetObjectMeta(), &bc)
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 +423,26 @@ 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 {
if !apierrors.IsNotFound(err) {
logger.Error(err, "Unable to list BucketClaims for WorkloadMonitor", "monitor", monitor.Name)
return ctrl.Result{}, err
}
}
for _, bc := range bucketClaimList.Items {
if err := r.reconcileBucketClaimForMonitor(ctx, monitor, bc); 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
@ -421,6 +489,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)

204
internal/controller/workloadmonitor_controller_test.go
View file

@ -6,9 +6,11 @@ import (
cozyv1alpha1 "github.com/cozystack/cozystack/api/v1alpha1"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/resource"
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 +19,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 +85,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 +143,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 +183,202 @@ 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")
}
qty, ok := workload.Status.Resources["s3-buckets"]
if !ok {
t.Fatal("expected s3-buckets resource to be set")
}
if qty.Cmp(resource.MustParse("1")) != 0 {
t.Errorf("expected s3-buckets=1, got %s", qty.String())
}
}
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")
}
}
}