Files
cozystack/internal/controller/workloadmonitor_controller_test.go
T
Aleksei Sviridkin 81ffcda86f fix(controller): keep publishing monitor status when DataVolumes are unread
The DataVolume List sits on the reconcile path of every
WorkloadMonitor, and returning its error before the status write froze
the replica counts and Operational of a monitor that selects no
DataVolume at all whenever CDI's API hiccupped.

The status is now written either way, with the last DataVolume verdict
kept whenever the DataVolumes could not be read, and the error still
returned so the request is retried. The same applies while no reader
is installed yet, so a stuck disk does not read as ready for the
window between a controller restart and the DataVolume informer
syncing.

Assisted-by: LLM
Signed-off-by: Aleksei Sviridkin <f@lex.la>
2026-10-02 14:02:32 +03:00

1975 lines
63 KiB
Go

package controller
import (
"context"
"errors"
"fmt"
"net/http"
"net/http/httptest"
"slices"
"strings"
"testing"
"time"
cozyv1alpha1 "github.com/cozystack/cozystack/api/v1alpha1"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/meta"
"k8s.io/apimachinery/pkg/api/resource"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/apimachinery/pkg/types"
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/client-go/util/workqueue"
"k8s.io/utils/ptr"
cosiv1alpha1 "sigs.k8s.io/container-object-storage-interface-api/apis/objectstorage/v1alpha1"
"sigs.k8s.io/controller-runtime/pkg/cache/informertest"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/client/fake"
"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
"sigs.k8s.io/controller-runtime/pkg/controller/controllertest"
"sigs.k8s.io/controller-runtime/pkg/reconcile"
"sigs.k8s.io/controller-runtime/pkg/source"
)
func TestReconcile_OperationalStatusPersisted(t *testing.T) {
scheme := runtime.NewScheme()
_ = cozyv1alpha1.AddToScheme(scheme)
_ = corev1.AddToScheme(scheme)
_ = cosiv1alpha1.AddToScheme(scheme)
minReplicas := int32(2)
monitor := &cozyv1alpha1.WorkloadMonitor{
ObjectMeta: metav1.ObjectMeta{
Name: "test-monitor",
Namespace: "default",
},
Spec: cozyv1alpha1.WorkloadMonitorSpec{
Selector: map[string]string{"app": "test"},
MinReplicas: &minReplicas,
},
}
// Create one pod that is ready — availableReplicas=1 < minReplicas=2, so Operational should be false
pod := &corev1.Pod{
ObjectMeta: metav1.ObjectMeta{
Name: "test-pod-1",
Namespace: "default",
Labels: map[string]string{"app": "test"},
},
Status: corev1.PodStatus{
Conditions: []corev1.PodCondition{
{Type: corev1.PodReady, Status: corev1.ConditionTrue},
},
},
}
fakeClient := fake.NewClientBuilder().
WithScheme(scheme).
WithObjects(monitor, pod).
WithStatusSubresource(monitor).
Build()
reconciler := &WorkloadMonitorReconciler{Client: fakeClient, Scheme: scheme}
req := reconcile.Request{NamespacedName: types.NamespacedName{Name: "test-monitor", Namespace: "default"}}
_, err := reconciler.Reconcile(context.TODO(), req)
if err != nil {
t.Fatalf("Reconcile returned error: %v", err)
}
// Fetch the monitor back from fake client and check Operational is persisted
updated := &cozyv1alpha1.WorkloadMonitor{}
if err := fakeClient.Get(context.TODO(), req.NamespacedName, updated); err != nil {
t.Fatalf("Failed to get updated WorkloadMonitor: %v", err)
}
if updated.Status.Operational == nil {
t.Fatal("Expected Operational to be set, got nil")
}
if *updated.Status.Operational {
t.Error("Expected Operational=false (1 available < 2 minReplicas), got true")
}
if updated.Status.ObservedReplicas != 1 {
t.Errorf("Expected ObservedReplicas=1, got %d", updated.Status.ObservedReplicas)
}
if updated.Status.AvailableReplicas != 1 {
t.Errorf("Expected AvailableReplicas=1, got %d", updated.Status.AvailableReplicas)
}
}
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{
ObjectMeta: metav1.ObjectMeta{
Name: "test-monitor",
Namespace: "default",
},
Spec: cozyv1alpha1.WorkloadMonitorSpec{
Selector: map[string]string{"app": "test"},
MinReplicas: &minReplicas,
},
}
pod := &corev1.Pod{
ObjectMeta: metav1.ObjectMeta{
Name: "test-pod-1",
Namespace: "default",
Labels: map[string]string{"app": "test"},
},
Status: corev1.PodStatus{
Conditions: []corev1.PodCondition{
{Type: corev1.PodReady, Status: corev1.ConditionTrue},
},
},
}
fakeClient := fake.NewClientBuilder().
WithScheme(scheme).
WithObjects(monitor, pod).
WithStatusSubresource(monitor).
Build()
reconciler := &WorkloadMonitorReconciler{Client: fakeClient, Scheme: scheme}
req := reconcile.Request{NamespacedName: types.NamespacedName{Name: "test-monitor", Namespace: "default"}}
_, err := reconciler.Reconcile(context.TODO(), req)
if err != nil {
t.Fatalf("Reconcile returned error: %v", err)
}
updated := &cozyv1alpha1.WorkloadMonitor{}
if err := fakeClient.Get(context.TODO(), req.NamespacedName, updated); err != nil {
t.Fatalf("Failed to get updated WorkloadMonitor: %v", err)
}
if updated.Status.Operational == nil {
t.Fatal("Expected Operational to be set, got nil")
}
if !*updated.Status.Operational {
t.Error("Expected Operational=true (1 available >= 1 minReplicas), got false")
}
}
func TestGetMonitorLabels(t *testing.T) {
tests := []struct {
name string
labels map[string]string
expected map[string]string
}{
{
name: "nil labels",
labels: nil,
expected: map[string]string{},
},
{
name: "only workloads.cozystack.io/* labels are propagated",
labels: map[string]string{
"workloads.cozystack.io/resource-preset": "medium",
"app.kubernetes.io/name": "postgres",
"custom.example.com/team": "platform",
},
expected: map[string]string{
"workloads.cozystack.io/resource-preset": "medium",
},
},
{
name: "monitor label is reserved and excluded",
labels: map[string]string{
"workloads.cozystack.io/resource-preset": "small",
"workloads.cozystack.io/monitor": "should-be-dropped",
},
expected: map[string]string{
"workloads.cozystack.io/resource-preset": "small",
},
},
{
name: "multiple workloads.cozystack.io labels propagate",
labels: map[string]string{
"workloads.cozystack.io/resource-preset": "large",
"workloads.cozystack.io/tier": "db",
},
expected: map[string]string{
"workloads.cozystack.io/resource-preset": "large",
"workloads.cozystack.io/tier": "db",
},
},
{
name: "no matching labels returns empty map",
labels: map[string]string{
"app.kubernetes.io/name": "postgres",
},
expected: map[string]string{},
},
}
r := &WorkloadMonitorReconciler{}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
monitor := &cozyv1alpha1.WorkloadMonitor{
ObjectMeta: metav1.ObjectMeta{Labels: tc.labels},
}
got := r.getMonitorLabels(monitor)
if len(got) != len(tc.expected) {
t.Fatalf("expected %d labels, got %d (%v)", len(tc.expected), len(got), got)
}
for k, v := range tc.expected {
if gv, ok := got[k]; !ok || gv != v {
t.Errorf("expected label %q=%q, got %q", k, v, gv)
}
}
})
}
}
func TestReconcile_MonitorLabelsPropagatedToPodWorkload(t *testing.T) {
scheme := runtime.NewScheme()
_ = cozyv1alpha1.AddToScheme(scheme)
_ = corev1.AddToScheme(scheme)
_ = cosiv1alpha1.AddToScheme(scheme)
monitor := &cozyv1alpha1.WorkloadMonitor{
ObjectMeta: metav1.ObjectMeta{
Name: "test-monitor",
Namespace: "default",
Labels: map[string]string{
"workloads.cozystack.io/resource-preset": "medium",
"app.kubernetes.io/name": "ignored-not-propagated",
},
},
Spec: cozyv1alpha1.WorkloadMonitorSpec{
Selector: map[string]string{"app": "test"},
Kind: "postgres",
Type: "postgres",
},
}
pod := &corev1.Pod{
ObjectMeta: metav1.ObjectMeta{
Name: "test-pod-1",
Namespace: "default",
Labels: map[string]string{
"app": "test",
"app.kubernetes.io/name": "pod-wins-on-conflict",
},
},
Status: corev1.PodStatus{
Conditions: []corev1.PodCondition{
{Type: corev1.PodReady, Status: corev1.ConditionTrue},
},
},
}
fakeClient := fake.NewClientBuilder().
WithScheme(scheme).
WithObjects(monitor, pod).
WithStatusSubresource(monitor).
Build()
reconciler := &WorkloadMonitorReconciler{Client: fakeClient, Scheme: scheme}
req := reconcile.Request{NamespacedName: types.NamespacedName{Name: "test-monitor", Namespace: "default"}}
if _, err := reconciler.Reconcile(context.TODO(), req); err != nil {
t.Fatalf("Reconcile returned error: %v", err)
}
workload := &cozyv1alpha1.Workload{}
if err := fakeClient.Get(context.TODO(), types.NamespacedName{Name: "pod-test-pod-1", Namespace: "default"}, workload); err != nil {
t.Fatalf("Failed to get Workload: %v", err)
}
if got := workload.Labels["workloads.cozystack.io/resource-preset"]; got != "medium" {
t.Errorf("expected monitor label propagated, got %q", got)
}
// Non-workloads.cozystack.io monitor labels must not be copied
if _, ok := workload.Labels["app.kubernetes.io/name"]; !ok {
t.Error("expected pod label to be present on Workload")
}
// Source-object label takes precedence on conflict
if got := workload.Labels["app.kubernetes.io/name"]; got != "pod-wins-on-conflict" {
t.Errorf("expected pod label to win on conflict, got %q", got)
}
// Reserved monitor label is always set from the monitor name
if got := workload.Labels["workloads.cozystack.io/monitor"]; got != "test-monitor" {
t.Errorf("expected monitor-name label, got %q", got)
}
}
func TestReconcile_BackwardCompat_NoMonitorLabels(t *testing.T) {
scheme := runtime.NewScheme()
_ = cozyv1alpha1.AddToScheme(scheme)
_ = corev1.AddToScheme(scheme)
_ = cosiv1alpha1.AddToScheme(scheme)
monitor := &cozyv1alpha1.WorkloadMonitor{
ObjectMeta: metav1.ObjectMeta{
Name: "test-monitor",
Namespace: "default",
},
Spec: cozyv1alpha1.WorkloadMonitorSpec{
Selector: map[string]string{"app": "test"},
},
}
pod := &corev1.Pod{
ObjectMeta: metav1.ObjectMeta{
Name: "test-pod-1",
Namespace: "default",
Labels: map[string]string{"app": "test"},
},
Status: corev1.PodStatus{
Conditions: []corev1.PodCondition{
{Type: corev1.PodReady, Status: corev1.ConditionTrue},
},
},
}
fakeClient := fake.NewClientBuilder().
WithScheme(scheme).
WithObjects(monitor, pod).
WithStatusSubresource(monitor).
Build()
reconciler := &WorkloadMonitorReconciler{Client: fakeClient, Scheme: scheme}
req := reconcile.Request{NamespacedName: types.NamespacedName{Name: "test-monitor", Namespace: "default"}}
if _, err := reconciler.Reconcile(context.TODO(), req); err != nil {
t.Fatalf("Reconcile returned error: %v", err)
}
workload := &cozyv1alpha1.Workload{}
if err := fakeClient.Get(context.TODO(), types.NamespacedName{Name: "pod-test-pod-1", Namespace: "default"}, workload); err != nil {
t.Fatalf("Failed to get Workload: %v", err)
}
for k := range workload.Labels {
if strings.HasPrefix(k, "workloads.cozystack.io/") && k != "workloads.cozystack.io/monitor" {
t.Errorf("unexpected workload label present: %q", k)
}
}
}
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{
Name: "test-monitor",
Namespace: "default",
},
Spec: cozyv1alpha1.WorkloadMonitorSpec{
Selector: map[string]string{"app": "test"},
// No MinReplicas — should default to operational=true
},
}
fakeClient := fake.NewClientBuilder().
WithScheme(scheme).
WithObjects(monitor).
WithStatusSubresource(monitor).
Build()
reconciler := &WorkloadMonitorReconciler{Client: fakeClient, Scheme: scheme}
req := reconcile.Request{NamespacedName: types.NamespacedName{Name: "test-monitor", Namespace: "default"}}
_, err := reconciler.Reconcile(context.TODO(), req)
if err != nil {
t.Fatalf("Reconcile returned error: %v", err)
}
updated := &cozyv1alpha1.WorkloadMonitor{}
if err := fakeClient.Get(context.TODO(), req.NamespacedName, updated); err != nil {
t.Fatalf("Failed to get updated WorkloadMonitor: %v", err)
}
if updated.Status.Operational == nil {
t.Fatal("Expected Operational to be set, got nil")
}
if !*updated.Status.Operational {
t.Error("Expected Operational=true (no MinReplicas constraint), got false")
}
}
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 TestReconcileBucketClaim_BucketClassLabelPropagated(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-encrypted",
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)
}
got := workload.Labels["workloads.cozystack.io/bucket-class"]
if got != "seaweedfs-encrypted" {
t.Errorf("expected workloads.cozystack.io/bucket-class=%q, got %q", "seaweedfs-encrypted", got)
}
}
func TestReconcileBucketClaim_InvalidBucketClassSkipsLabel(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",
},
},
}
overlong := strings.Repeat("a", 64)
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: overlong,
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 despite invalid bucket class, got error: %v", err)
}
if v, ok := workload.Labels["workloads.cozystack.io/bucket-class"]; ok {
t.Errorf("expected bucket-class label to be skipped for an invalid label value, got %q", v)
}
}
func TestReconcileBucketClaim_InvalidBucketClassDropsStaleLabel(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",
},
},
}
overlong := strings.Repeat("a", 64)
bc := &cosiv1alpha1.BucketClaim{
ObjectMeta: metav1.ObjectMeta{
Name: "my-bucket",
Namespace: "tenant-demo",
Labels: map[string]string{
"app.kubernetes.io/instance": "my-bucket",
"workloads.cozystack.io/bucket-class": "seaweedfs-encrypted",
},
},
Spec: cosiv1alpha1.BucketClaimSpec{
BucketClassName: overlong,
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",
}}
if _, err := reconciler.Reconcile(context.TODO(), req); err != nil {
t.Fatalf("Reconcile returned error: %v", err)
}
workload := &cozyv1alpha1.Workload{}
if err := fakeClient.Get(context.TODO(), types.NamespacedName{
Name: "bucket-my-bucket",
Namespace: "tenant-demo",
}, workload); err != nil {
t.Fatalf("expected Workload to be created, got error: %v", err)
}
if v, ok := workload.Labels["workloads.cozystack.io/bucket-class"]; ok {
t.Errorf("expected stale bucket-class label to be dropped, got %q", v)
}
}
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 TestReconcile_MonitorLabelsPropagatedToBucketClaimWorkload(t *testing.T) {
s := newTestScheme()
monitor := &cozyv1alpha1.WorkloadMonitor{
ObjectMeta: metav1.ObjectMeta{
Name: "my-bucket",
Namespace: "tenant-demo",
Labels: map[string]string{
"workloads.cozystack.io/resource-preset": "medium",
"app.kubernetes.io/name": "ignored-not-propagated",
},
},
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",
"app.kubernetes.io/name": "bucket-wins-on-conflict",
},
},
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",
}}
if _, err := reconciler.Reconcile(context.TODO(), req); err != nil {
t.Fatalf("Reconcile returned error: %v", err)
}
workload := &cozyv1alpha1.Workload{}
if err := fakeClient.Get(context.TODO(), types.NamespacedName{
Name: "bucket-my-bucket",
Namespace: "tenant-demo",
}, workload); err != nil {
t.Fatalf("Failed to get Workload: %v", err)
}
if got := workload.Labels["workloads.cozystack.io/resource-preset"]; got != "medium" {
t.Errorf("expected monitor label propagated, got %q", got)
}
// Source-object label takes precedence on conflict
if got := workload.Labels["app.kubernetes.io/name"]; got != "bucket-wins-on-conflict" {
t.Errorf("expected bucket claim label to win on conflict, got %q", got)
}
// Reserved monitor label is always set from the monitor name
if got := workload.Labels["workloads.cozystack.io/monitor"]; got != "my-bucket" {
t.Errorf("expected monitor-name label, got %q", got)
}
}
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, err := reconciler.queryAllBucketMetrics(context.TODO(), srv.URL, []string{"bucket-aaa", "bucket-bbb"})
if err != nil {
t.Fatalf("queryAllBucketMetrics returned error: %v", err)
}
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, err := reconciler.queryAllBucketMetrics(context.TODO(), srv.URL, []string{"bucket-aaa", "bucket-bbb"})
if err != nil {
t.Fatalf("queryAllBucketMetrics returned error: %v", err)
}
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{}
_, err := reconciler.queryAllBucketMetrics(context.TODO(), srv.URL, []string{"bucket-aaa", "bucket-bbb"})
if err == nil {
t.Error("expected error on server failure, got nil")
}
}
func TestQueryAllBucketMetricsNoURL(t *testing.T) {
reconciler := &WorkloadMonitorReconciler{}
metrics, err := reconciler.queryAllBucketMetrics(context.TODO(), "", nil)
if err != nil {
t.Fatalf("queryAllBucketMetrics returned error: %v", err)
}
if len(metrics) != 0 {
t.Errorf("expected empty metrics when URL is empty, got %d", len(metrics))
}
}
func TestQueryAllBucketMetricsPathPrefixAndUserinfo(t *testing.T) {
var gotPath, gotUser, gotPass string
var gotOK bool
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
gotPath = r.URL.Path
gotUser, gotPass, gotOK = r.BasicAuth()
fmt.Fprint(w, `{"status":"success","data":{"resultType":"vector","result":[]}}`)
}))
defer srv.Close()
reconciler := &WorkloadMonitorReconciler{}
// Trailing slash on the base URL must not produce a double slash before
// /api/v1/query, and userinfo must arrive as basic auth.
baseURL := strings.Replace(srv.URL, "http://", "http://billing:hunter2@", 1) + "/path/to/prometheus/"
if _, err := reconciler.queryAllBucketMetrics(context.TODO(), baseURL, []string{"bucket-aaa"}); err != nil {
t.Fatalf("queryAllBucketMetrics returned error: %v", err)
}
if gotPath != "/path/to/prometheus/api/v1/query" {
t.Errorf("expected path prefix preserved, got %q", gotPath)
}
if !gotOK || gotUser != "billing" || gotPass != "hunter2" {
t.Errorf("expected basic auth billing/hunter2 from userinfo, got %q/%q ok=%v", gotUser, gotPass, gotOK)
}
}
func TestParseMetricsEndpointURL(t *testing.T) {
valid := map[string]string{
"https://vm.example.com": "https://vm.example.com",
"https://vm.example.com/": "https://vm.example.com",
"https://vm.example.com:8427/vm/prometheus": "https://vm.example.com:8427/vm/prometheus",
"https://user:pass@vm.example.com/prometheus": "https://user:pass@vm.example.com/prometheus",
"http://vm.example.com/select/0/prometheus///": "http://vm.example.com/select/0/prometheus",
}
for raw, want := range valid {
got, err := ParseMetricsEndpointURL(raw)
if err != nil {
t.Errorf("ParseMetricsEndpointURL(%q) returned error: %v", raw, err)
continue
}
if got != want {
t.Errorf("ParseMetricsEndpointURL(%q) = %q, want %q", raw, got, want)
}
}
invalid := []string{
"",
"vm.example.com/prometheus",
"ftp://vm.example.com",
"https://",
"https://vm.example.com/prometheus?query=up",
"https://vm.example.com/prometheus#frag",
}
for _, raw := range invalid {
if got, err := ParseMetricsEndpointURL(raw); err == nil {
t.Errorf("ParseMetricsEndpointURL(%q) = %q, want error", raw, got)
}
}
}
func TestResolvePrometheusURLFromLabel(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, err := reconciler.resolvePrometheusURL(context.TODO(), "tenant-demo")
if err != nil {
t.Fatalf("resolvePrometheusURL returned error: %v", err)
}
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, err := reconciler.resolvePrometheusURL(context.TODO(), "tenant-demo")
if err != nil {
t.Fatalf("resolvePrometheusURL returned error: %v", err)
}
if url != "" {
t.Errorf("expected empty URL when no monitoring label, got %q", url)
}
}
func TestResolvePrometheusURLEndpointOverrideWinsOverLabel(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,
SeaweedfsMetricsEndpoint: "https://vm.example.com/path/to/prometheus",
}
url, err := reconciler.resolvePrometheusURL(context.TODO(), "tenant-demo")
if err != nil {
t.Fatalf("resolvePrometheusURL returned error: %v", err)
}
if url != "https://vm.example.com/path/to/prometheus" {
t.Errorf("expected the endpoint override to win over label discovery, 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)
}
}
// newBucketFixture returns a plain namespace, a bucket WorkloadMonitor, and a
// ready BucketClaim named my-bucket whose SeaweedFS bucket is cosi-abc123.
func newBucketFixture() (*corev1.Namespace, *cozyv1alpha1.WorkloadMonitor, *cosiv1alpha1.BucketClaim) {
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",
},
}
return ns, monitor, bc
}
func TestReconcileBucketMetricsFromEndpointOverride(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/vm/select/0/prometheus/api/v1/query" {
t.Errorf("unexpected query path %q", r.URL.Path)
}
fmt.Fprint(w, `{"status":"success","data":{"resultType":"vector","result":[
{"metric":{"__name__":"SeaweedFS_s3_bucket_size_bytes","bucket":"cosi-abc123"},"value":[1713000000,"4096"]}
]}}`)
}))
defer srv.Close()
s := newTestScheme()
ns, monitor, bc := newBucketFixture()
fakeClient := fake.NewClientBuilder().
WithScheme(s).
WithObjects(ns, monitor, bc).
WithStatusSubresource(monitor).
Build()
reconciler := &WorkloadMonitorReconciler{
Client: fakeClient,
Scheme: s,
SeaweedfsMetricsEndpoint: srv.URL + "/vm/select/0/prometheus",
}
req := reconcile.Request{NamespacedName: types.NamespacedName{
Name: "my-bucket",
Namespace: "tenant-demo",
}}
if _, err := reconciler.Reconcile(context.TODO(), req); err != nil {
t.Fatalf("Reconcile returned error: %v", err)
}
workload := &cozyv1alpha1.Workload{}
if err := fakeClient.Get(context.TODO(), types.NamespacedName{
Name: "bucket-my-bucket",
Namespace: "tenant-demo",
}, workload); err != nil {
t.Fatalf("Failed to get Workload: %v", err)
}
q, ok := workload.Status.Resources["s3-storage-bytes"]
if !ok {
t.Fatal("expected s3-storage-bytes from the overridden endpoint")
}
if q.Value() != 4096 {
t.Errorf("expected s3-storage-bytes=4096, got %d", q.Value())
}
}
func TestReconcileRetainsLastKnownSizesOnQueryFailure(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusBadGateway)
}))
defer srv.Close()
s := newTestScheme()
ns, monitor, bc := newBucketFixture()
// A Workload from a previous successful reconcile already carries sizes.
workload := &cozyv1alpha1.Workload{
ObjectMeta: metav1.ObjectMeta{
Name: "bucket-my-bucket",
Namespace: "tenant-demo",
},
Status: cozyv1alpha1.WorkloadStatus{
Kind: "bucket",
Type: "s3",
Resources: map[string]resource.Quantity{
"s3-storage-bytes": *resource.NewQuantity(4096, resource.BinarySI),
"s3-physical-storage-bytes": *resource.NewQuantity(8192, resource.BinarySI),
},
},
}
fakeClient := fake.NewClientBuilder().
WithScheme(s).
WithObjects(ns, monitor, bc, workload).
WithStatusSubresource(monitor).
Build()
reconciler := &WorkloadMonitorReconciler{
Client: fakeClient,
Scheme: s,
SeaweedfsMetricsEndpoint: srv.URL,
}
req := reconcile.Request{NamespacedName: types.NamespacedName{
Name: "my-bucket",
Namespace: "tenant-demo",
}}
// The failure must be loud: the reconcile returns an error instead of
// pretending the buckets are empty.
if _, err := reconciler.Reconcile(context.TODO(), req); err == nil {
t.Error("expected Reconcile to return an error when the metrics endpoint fails")
}
updated := &cozyv1alpha1.Workload{}
if err := fakeClient.Get(context.TODO(), types.NamespacedName{
Name: "bucket-my-bucket",
Namespace: "tenant-demo",
}, updated); err != nil {
t.Fatalf("Failed to get Workload: %v", err)
}
q, ok := updated.Status.Resources["s3-storage-bytes"]
if !ok || q.Value() != 4096 {
t.Errorf("expected last known s3-storage-bytes=4096 retained, got %v (present=%v)", q.Value(), ok)
}
qp, ok := updated.Status.Resources["s3-physical-storage-bytes"]
if !ok || qp.Value() != 8192 {
t.Errorf("expected last known s3-physical-storage-bytes=8192 retained, got %v (present=%v)", qp.Value(), ok)
}
}
func TestReconcileRetainsLastKnownSizesWhenBucketMissingFromResult(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()
s := newTestScheme()
ns, monitor, bc := newBucketFixture()
workload := &cozyv1alpha1.Workload{
ObjectMeta: metav1.ObjectMeta{
Name: "bucket-my-bucket",
Namespace: "tenant-demo",
},
Status: cozyv1alpha1.WorkloadStatus{
Kind: "bucket",
Type: "s3",
Resources: map[string]resource.Quantity{
"s3-storage-bytes": *resource.NewQuantity(4096, resource.BinarySI),
},
},
}
fakeClient := fake.NewClientBuilder().
WithScheme(s).
WithObjects(ns, monitor, bc, workload).
WithStatusSubresource(monitor).
Build()
reconciler := &WorkloadMonitorReconciler{
Client: fakeClient,
Scheme: s,
SeaweedfsMetricsEndpoint: srv.URL,
}
req := reconcile.Request{NamespacedName: types.NamespacedName{
Name: "my-bucket",
Namespace: "tenant-demo",
}}
if _, err := reconciler.Reconcile(context.TODO(), req); err != nil {
t.Fatalf("Reconcile returned error: %v", err)
}
updated := &cozyv1alpha1.Workload{}
if err := fakeClient.Get(context.TODO(), types.NamespacedName{
Name: "bucket-my-bucket",
Namespace: "tenant-demo",
}, updated); err != nil {
t.Fatalf("Failed to get Workload: %v", err)
}
q, ok := updated.Status.Resources["s3-storage-bytes"]
if !ok || q.Value() != 4096 {
t.Errorf("expected last known s3-storage-bytes=4096 retained when series is absent, got %v (present=%v)", q.Value(), ok)
}
}
func TestGetWorkloadMetadata(t *testing.T) {
tests := []struct {
name string
annotations map[string]string
expected map[string]string
}{
{
name: "nil annotations returns empty map",
annotations: nil,
expected: map[string]string{},
},
{
// KubeVirt stamps the VM preference (which carries the guest OS
// profile, e.g. Windows) under kubevirt.io/cluster-preference-name.
// This value is taken verbatim from a real Windows VMInstance
// virt-launcher pod on a live cluster.
name: "instance profile from cluster-preference-name annotation",
annotations: map[string]string{
"kubevirt.io/cluster-preference-name": "windows.2k22.virtio",
},
expected: map[string]string{
"workloads.cozystack.io/kubevirt-vmi-instance-profile": "windows.2k22.virtio",
},
},
{
name: "instance type from cluster-instancetype-name annotation",
annotations: map[string]string{
"kubevirt.io/cluster-instancetype-name": "cx1.large",
},
expected: map[string]string{
"workloads.cozystack.io/kubevirt-vmi-instance-type": "cx1.large",
},
},
{
name: "both instance type and profile propagate",
annotations: map[string]string{
"kubevirt.io/cluster-instancetype-name": "cx1.large",
"kubevirt.io/cluster-preference-name": "windows.2k22.virtio",
},
expected: map[string]string{
"workloads.cozystack.io/kubevirt-vmi-instance-type": "cx1.large",
"workloads.cozystack.io/kubevirt-vmi-instance-profile": "windows.2k22.virtio",
},
},
{
// KubeVirt never stamps kubevirt.io/cluster-instanceprofile-name, so
// reading it must not yield a profile label.
name: "nonexistent cluster-instanceprofile-name annotation is ignored",
annotations: map[string]string{
"kubevirt.io/cluster-instanceprofile-name": "windows.2k22.virtio",
},
expected: map[string]string{},
},
}
r := &WorkloadMonitorReconciler{}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
pod := &corev1.Pod{
ObjectMeta: metav1.ObjectMeta{Annotations: tc.annotations},
}
got := r.getWorkloadMetadata(pod)
if len(got) != len(tc.expected) {
t.Fatalf("expected %d labels, got %d (%v)", len(tc.expected), len(got), got)
}
for k, v := range tc.expected {
if gv, ok := got[k]; !ok || gv != v {
t.Errorf("expected label %q=%q, got %q", k, v, gv)
}
}
})
}
}
func TestReconcile_PVCStorageClassResourceKey(t *testing.T) {
cases := []struct {
name string
storageClass *string
wantKey string
}{
{"unset falls back to default", nil, "default.storageclass.storage.k8s.io/requests.storage"},
{"empty falls back to default", ptr.To(""), "default.storageclass.storage.k8s.io/requests.storage"},
{"explicit class is kept", ptr.To("replicated"), "replicated.storageclass.storage.k8s.io/requests.storage"},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
scheme := newTestScheme()
monitor := &cozyv1alpha1.WorkloadMonitor{
ObjectMeta: metav1.ObjectMeta{Name: "test-monitor", Namespace: "default"},
Spec: cozyv1alpha1.WorkloadMonitorSpec{
Selector: map[string]string{"app": "test"},
Kind: "postgres",
Type: "postgres",
},
}
pvc := &corev1.PersistentVolumeClaim{
ObjectMeta: metav1.ObjectMeta{
Name: "data",
Namespace: "default",
Labels: map[string]string{"app": "test"},
},
Spec: corev1.PersistentVolumeClaimSpec{StorageClassName: tc.storageClass},
Status: corev1.PersistentVolumeClaimStatus{
Phase: corev1.ClaimBound,
Capacity: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("10Gi")},
},
}
fakeClient := fake.NewClientBuilder().
WithScheme(scheme).
WithObjects(monitor, pvc).
WithStatusSubresource(monitor).
Build()
reconciler := &WorkloadMonitorReconciler{Client: fakeClient, Scheme: scheme}
req := reconcile.Request{NamespacedName: types.NamespacedName{Name: "test-monitor", Namespace: "default"}}
if _, err := reconciler.Reconcile(context.TODO(), req); err != nil {
t.Fatalf("Reconcile returned error: %v", err)
}
workload := &cozyv1alpha1.Workload{}
if err := fakeClient.Get(context.TODO(), types.NamespacedName{Name: "pvc-data", Namespace: "default"}, workload); err != nil {
t.Fatalf("Failed to get Workload: %v", err)
}
if len(workload.Status.Resources) != 1 {
t.Fatalf("expected one resource entry, got %v", workload.Status.Resources)
}
if _, ok := workload.Status.Resources[tc.wantKey]; !ok {
t.Errorf("expected resource key %q, got %v", tc.wantKey, workload.Status.Resources)
}
})
}
}
func newDataVolume(name string, labels map[string]string, phase *string) *unstructured.Unstructured {
dv := &unstructured.Unstructured{}
dv.SetGroupVersionKind(dataVolumeGVK)
dv.SetName(name)
dv.SetNamespace("default")
dv.SetLabels(labels)
if phase != nil {
dv.Object["status"] = map[string]any{"phase": *phase}
}
return dv
}
func reconcileDataVolumeMonitor(t *testing.T, withReader bool, listErr error, dvs ...client.Object) *cozyv1alpha1.WorkloadMonitor {
t.Helper()
s := newTestScheme()
monitor := &cozyv1alpha1.WorkloadMonitor{
ObjectMeta: metav1.ObjectMeta{Name: "vm-disk-test", Namespace: "default"},
Spec: cozyv1alpha1.WorkloadMonitorSpec{
Selector: map[string]string{"app.kubernetes.io/instance": "vm-disk-test"},
MinReplicas: ptr.To[int32](0),
},
}
builder := fake.NewClientBuilder().
WithScheme(s).
WithObjects(append([]client.Object{monitor}, dvs...)...).
WithStatusSubresource(monitor)
if listErr != nil {
builder = builder.WithInterceptorFuncs(interceptor.Funcs{
List: func(ctx context.Context, c client.WithWatch, list client.ObjectList, opts ...client.ListOption) error {
if u, ok := list.(*unstructured.UnstructuredList); ok && u.GroupVersionKind().Group == dataVolumeGVK.Group {
return listErr
}
return c.List(ctx, list, opts...)
},
})
}
fakeClient := builder.Build()
reconciler := &WorkloadMonitorReconciler{Client: fakeClient, Scheme: s}
if withReader {
reconciler.DataVolumeReader = fakeClient
}
req := reconcile.Request{NamespacedName: types.NamespacedName{Name: monitor.Name, Namespace: monitor.Namespace}}
if _, err := reconciler.Reconcile(context.TODO(), req); err != nil {
t.Fatalf("Reconcile returned error: %v", err)
}
updated := &cozyv1alpha1.WorkloadMonitor{}
if err := fakeClient.Get(context.TODO(), req.NamespacedName, updated); err != nil {
t.Fatalf("Failed to get updated WorkloadMonitor: %v", err)
}
if updated.Status.Operational == nil {
t.Fatal("Expected Operational to be set, got nil")
}
return updated
}
func TestReconcile_DataVolumePhaseDecidesOperational(t *testing.T) {
selected := map[string]string{"app.kubernetes.io/instance": "vm-disk-test"}
phase := func(p string) *string { return &p }
cases := []struct {
name string
phase *string
operational bool
}{
{"Pending", phase("Pending"), false},
{"PVCBound", phase("PVCBound"), false},
{"ImportScheduled", phase("ImportScheduled"), false},
{"ImportInProgress", phase("ImportInProgress"), false},
{"CloneScheduled", phase("CloneScheduled"), false},
{"CloneInProgress", phase("CloneInProgress"), false},
{"SnapshotForSmartCloneInProgress", phase("SnapshotForSmartCloneInProgress"), false},
{"CloneFromSnapshotSourceInProgress", phase("CloneFromSnapshotSourceInProgress"), false},
{"SmartClonePVCInProgress", phase("SmartClonePVCInProgress"), false},
{"CSICloneInProgress", phase("CSICloneInProgress"), false},
{"PrepClaimInProgress", phase("PrepClaimInProgress"), false},
{"RebindInProgress", phase("RebindInProgress"), false},
{"ExpansionInProgress", phase("ExpansionInProgress"), false},
{"NamespaceTransferInProgress", phase("NamespaceTransferInProgress"), false},
{"UploadScheduled", phase("UploadScheduled"), false},
{"UploadReady", phase("UploadReady"), false},
{"failed", phase("Failed"), false},
{"unknown", phase("Unknown"), false},
{"status without phase", phase(""), false},
{"no status at all", nil, false},
{"succeeded", phase("Succeeded"), true},
{"waiting for the consuming VM", phase("PendingPopulation"), true},
{"wait for first consumer", phase("WaitForFirstConsumer"), true},
{"paused multi-stage import", phase("Paused"), true},
{"phase this controller does not know", phase("SomeFutureCDIPhase"), true},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
dv := newDataVolume("vm-disk-test", selected, tc.phase)
got := reconcileDataVolumeMonitor(t, true, nil, dv)
if *got.Status.Operational != tc.operational {
t.Errorf("Operational = %v, want %v", *got.Status.Operational, tc.operational)
}
})
}
}
func TestReconcile_DataVolumeMessageNamesTheDiskAndPhase(t *testing.T) {
selected := map[string]string{"app.kubernetes.io/instance": "vm-disk-test"}
phase := func(p string) *string { return &p }
cases := []struct {
name string
phase *string
want string
}{
{"in flight", phase("ImportInProgress"), "DataVolume vm-disk-test is ImportInProgress"},
{"failed", phase("Failed"), "DataVolume vm-disk-test is Failed"},
{"no phase", nil, "DataVolume vm-disk-test has no phase"},
{"settled", phase("Succeeded"), ""},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
got := reconcileDataVolumeMonitor(t, true, nil, newDataVolume("vm-disk-test", selected, tc.phase))
if got.Status.Message != tc.want {
t.Errorf("Message = %q, want %q", got.Status.Message, tc.want)
}
})
}
}
func TestReconcile_OneDataVolumeInFlightOutweighsSettledOnes(t *testing.T) {
selected := map[string]string{"app.kubernetes.io/instance": "vm-disk-test"}
done, busy := "Succeeded", "ImportInProgress"
got := reconcileDataVolumeMonitor(t, true, nil,
newDataVolume("a", selected, &done),
newDataVolume("b", selected, &busy),
)
if *got.Status.Operational {
t.Error("Operational = true with one DataVolume still importing, want false")
}
}
// reversedReader lists in reverse, because a cache List guarantees no order.
type reversedReader struct{ client.Reader }
func (r reversedReader) List(ctx context.Context, list client.ObjectList, opts ...client.ListOption) error {
if err := r.Reader.List(ctx, list, opts...); err != nil {
return err
}
items, err := meta.ExtractList(list)
if err != nil {
return err
}
slices.Reverse(items)
return meta.SetList(list, items)
}
func TestDataVolumesMessage_DoesNotDependOnListOrder(t *testing.T) {
selected := map[string]string{"app.kubernetes.io/instance": "vm-disk-test"}
busy, failed := "ImportInProgress", "Failed"
monitor := &cozyv1alpha1.WorkloadMonitor{
ObjectMeta: metav1.ObjectMeta{Name: "vm-disk-test", Namespace: "default"},
Spec: cozyv1alpha1.WorkloadMonitorSpec{Selector: selected},
}
fakeClient := fake.NewClientBuilder().WithScheme(newTestScheme()).
WithObjects(newDataVolume("a", selected, &busy), newDataVolume("b", selected, &failed)).
Build()
const want = "DataVolume a is ImportInProgress; DataVolume b is Failed"
for _, reader := range []client.Reader{fakeClient, reversedReader{fakeClient}} {
r := &WorkloadMonitorReconciler{DataVolumeReader: reader}
got, _, err := r.dataVolumesMessage(context.Background(), monitor)
if err != nil || got != want {
t.Errorf("message = %q, %v; want %q", got, err, want)
}
}
}
func TestReconcile_DataVolumeOutsideSelectorIgnored(t *testing.T) {
busy := "ImportInProgress"
got := reconcileDataVolumeMonitor(t, true, nil,
newDataVolume("other", map[string]string{"app.kubernetes.io/instance": "other"}, &busy),
)
if !*got.Status.Operational {
t.Error("Operational = false from a DataVolume the selector does not match, want true")
}
}
func TestReconcile_NoDataVolumesKeepsOperational(t *testing.T) {
got := reconcileDataVolumeMonitor(t, true, nil)
if !*got.Status.Operational {
t.Error("Operational = false with no DataVolumes and minReplicas 0, want true")
}
}
func TestReconcile_DataVolumeKindNotServedIsNoDataVolumes(t *testing.T) {
noMatch := &meta.NoKindMatchError{GroupKind: dataVolumeGVK.GroupKind(), SearchedVersions: []string{dataVolumeGVK.Version}}
got := reconcileDataVolumeMonitor(t, true, noMatch)
if !*got.Status.Operational {
t.Error("Operational = false when the DataVolume kind is not served, want true")
}
}
// reconcileWithUnreadableDataVolumes reconciles a monitor that selects one
// ready pod while every DataVolume List fails, or while withReader is false.
func reconcileWithUnreadableDataVolumes(t *testing.T, withReader bool, status cozyv1alpha1.WorkloadMonitorStatus) (*cozyv1alpha1.WorkloadMonitor, error) {
t.Helper()
s := newTestScheme()
selector := map[string]string{"app.kubernetes.io/instance": "m"}
monitor := &cozyv1alpha1.WorkloadMonitor{
ObjectMeta: metav1.ObjectMeta{Name: "m", Namespace: "default"},
Spec: cozyv1alpha1.WorkloadMonitorSpec{Selector: selector, MinReplicas: ptr.To[int32](1)},
Status: status,
}
pod := &corev1.Pod{
ObjectMeta: metav1.ObjectMeta{Name: "m-0", Namespace: "default", Labels: selector},
Status: corev1.PodStatus{Conditions: []corev1.PodCondition{
{Type: corev1.PodReady, Status: corev1.ConditionTrue},
}},
}
fakeClient := fake.NewClientBuilder().
WithScheme(s).
WithObjects(monitor, pod).
WithStatusSubresource(monitor).
WithInterceptorFuncs(interceptor.Funcs{
List: func(ctx context.Context, c client.WithWatch, list client.ObjectList, opts ...client.ListOption) error {
if _, ok := list.(*unstructured.UnstructuredList); ok {
return fmt.Errorf("apiserver unavailable")
}
return c.List(ctx, list, opts...)
},
}).
Build()
reconciler := &WorkloadMonitorReconciler{Client: fakeClient, Scheme: s}
if withReader {
reconciler.DataVolumeReader = fakeClient
}
req := reconcile.Request{NamespacedName: types.NamespacedName{Name: "m", Namespace: "default"}}
_, err := reconciler.Reconcile(context.TODO(), req)
updated := &cozyv1alpha1.WorkloadMonitor{}
if gerr := fakeClient.Get(context.TODO(), req.NamespacedName, updated); gerr != nil {
t.Fatalf("Failed to get updated WorkloadMonitor: %v", gerr)
}
return updated, err
}
func TestReconcile_DataVolumeListErrorStillPublishesStatus(t *testing.T) {
got, err := reconcileWithUnreadableDataVolumes(t, true, cozyv1alpha1.WorkloadMonitorStatus{})
if err == nil {
t.Error("Reconcile returned nil on a DataVolume list failure, want the error so the request is retried")
}
if got.Status.ObservedReplicas != 1 || got.Status.AvailableReplicas != 1 {
t.Errorf("replicas observed=%d available=%d, want 1 and 1", got.Status.ObservedReplicas, got.Status.AvailableReplicas)
}
if got.Status.Operational == nil || !*got.Status.Operational || got.Status.Message != "" {
t.Errorf("Operational=%v Message=%q, want a monitor with no DataVolume verdict to stay operational", got.Status.Operational, got.Status.Message)
}
}
func TestReconcile_UnreadDataVolumesKeepTheLastVerdict(t *testing.T) {
const stuck = "DataVolume m is ImportInProgress"
for _, withReader := range []bool{true, false} {
t.Run(fmt.Sprintf("reader=%v", withReader), func(t *testing.T) {
got, _ := reconcileWithUnreadableDataVolumes(t, withReader, cozyv1alpha1.WorkloadMonitorStatus{
Operational: ptr.To(false),
Message: stuck,
})
if got.Status.ObservedReplicas != 1 || got.Status.AvailableReplicas != 1 {
t.Errorf("replicas observed=%d available=%d, want 1 and 1", got.Status.ObservedReplicas, got.Status.AvailableReplicas)
}
if got.Status.Operational == nil || *got.Status.Operational || got.Status.Message != stuck {
t.Errorf("Operational=%v Message=%q, want false and %q", got.Status.Operational, got.Status.Message, stuck)
}
})
}
}
func TestReconcile_NoDataVolumeReaderSkipsDataVolumes(t *testing.T) {
busy := "ImportInProgress"
got := reconcileDataVolumeMonitor(t, false, nil,
newDataVolume("vm-disk-test", map[string]string{"app.kubernetes.io/instance": "vm-disk-test"}, &busy),
)
if !*got.Status.Operational {
t.Error("Operational = false without a DataVolume reader, want true")
}
}
func TestDataVolumeAPIServed(t *testing.T) {
empty := meta.NewDefaultRESTMapper(nil)
served, err := dataVolumeAPIServed(empty)
if err != nil || served {
t.Errorf("empty mapper: served=%v err=%v, want false, nil", served, err)
}
withCDI := meta.NewDefaultRESTMapper(nil)
withCDI.Add(dataVolumeGVK, meta.RESTScopeNamespace)
served, err = dataVolumeAPIServed(withCDI)
if err != nil || !served {
t.Errorf("mapper with DataVolume: served=%v err=%v, want true, nil", served, err)
}
}
type failingRESTMapper struct{ meta.RESTMapper }
func (failingRESTMapper) RESTMapping(schema.GroupKind, ...string) (*meta.RESTMapping, error) {
return nil, fmt.Errorf("discovery unavailable")
}
func TestDataVolumeAPIServed_DiscoveryErrorIsReturned(t *testing.T) {
served, err := dataVolumeAPIServed(failingRESTMapper{meta.NewDefaultRESTMapper(nil)})
if err == nil || served {
t.Errorf("served=%v err=%v, want false and the discovery error", served, err)
}
}
// startingWatcher starts every source it is given, as a controller that has
// already started its sources does.
type startingWatcher struct {
sources []source.Source
err error
}
func (w *startingWatcher) Watch(src source.Source) error {
if w.err != nil {
return w.err
}
w.sources = append(w.sources, src)
queue := workqueue.NewTypedRateLimitingQueue(workqueue.DefaultTypedControllerRateLimiter[reconcile.Request]())
return src.Start(context.Background(), queue)
}
func servedMapper() meta.RESTMapper {
mapper := meta.NewDefaultRESTMapper(nil)
mapper.Add(dataVolumeGVK, meta.RESTScopeNamespace)
return mapper
}
func fakeDataVolumeInformer(t *testing.T, informers *informertest.FakeInformers) *controllertest.FakeInformer {
t.Helper()
dv := &unstructured.Unstructured{}
dv.SetGroupVersionKind(dataVolumeGVK)
fi, err := informers.FakeInformerFor(context.Background(), dv)
if err != nil {
t.Fatal(err)
}
return fi
}
func TestTryStartDataVolumeWatch_WaitsUntilTheKindIsServed(t *testing.T) {
r := &WorkloadMonitorReconciler{}
w := &startingWatcher{}
mapper := meta.NewDefaultRESTMapper(nil)
informers := &informertest.FakeInformers{}
done, err := r.tryStartDataVolumeWatch(context.Background(), mapper, w, informers, time.Second)
if err != nil || done {
t.Fatalf("before CDI: done=%v err=%v, want false, nil", done, err)
}
if len(w.sources) != 0 || r.dataVolumeReader() != nil {
t.Fatalf("before CDI: %d watches, reader %v; want none", len(w.sources), r.dataVolumeReader())
}
mapper.Add(dataVolumeGVK, meta.RESTScopeNamespace)
done, err = r.tryStartDataVolumeWatch(context.Background(), mapper, w, informers, time.Second)
if err != nil || !done {
t.Fatalf("after CDI: done=%v err=%v, want true, nil", done, err)
}
if len(w.sources) != 1 || r.dataVolumeReader() == nil {
t.Fatalf("after CDI: %d watches, reader %v; want one watch and a reader", len(w.sources), r.dataVolumeReader())
}
}
// Before the controller has started its sources, Watch only queues a source,
// and WaitForSync on a source nobody started cannot report anything, so the
// watch waits for the first reconcile. It starts right after it rather than on
// the next discovery poll, so a disk created just after a restart is not read
// as populated for a whole poll interval.
func TestWatchDataVolumes_StartsOnTheFirstReconcile(t *testing.T) {
s := newTestScheme()
monitor := &cozyv1alpha1.WorkloadMonitor{
ObjectMeta: metav1.ObjectMeta{Name: "m", Namespace: "default"},
Spec: cozyv1alpha1.WorkloadMonitorSpec{MinReplicas: ptr.To[int32](0)},
}
fakeClient := fake.NewClientBuilder().WithScheme(s).WithObjects(monitor).WithStatusSubresource(monitor).Build()
r := &WorkloadMonitorReconciler{Client: fakeClient, Scheme: s}
informers := &informertest.FakeInformers{}
fakeDataVolumeInformer(t, informers)
ctx := t.Context()
go r.watchDataVolumes(ctx, servedMapper(), &startingWatcher{}, informers, time.Hour)
time.Sleep(100 * time.Millisecond)
if r.dataVolumeReader() != nil {
t.Fatal("the DataVolume reader was installed before the first reconcile")
}
if _, err := r.Reconcile(ctx, reconcile.Request{NamespacedName: types.NamespacedName{Name: "m", Namespace: "default"}}); err != nil {
t.Fatalf("Reconcile: %v", err)
}
err := wait.PollUntilContextTimeout(ctx, 10*time.Millisecond, 5*time.Second, true, func(context.Context) (bool, error) {
return r.dataVolumeReader() != nil, nil
})
if err != nil {
t.Fatal("the DataVolume reader was not installed after the first reconcile, within a fraction of the poll interval")
}
}
func TestTryStartDataVolumeWatch_UnsyncedSourceLeavesNoReaderAndRetries(t *testing.T) {
r := &WorkloadMonitorReconciler{}
w := &startingWatcher{}
informers := &informertest.FakeInformers{}
fi := fakeDataVolumeInformer(t, informers)
fi.Synced = false
done, err := r.tryStartDataVolumeWatch(context.Background(), servedMapper(), w, informers, 50*time.Millisecond)
if err == nil || done || r.dataVolumeReader() != nil {
t.Fatalf("informer never synced: done=%v err=%v reader=%v, want an error, not done, no reader", done, err, r.dataVolumeReader())
}
fi.SyncedLock.Lock()
fi.Synced = true
fi.SyncedLock.Unlock()
done, err = r.tryStartDataVolumeWatch(context.Background(), servedMapper(), w, informers, time.Second)
if err != nil || !done || r.dataVolumeReader() == nil {
t.Fatalf("retry after sync: done=%v err=%v reader=%v, want done and a reader", done, err, r.dataVolumeReader())
}
if len(w.sources) != 2 {
t.Fatalf("got %d watches, want a fresh source on the retry", len(w.sources))
}
}
// WaitForSync returns nil once the caller's own context is cancelled, which
// says nothing about the informer.
func TestTryStartDataVolumeWatch_StoppingManagerLeavesNoReader(t *testing.T) {
r := &WorkloadMonitorReconciler{}
informers := &informertest.FakeInformers{}
fakeDataVolumeInformer(t, informers).Synced = false
ctx, cancel := context.WithCancel(context.Background())
cancel()
done, err := r.tryStartDataVolumeWatch(ctx, servedMapper(), &startingWatcher{}, informers, time.Second)
if done || r.dataVolumeReader() != nil {
t.Fatalf("done=%v reader=%v, want no reader when the manager is stopping", done, r.dataVolumeReader())
}
if !errors.Is(err, context.Canceled) || strings.Contains(err.Error(), "%!") {
t.Errorf("err = %q, want context.Canceled and nothing formatted from a nil error", err)
}
}
func TestTryStartDataVolumeWatch_DiscoveryErrorRetries(t *testing.T) {
r := &WorkloadMonitorReconciler{}
w := &startingWatcher{}
done, err := r.tryStartDataVolumeWatch(context.Background(), failingRESTMapper{meta.NewDefaultRESTMapper(nil)}, w, &informertest.FakeInformers{}, time.Second)
if err == nil || done || len(w.sources) != 0 || r.dataVolumeReader() != nil {
t.Fatalf("done=%v err=%v watches=%d reader=%v, want an error and nothing registered", done, err, len(w.sources), r.dataVolumeReader())
}
}
func TestTryStartDataVolumeWatch_FailedWatchLeavesNoReader(t *testing.T) {
r := &WorkloadMonitorReconciler{}
w := &startingWatcher{err: fmt.Errorf("watch failed")}
done, err := r.tryStartDataVolumeWatch(context.Background(), servedMapper(), w, &informertest.FakeInformers{}, time.Second)
if err == nil || done || r.dataVolumeReader() != nil {
t.Fatalf("done=%v err=%v reader=%v, want an error, not done, no reader", done, err, r.dataVolumeReader())
}
}