admin/cozystack
1
0 Fork 0
mirror of https://github.com/cozystack/cozystack.git synced 2026-10-28 23:57:08 +09:00
Code Issues Packages Projects Releases Wiki Activity
Files
main
cozystack/internal/controller/workloadmonitor_controller_test.go
T
Andrey Kolkov a6c14f6834 fix(workloadmonitor): migrate ownerless legacy bucket Workloads and prefer legacy sizes on takeover 
A BucketClaim deleted with --cascade=orphan and recreated leaves its
bucket-bucket-<name> Workload without owners. The migration recognised a
legacy Workload only by the claim's UID among its owners, so that object
stayed next to the new one with a frozen size and the bucket was metered
twice. Treat an ownerless legacy Workload that carries the monitor's
label as the claim's: delete it with the same preconditions, but take no
sizes from it, since they may predate the claim.

The size carry-over kept whatever the Workload at the new name had
recorded. When that Workload was not yet owned by the claim, an ownerless
leftover, its sizes were not written for this bucket, yet they won over
the migrating legacy Workload's. Prefer the recorded sizes only when the
claim already owned the Workload.

Assisted-by: LLM
Signed-off-by: Andrey Kolkov <androndo@gmail.com>
2026-10-23 17:45:04 +04:00

2978 lines
108 KiB
Go
Raw Permalink Blame History

package controller
import (
"context"
"errors"
"fmt"
"net/http"
"net/http/httptest"
"os"
"slices"
"strings"
"testing"
"time"
cozyv1alpha1 "github.com/cozystack/cozystack/api/v1alpha1"
corev1 "k8s.io/api/core/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
"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"
toolscache "k8s.io/client-go/tools/cache"
"k8s.io/client-go/tools/record"
"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)
}
wantReason := ""
if tc.want != "" {
wantReason = cozyv1alpha1.WorkloadMonitorReasonDataVolumeNotReady
}
if got.Status.Reason != wantReason {
t.Errorf("Reason = %q, want %q", got.Status.Reason, wantReason)
}
})
}
}
func withRunningCondition(dv *unstructured.Unstructured, status, reason, message string) *unstructured.Unstructured {
_ = unstructured.SetNestedSlice(dv.Object, []any{map[string]any{
"type": "Running",
"status": status,
"reason": reason,
"message": message,
}}, "status", "conditions")
return dv
}
func TestReconcile_DataVolumeMessageCarriesTheImporterError(t *testing.T) {
selected := map[string]string{"app.kubernetes.io/instance": "vm-disk-test"}
phase := func(p string) *string { return &p }
// Running conditions CDI v1.64 wrote for importers that kept failing; only
// the host name is changed. v1.66 builds the condition the same way for a
// single-container importer.
const (
http404 = "Unable to connect to http data source: expected status code 200, got 404. Status: 404 Not Found"
noHost = `Unable to connect to http data source: HTTP request errored: Get "http://image.example.invalid/disk.img": dial tcp: lookup image.example.invalid on 10.96.0.10:53: no such host`
)
cases := []struct {
name string
phase *string
status, reason, message string
want string
}{
{"404 from the source", phase("ImportInProgress"), "False", "Error", http404,
"DataVolume vm-disk-test is ImportInProgress: " + http404},
{"unresolvable host", phase("ImportInProgress"), "False", "Error", noHost,
"DataVolume vm-disk-test is ImportInProgress: " + noHost},
{"unschedulable importer", phase("ImportScheduled"), "False", "Unschedulable", "Importer pod cannot be scheduled",
"DataVolume vm-disk-test is ImportScheduled: Importer pod cannot be scheduled"},
{"fatal checksum error", phase("Failed"), "False", "ChecksumError", "checksum mismatch",
"DataVolume vm-disk-test is Failed: checksum mismatch"},
{"no phase", nil, "False", "Error", http404,
"DataVolume vm-disk-test has no phase: " + http404},
{"importer not started yet", phase("ImportScheduled"), "False", "", "",
"DataVolume vm-disk-test is ImportScheduled"},
{"importer running", phase("ImportInProgress"), "True", "Pod is running", "transferring",
"DataVolume vm-disk-test is ImportInProgress"},
{"source state unknown", phase("CloneInProgress"), "Unknown", "Pending and Unknown", "Pod Pending and source state unknown",
"DataVolume vm-disk-test is CloneInProgress"},
{"clone pods both waiting without a message", phase("CloneInProgress"), "False", "ContainerCreating and ContainerCreating", " and ",
"DataVolume vm-disk-test is CloneInProgress"},
{"only the source pod has a message", phase("CloneInProgress"), "False", "ContainerCreating and Error", " and source unreachable",
"DataVolume vm-disk-test is CloneInProgress: source unreachable"},
{"both clone pods have a message", phase("CloneInProgress"), "False", "Pending and Pending", "Pod Pending and Pod Pending",
"DataVolume vm-disk-test is CloneInProgress: Pod Pending and Pod Pending"},
{"populated", phase("Succeeded"), "False", "Completed", "Import Complete", ""},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
dv := withRunningCondition(newDataVolume("vm-disk-test", selected, tc.phase), tc.status, tc.reason, tc.message)
got := reconcileDataVolumeMonitor(t, true, nil, dv)
if got.Status.Message != tc.want {
t.Errorf("Message = %q, want %q", got.Status.Message, tc.want)
}
})
}
t.Run("only the Running condition is read", func(t *testing.T) {
dv := withRunningCondition(newDataVolume("vm-disk-test", selected, phase("ImportInProgress")), "True", "Pod is running", "")
conditions, _, _ := unstructured.NestedSlice(dv.Object, "status", "conditions")
conditions = append([]any{map[string]any{
"type": "Bound", "status": "False", "reason": "Pending",
"message": "target PVC vm-disk-test Pending and [prime-1] : " + http404,
}}, conditions...)
_ = unstructured.SetNestedSlice(dv.Object, conditions, "status", "conditions")
got := reconcileDataVolumeMonitor(t, true, nil, dv)
if want := "DataVolume vm-disk-test is ImportInProgress"; got.Status.Message != want {
t.Errorf("Message = %q, want %q", got.Status.Message, want)
}
})
}
// The fixture is a DataVolume read off a CDI v1.64 cluster after its importer
// had restarted 1988 times on a 404; only names and the URL are changed. For a
// single-container importer failing like this one, the pinned v1.66 differs
// only in requiring every container to run before it reports Running=True.
func TestReconcile_LiveStuckImportNamesTheSourceError(t *testing.T) {
raw, err := os.ReadFile("testdata/datavolume-http-404.json")
if err != nil {
t.Fatal(err)
}
dv := &unstructured.Unstructured{}
if err := dv.UnmarshalJSON(raw); err != nil {
t.Fatal(err)
}
got := reconcileDataVolumeMonitor(t, true, nil, dv)
const want = "DataVolume vm-disk-test is ImportInProgress: Unable to connect to http data source: expected status code 200, got 404. Status: 404 Not Found"
if *got.Status.Operational || got.Status.Message != want {
t.Errorf("Operational=%v Message=%q, want false and %q", *got.Status.Operational, got.Status.Message, want)
}
}
func TestReconcile_PopulatedDataVolumeClearsTheLastVerdict(t *testing.T) {
s := newTestScheme()
selector := map[string]string{"app.kubernetes.io/instance": "vm-disk-test"}
monitor := &cozyv1alpha1.WorkloadMonitor{
ObjectMeta: metav1.ObjectMeta{Name: "vm-disk-test", Namespace: "default"},
Spec: cozyv1alpha1.WorkloadMonitorSpec{Selector: selector, MinReplicas: ptr.To[int32](0)},
Status: cozyv1alpha1.WorkloadMonitorStatus{
Operational: ptr.To(false),
Message: "DataVolume vm-disk-test is ImportInProgress",
Reason: cozyv1alpha1.WorkloadMonitorReasonDataVolumeNotReady,
},
}
done := "Succeeded"
fakeClient := fake.NewClientBuilder().
WithScheme(s).
WithObjects(monitor, newDataVolume("vm-disk-test", selector, &done)).
WithStatusSubresource(monitor).
Build()
reconciler := &WorkloadMonitorReconciler{Client: fakeClient, Scheme: s, 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)
}
got := &cozyv1alpha1.WorkloadMonitor{}
if err := fakeClient.Get(context.TODO(), req.NamespacedName, got); err != nil {
t.Fatal(err)
}
if got.Status.Operational == nil || !*got.Status.Operational || got.Status.Message != "" || got.Status.Reason != "" {
t.Errorf("Operational=%v Message=%q Reason=%q, want true and both cleared", got.Status.Operational, got.Status.Message, got.Status.Reason)
}
}
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;
// syncing marks the DataVolume watch as started and not yet synced.
func reconcileWithUnreadableDataVolumes(t *testing.T, withReader, syncing bool, status cozyv1alpha1.WorkloadMonitorStatus) (*cozyv1alpha1.WorkloadMonitor, reconcile.Result, error) {
t.Helper()
got, result, _, err := reconcileWithUnreadableDataVolumesEvents(t, withReader, syncing, status)
return got, result, err
}
func reconcileWithUnreadableDataVolumesEvents(t *testing.T, withReader, syncing bool, status cozyv1alpha1.WorkloadMonitorStatus) (*cozyv1alpha1.WorkloadMonitor, reconcile.Result, []string, 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()
recorder := record.NewFakeRecorder(10)
reconciler := &WorkloadMonitorReconciler{Client: fakeClient, Scheme: s, Recorder: recorder}
if withReader {
reconciler.DataVolumeReader = fakeClient
}
reconciler.dataVolumeWatchSyncing = syncing
req := reconcile.Request{NamespacedName: types.NamespacedName{Name: "m", Namespace: "default"}}
result, 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)
}
close(recorder.Events)
var events []string
for e := range recorder.Events {
events = append(events, e)
}
return updated, result, events, err
}
func TestReconcile_DataVolumeListErrorStillPublishesStatus(t *testing.T) {
got, _, events, err := reconcileWithUnreadableDataVolumesEvents(t, true, false, cozyv1alpha1.WorkloadMonitorStatus{})
if err == nil {
t.Error("Reconcile returned nil on a DataVolume list failure, want the error so the request is retried")
}
if len(events) != 1 || !strings.HasPrefix(events[0], "Warning DataVolumesUnavailable ") || !strings.Contains(events[0], "apiserver unavailable") {
t.Errorf("events = %q, want one Warning DataVolumesUnavailable event carrying the List error", events)
}
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, false, cozyv1alpha1.WorkloadMonitorStatus{
Operational: ptr.To(false),
Message: stuck,
Reason: cozyv1alpha1.WorkloadMonitorReasonDataVolumeNotReady,
})
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 || got.Status.Reason != cozyv1alpha1.WorkloadMonitorReasonDataVolumeNotReady {
t.Errorf("Operational=%v Message=%q, want false and %q", got.Status.Operational, got.Status.Message, stuck)
}
})
}
}
func TestReconcile_SyncingDataVolumeWatchPublishesAndRequeues(t *testing.T) {
const stuck = "DataVolume m is Failed"
got, result, events, err := reconcileWithUnreadableDataVolumesEvents(t, false, true, cozyv1alpha1.WorkloadMonitorStatus{
Operational: ptr.To(false),
Message: stuck,
Reason: cozyv1alpha1.WorkloadMonitorReasonDataVolumeNotReady,
})
if err != nil || result.RequeueAfter <= 0 || len(events) != 0 {
t.Errorf("Reconcile = %+v, %v, events %q while the DataVolume watch syncs, want a requeue, no error and no event: the window is expected, not a failure", result, err, events)
}
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
// onWatch runs when Watch is called, before the source starts.
onWatch func()
queue workqueue.TypedRateLimitingInterface[reconcile.Request]
}
func (w *startingWatcher) Watch(src source.Source) error {
if w.onWatch != nil {
w.onWatch()
}
if w.err != nil {
return w.err
}
w.sources = append(w.sources, src)
if w.queue == nil {
w.queue = workqueue.NewTypedRateLimitingQueue(workqueue.DefaultTypedControllerRateLimiter[reconcile.Request]())
}
return src.Start(context.Background(), w.queue)
}
func readerOf(r *WorkloadMonitorReconciler) client.Reader {
r.dataVolumeMu.RLock()
defer r.dataVolumeMu.RUnlock()
return r.DataVolumeReader
}
func watchReconciler() *WorkloadMonitorReconciler {
s := newTestScheme()
return &WorkloadMonitorReconciler{Client: fake.NewClientBuilder().WithScheme(s).Build(), Scheme: s}
}
// kindSources leaves out the source that queues the monitors once the reader
// is installed.
func (w *startingWatcher) kindSources() int {
n := 0
for _, src := range w.sources {
if _, ok := src.(source.Func); !ok {
n++
}
}
return n
}
func servedMapper() meta.RESTMapper {
mapper := meta.NewDefaultRESTMapper(nil)
mapper.Add(dataVolumeGVK, meta.RESTScopeNamespace)
return mapper
}
// fakeDataVolumeInformer also fills the lazily built fields of FakeInformers,
// which is not safe for concurrent use while the source reads them.
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
}
// dataVolumeInformers seeds the DataVolume informer, because FakeInformers
// creates every informer it builds itself already synced.
func dataVolumeInformers(synced bool) (*informertest.FakeInformers, *controllertest.FakeInformer) {
fi := controllertest.NewFakeInformer()
if synced {
fi.Synced()
}
return &informertest.FakeInformers{
InformersByGVK: map[schema.GroupVersionKind]toolscache.SharedIndexInformer{dataVolumeGVK: fi},
}, fi
}
func TestTryStartDataVolumeWatch_WaitsUntilTheKindIsServed(t *testing.T) {
r := watchReconciler()
w := &startingWatcher{}
mapper := meta.NewDefaultRESTMapper(nil)
informers := &informertest.FakeInformers{}
fakeDataVolumeInformer(t, informers)
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 w.kindSources() != 0 || readerOf(r) != nil {
t.Fatalf("before CDI: %d watches, reader %v; want none", len(w.sources), readerOf(r))
}
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 w.kindSources() != 1 || readerOf(r) == nil {
t.Fatalf("after CDI: %d watches, reader %v; want one watch and a reader", len(w.sources), readerOf(r))
}
}
// Before the controller has started its sources, Watch only queues a source,
// 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 readerOf(r) != 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 readerOf(r) != 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 := watchReconciler()
w := &startingWatcher{}
informers, fi := dataVolumeInformers(false)
for attempt := 0; attempt < 3; attempt++ {
done, err := r.tryStartDataVolumeWatch(context.Background(), servedMapper(), w, informers, 50*time.Millisecond)
if err == nil || done || readerOf(r) != nil {
t.Fatalf("informer never synced: done=%v err=%v reader=%v, want an error, not done, no reader", done, err, readerOf(r))
}
}
fi.Synced()
done, err := r.tryStartDataVolumeWatch(context.Background(), servedMapper(), w, informers, time.Second)
if err != nil || !done || readerOf(r) == nil {
t.Fatalf("retry after sync: done=%v err=%v reader=%v, want done and a reader", done, err, readerOf(r))
}
// Every started source adds an event handler to the informer for good, so
// a retry must wait on the source it already has.
if w.kindSources() != 1 {
t.Fatalf("got %d watches over four attempts, want the first source reused", len(w.sources))
}
}
// A stopping manager ends the sync wait, which says nothing about the informer.
func TestTryStartDataVolumeWatch_StoppingManagerLeavesNoReader(t *testing.T) {
r := watchReconciler()
informers, _ := dataVolumeInformers(false)
ctx, cancel := context.WithCancel(context.Background())
cancel()
done, err := r.tryStartDataVolumeWatch(ctx, servedMapper(), &startingWatcher{}, informers, time.Second)
if done || readerOf(r) != nil {
t.Fatalf("done=%v reader=%v, want no reader when the manager is stopping", done, readerOf(r))
}
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)
}
}
// The source replays every DataVolume to the handler as soon as Watch starts
// it, before the reader is installed. A reconcile in that window must be
// retried, or the events that queued it are lost.
func TestTryStartDataVolumeWatch_ReconcileBeforeTheReaderIsRetried(t *testing.T) {
for _, synced := range []bool{true, false} {
t.Run(fmt.Sprintf("synced=%v", synced), func(t *testing.T) {
r := watchReconciler()
informers, _ := dataVolumeInformers(synced)
monitor := &cozyv1alpha1.WorkloadMonitor{ObjectMeta: metav1.ObjectMeta{Name: "m", Namespace: "default"}}
var errDuringWatch error
w := &startingWatcher{}
w.onWatch = func() {
if len(w.sources) == 0 {
_, _, _, errDuringWatch = r.dataVolumesMessage(context.Background(), monitor)
}
}
_, _ = r.tryStartDataVolumeWatch(context.Background(), servedMapper(), w, informers, 50*time.Millisecond)
if errDuringWatch == nil {
t.Error("reading DataVolumes while the watch syncs returned no error, want one so the reconcile is retried")
}
// The started source keeps delivering events after a sync timeout,
// so a reconcile between attempts has to be retried as well.
if _, _, _, err := r.dataVolumesMessage(context.Background(), monitor); !synced && !errors.Is(err, errDataVolumeWatchSyncing) {
t.Errorf("after a sync timeout: %v, want errDataVolumeWatchSyncing until the reader is installed", err)
}
})
}
}
// A reconcile before the watch starts sees no reader and keeps the stored
// verdict, and the source's replay cannot deliver a DataVolume deleted while
// the controller was down, so installing the reader must queue the monitors
// that hold a verdict.
func TestTryStartDataVolumeWatch_StoredVerdictIsRecomputedOnceTheReaderIsInstalled(t *testing.T) {
s := newTestScheme()
stuck := &cozyv1alpha1.WorkloadMonitor{
ObjectMeta: metav1.ObjectMeta{Name: "stuck", Namespace: "default"},
Spec: cozyv1alpha1.WorkloadMonitorSpec{
Selector: map[string]string{"app.kubernetes.io/instance": "stuck"},
MinReplicas: ptr.To[int32](0),
},
Status: cozyv1alpha1.WorkloadMonitorStatus{
Operational: ptr.To(false),
Message: "DataVolume stuck is ImportInProgress",
Reason: cozyv1alpha1.WorkloadMonitorReasonDataVolumeNotReady,
},
}
noVerdict := &cozyv1alpha1.WorkloadMonitor{
ObjectMeta: metav1.ObjectMeta{Name: "no-verdict", Namespace: "default"},
Spec: cozyv1alpha1.WorkloadMonitorSpec{MinReplicas: ptr.To[int32](0)},
Status: cozyv1alpha1.WorkloadMonitorStatus{Operational: ptr.To(true)},
}
fakeClient := fake.NewClientBuilder().WithScheme(s).
WithObjects(stuck, noVerdict).WithStatusSubresource(stuck, noVerdict).Build()
r := &WorkloadMonitorReconciler{Client: fakeClient, Scheme: s}
req := reconcile.Request{NamespacedName: types.NamespacedName{Name: "stuck", Namespace: "default"}}
if _, err := r.Reconcile(context.Background(), req); err != nil {
t.Fatalf("Reconcile before the watch: %v", err)
}
informers := &informertest.FakeInformers{}
fakeDataVolumeInformer(t, informers)
w := &startingWatcher{}
w.onWatch = func() {
if len(w.sources) == 1 && readerOf(r) == nil {
t.Error("monitors queued before the reader is installed would reconcile without it")
}
}
done, err := r.tryStartDataVolumeWatch(context.Background(), servedMapper(), w, informers, time.Second)
if err != nil || !done {
t.Fatalf("done=%v err=%v, want the reader installed", done, err)
}
var queued []reconcile.Request
for w.queue.Len() > 0 {
item, _ := w.queue.Get()
queued = append(queued, item)
w.queue.Done(item)
}
if len(queued) != 1 || queued[0] != req {
t.Fatalf("queued %v, want only the monitor holding a DataVolume verdict", queued)
}
if _, err := r.Reconcile(context.Background(), queued[0]); err != nil {
t.Fatalf("Reconcile after the reader: %v", err)
}
got := &cozyv1alpha1.WorkloadMonitor{}
if err := fakeClient.Get(context.Background(), req.NamespacedName, got); err != nil {
t.Fatal(err)
}
if got.Status.Operational == nil || !*got.Status.Operational || got.Status.Message != "" || got.Status.Reason != "" {
t.Errorf("Operational=%v Message=%q Reason=%q, want the verdict of the deleted DataVolume cleared", got.Status.Operational, got.Status.Message, got.Status.Reason)
}
}
func TestTryStartDataVolumeWatch_FailedMonitorListIsRetried(t *testing.T) {
s := newTestScheme()
monitor := &cozyv1alpha1.WorkloadMonitor{
ObjectMeta: metav1.ObjectMeta{Name: "m", Namespace: "default"},
Status: cozyv1alpha1.WorkloadMonitorStatus{Message: "DataVolume m is Failed"},
}
listFails := true
fakeClient := fake.NewClientBuilder().WithScheme(s).WithObjects(monitor).WithInterceptorFuncs(interceptor.Funcs{
List: func(ctx context.Context, c client.WithWatch, list client.ObjectList, opts ...client.ListOption) error {
if listFails {
return fmt.Errorf("apiserver unavailable")
}
return c.List(ctx, list, opts...)
},
}).Build()
r := &WorkloadMonitorReconciler{Client: fakeClient, Scheme: s}
informers := &informertest.FakeInformers{}
fakeDataVolumeInformer(t, informers)
w := &startingWatcher{}
done, err := r.tryStartDataVolumeWatch(context.Background(), servedMapper(), w, informers, time.Second)
if err == nil || done {
t.Fatalf("done=%v err=%v, want an error and a retry: a monitor left unqueued keeps a stale verdict", done, err)
}
listFails = false
done, err = r.tryStartDataVolumeWatch(context.Background(), servedMapper(), w, informers, time.Second)
if err != nil || !done || w.kindSources() != 1 || w.queue.Len() != 1 {
t.Fatalf("retry: done=%v err=%v kind sources=%d queued=%d, want done, the source reused and the monitor queued", done, err, w.kindSources(), w.queue.Len())
}
}
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 || readerOf(r) != nil {
t.Fatalf("done=%v err=%v watches=%d reader=%v, want an error and nothing registered", done, err, len(w.sources), readerOf(r))
}
}
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 || readerOf(r) != nil {
t.Fatalf("done=%v err=%v reader=%v, want an error, not done, no reader", done, err, readerOf(r))
}
}
// While the DataVolume watch syncs, every monitor is requeued every few
// seconds, so a pass that changes nothing must not write status.
func TestReconcile_UnchangedStatusIsNotWritten(t *testing.T) {
settled := cozyv1alpha1.WorkloadMonitorStatus{Operational: ptr.To(true), AvailableReplicas: 1, ObservedReplicas: 1}
for _, tc := range []struct {
name string
status cozyv1alpha1.WorkloadMonitorStatus
writes int
}{
{"unchanged", settled, 0},
{"changed", cozyv1alpha1.WorkloadMonitorStatus{Operational: ptr.To(true)}, 1},
} {
t.Run(tc.name, func(t *testing.T) {
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: tc.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},
}},
}
writes := 0
fakeClient := fake.NewClientBuilder().
WithScheme(s).
WithObjects(monitor, pod).
WithStatusSubresource(monitor).
WithInterceptorFuncs(interceptor.Funcs{
SubResourceUpdate: func(ctx context.Context, c client.Client, subResourceName string, obj client.Object, opts ...client.SubResourceUpdateOption) error {
if _, ok := obj.(*cozyv1alpha1.WorkloadMonitor); ok {
writes++
}
return c.SubResource(subResourceName).Update(ctx, obj, opts...)
},
}).
Build()
r := &WorkloadMonitorReconciler{Client: fakeClient, Scheme: s, dataVolumeWatchSyncing: true}
result, err := r.Reconcile(context.TODO(), reconcile.Request{NamespacedName: types.NamespacedName{Name: "m", Namespace: "default"}})
if err != nil || result.RequeueAfter <= 0 {
t.Fatalf("Reconcile = %+v, %v, want a requeue while the watch syncs", result, err)
}
if writes != tc.writes {
t.Errorf("got %d status writes, want %d", writes, tc.writes)
}
})
}
}
// CDI copies the DataVolume labels onto the PVC it creates and makes the
// DataVolume its controller, so the monitor also publishes a Workload for the
// PVC. Bound alone would read it operational mid-import.
func TestReconcile_PVCOfANotReadyDataVolumeIsNotOperational(t *testing.T) {
selector := map[string]string{"app.kubernetes.io/instance": "vm-disk-test"}
for _, tc := range []struct {
name string
phase string
reader bool
stored *bool
controlled bool
// ownerGVK replaces the DataVolume GVK in the PVC's controller reference.
ownerGVK *schema.GroupVersionKind
want bool
}{
{name: "importing", phase: "ImportInProgress", reader: true, controlled: true, want: false},
{name: "populated", phase: "Succeeded", reader: true, controlled: true, want: true},
{name: "not owned by the DataVolume", phase: "ImportInProgress", reader: true, want: true},
{name: "controller of another kind", phase: "ImportInProgress", reader: true, controlled: true, ownerGVK: &schema.GroupVersionKind{Group: dataVolumeGVK.Group, Version: dataVolumeGVK.Version, Kind: "DataSource"}, want: true},
{name: "DataVolume kind of another group", phase: "ImportInProgress", reader: true, controlled: true, ownerGVK: &schema.GroupVersionKind{Group: "example.com", Version: "v1", Kind: dataVolumeGVK.Kind}, want: true},
{name: "unread keeps the stored verdict", phase: "Succeeded", stored: ptr.To(false), controlled: true, want: false},
{name: "unread and new reads bind state", phase: "ImportInProgress", controlled: true, want: true},
} {
t.Run(tc.name, func(t *testing.T) {
s := newTestScheme()
monitor := &cozyv1alpha1.WorkloadMonitor{
ObjectMeta: metav1.ObjectMeta{Name: "vm-disk-test", Namespace: "default"},
Spec: cozyv1alpha1.WorkloadMonitorSpec{Selector: selector, MinReplicas: ptr.To[int32](0)},
}
dv := newDataVolume("vm-disk-test", selector, &tc.phase)
dv.SetUID("dv-uid")
pvc := &corev1.PersistentVolumeClaim{
ObjectMeta: metav1.ObjectMeta{Name: "vm-disk-test", Namespace: "default", Labels: selector},
Spec: corev1.PersistentVolumeClaimSpec{StorageClassName: ptr.To("replicated")},
Status: corev1.PersistentVolumeClaimStatus{Phase: corev1.ClaimBound},
}
if tc.controlled {
gvk := dataVolumeGVK
if tc.ownerGVK != nil {
gvk = *tc.ownerGVK
}
pvc.OwnerReferences = []metav1.OwnerReference{*metav1.NewControllerRef(dv, gvk)}
}
objs := []client.Object{monitor, dv, pvc}
if tc.stored != nil {
objs = append(objs, &cozyv1alpha1.Workload{
ObjectMeta: metav1.ObjectMeta{Name: "pvc-vm-disk-test", Namespace: "default"},
Status: cozyv1alpha1.WorkloadStatus{Operational: *tc.stored},
})
}
fakeClient := fake.NewClientBuilder().WithScheme(s).WithObjects(objs...).WithStatusSubresource(monitor).Build()
r := &WorkloadMonitorReconciler{Client: fakeClient, Scheme: s}
if tc.reader {
r.DataVolumeReader = fakeClient
}
if _, err := r.Reconcile(context.TODO(), reconcile.Request{NamespacedName: types.NamespacedName{Name: "vm-disk-test", Namespace: "default"}}); err != nil {
t.Fatalf("Reconcile: %v", err)
}
workload := &cozyv1alpha1.Workload{}
if err := fakeClient.Get(context.TODO(), types.NamespacedName{Name: "pvc-vm-disk-test", Namespace: "default"}, workload); err != nil {
t.Fatalf("Failed to get the PVC Workload: %v", err)
}
if workload.Status.Operational != tc.want {
t.Errorf("PVC Workload Operational = %v, want %v", workload.Status.Operational, tc.want)
}
})
}
}
func TestBucketWorkloadName(t *testing.T) {
for claim, want := range map[string]string{
// a Bucket application's claim is named after its bucket- release
"bucket-cozy-backups": "bucket-cozy-backups",
"my-bucket": "bucket-my-bucket",
"harbor-x-registry": "bucket-harbor-x-registry",
} {
if got := bucketWorkloadName(claim); got != want {
t.Errorf("bucketWorkloadName(%q) = %q, want %q", claim, got, want)
}
}
}
// reconcileBucketApp reconciles a Bucket application named cozy-backups
// alongside the given existing objects and returns the client.
func reconcileBucketApp(t *testing.T, existing ...client.Object) client.Client {
t.Helper()
s := newTestScheme()
monitor := &cozyv1alpha1.WorkloadMonitor{
ObjectMeta: metav1.ObjectMeta{Name: "bucket-cozy-backups", Namespace: "tenant-demo"},
Spec: cozyv1alpha1.WorkloadMonitorSpec{
Kind: "bucket",
Type: "s3",
Selector: map[string]string{"app.kubernetes.io/instance": "bucket-cozy-backups"},
},
}
bc := &cosiv1alpha1.BucketClaim{
ObjectMeta: metav1.ObjectMeta{
Name: "bucket-cozy-backups",
Namespace: "tenant-demo",
UID: "claim-uid",
Labels: map[string]string{"app.kubernetes.io/instance": "bucket-cozy-backups"},
},
Spec: cosiv1alpha1.BucketClaimSpec{BucketClassName: "seaweedfs", Protocols: []cosiv1alpha1.Protocol{cosiv1alpha1.ProtocolS3}},
Status: cosiv1alpha1.BucketClaimStatus{BucketReady: true, BucketName: "cosi-abc123"},
}
c := fake.NewClientBuilder().
WithScheme(s).
WithObjects(append([]client.Object{monitor, bc}, existing...)...).
WithStatusSubresource(monitor).
Build()
reconciler := &WorkloadMonitorReconciler{Client: c, Scheme: s}
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)
}
return c
}
func legacyBucketWorkloadOwnedBy(uid types.UID) *cozyv1alpha1.Workload {
return &cozyv1alpha1.Workload{
ObjectMeta: metav1.ObjectMeta{
Name: "bucket-bucket-cozy-backups",
Namespace: "tenant-demo",
OwnerReferences: []metav1.OwnerReference{{
APIVersion: "objectstorage.k8s.io/v1alpha1",
Kind: "BucketClaim",
Name: "bucket-cozy-backups",
UID: uid,
}},
},
Status: cozyv1alpha1.WorkloadStatus{
Kind: "bucket",
Type: "s3",
Resources: map[string]resource.Quantity{
resourceS3StorageBytes: resource.MustParse("689Gi"),
resourceS3PhysicalStorageBytes: resource.MustParse("1378Gi"),
},
},
}
}
func TestReconcileBucketApp_NoDoublePrefix(t *testing.T) {
c := reconcileBucketApp(t)
if err := c.Get(context.TODO(), types.NamespacedName{Name: "bucket-cozy-backups", Namespace: "tenant-demo"}, &cozyv1alpha1.Workload{}); err != nil {
t.Fatalf("expected Workload bucket-cozy-backups, got error: %v", err)
}
err := c.Get(context.TODO(), types.NamespacedName{Name: "bucket-bucket-cozy-backups", Namespace: "tenant-demo"}, &cozyv1alpha1.Workload{})
if !apierrors.IsNotFound(err) {
t.Errorf("expected no Workload bucket-bucket-cozy-backups, got error %v", err)
}
}
func TestReconcileBucketApp_ReplacesLegacyWorkload(t *testing.T) {
c := reconcileBucketApp(t, legacyBucketWorkloadOwnedBy("claim-uid"))
err := c.Get(context.TODO(), types.NamespacedName{Name: "bucket-bucket-cozy-backups", Namespace: "tenant-demo"}, &cozyv1alpha1.Workload{})
if !apierrors.IsNotFound(err) {
t.Errorf("expected the legacy Workload to be deleted, got error %v", err)
}
workload := &cozyv1alpha1.Workload{}
if err := c.Get(context.TODO(), types.NamespacedName{Name: "bucket-cozy-backups", Namespace: "tenant-demo"}, workload); err != nil {
t.Fatalf("expected Workload bucket-cozy-backups, got error: %v", err)
}
// no metrics endpoint in this test, so the sizes can only come from the legacy Workload
for key, size := range map[string]string{resourceS3StorageBytes: "689Gi", resourceS3PhysicalStorageBytes: "1378Gi"} {
got := workload.Status.Resources[key]
if want := resource.MustParse(size); got.Cmp(want) != 0 {
t.Errorf("%s = %s, want %s carried over from the legacy Workload", key, got.String(), want.String())
}
}
}
func TestReconcileBucketApp_KeepsLegacyNameOwnedByAnotherClaim(t *testing.T) {
c := reconcileBucketApp(t, legacyBucketWorkloadOwnedBy("other-claim-uid"))
foreign := &cozyv1alpha1.Workload{}
if err := c.Get(context.TODO(), types.NamespacedName{Name: "bucket-bucket-cozy-backups", Namespace: "tenant-demo"}, foreign); err != nil {
t.Fatalf("expected a Workload owned by another claim to stay, got error: %v", err)
}
workload := &cozyv1alpha1.Workload{}
if err := c.Get(context.TODO(), types.NamespacedName{Name: "bucket-cozy-backups", Namespace: "tenant-demo"}, workload); err != nil {
t.Fatalf("expected Workload bucket-cozy-backups, got error: %v", err)
}
if _, ok := workload.Status.Resources[resourceS3StorageBytes]; ok {
t.Errorf("expected no size taken from a Workload owned by another claim, got %v", workload.Status.Resources)
}
}
// Bucket applications "x" and "bucket-x" in one namespace: the legacy name
// of the first (bucket-bucket-x) is the new name of the second.
func TestReconcileBucketApps_LegacyNameIsAnotherAppsNewName(t *testing.T) {
for _, order := range [][]string{
{"bucket-bucket-x", "bucket-x", "bucket-bucket-x"},
{"bucket-x", "bucket-bucket-x", "bucket-x"},
} {
t.Run(order[0]+" first", func(t *testing.T) {
s := newTestScheme()
monX, claimX := bucketMonitorAndClaim("bucket-x", "uid-x")
monBX, claimBX := bucketMonitorAndClaim("bucket-bucket-x", "uid-bx")
c := fake.NewClientBuilder().WithScheme(s).
WithObjects(monX, claimX, monBX, claimBX,
bucketWorkloadWithSize("bucket-bucket-x", "bucket-x", "uid-x", "689Gi"),
bucketWorkloadWithSize("bucket-bucket-bucket-x", "bucket-bucket-x", "uid-bx", "100Gi")).
WithStatusSubresource(monX, monBX).
Build()
reconciler := &WorkloadMonitorReconciler{Client: c, Scheme: s}
for _, monitor := range order {
req := reconcile.Request{NamespacedName: types.NamespacedName{Name: monitor, Namespace: "tenant-demo"}}
if _, err := reconciler.Reconcile(context.TODO(), req); err != nil {
t.Fatalf("Reconcile(%s) returned error: %v", monitor, err)
}
}
for name, want := range map[string]struct {
owner types.UID
size string
}{
"bucket-x": {"uid-x", "689Gi"},
"bucket-bucket-x": {"uid-bx", "100Gi"},
} {
w := &cozyv1alpha1.Workload{}
if err := c.Get(context.TODO(), types.NamespacedName{Name: name, Namespace: "tenant-demo"}, w); err != nil {
t.Errorf("Workload %s: %v", name, err)
continue
}
if len(w.OwnerReferences) != 1 || w.OwnerReferences[0].UID != want.owner {
t.Errorf("Workload %s owners = %v, want only %s", name, w.OwnerReferences, want.owner)
}
got := w.Status.Resources[resourceS3StorageBytes]
if got.Cmp(resource.MustParse(want.size)) != 0 {
t.Errorf("Workload %s size = %s, want %s", name, got.String(), want.size)
}
}
err := c.Get(context.TODO(), types.NamespacedName{Name: "bucket-bucket-bucket-x", Namespace: "tenant-demo"}, &cozyv1alpha1.Workload{})
if !apierrors.IsNotFound(err) {
t.Errorf("expected legacy bucket-bucket-bucket-x to be deleted, got error %v", err)
}
})
}
}
// bucketMonitorAndClaim returns a bucket WorkloadMonitor and the claim it
// selects, both named name.
func bucketMonitorAndClaim(name string, uid types.UID) (*cozyv1alpha1.WorkloadMonitor, *cosiv1alpha1.BucketClaim) {
labels := map[string]string{"app.kubernetes.io/instance": name}
return &cozyv1alpha1.WorkloadMonitor{
ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "tenant-demo"},
Spec: cozyv1alpha1.WorkloadMonitorSpec{Kind: "bucket", Type: "s3", Selector: labels},
}, &cosiv1alpha1.BucketClaim{
ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "tenant-demo", UID: uid, Labels: labels},
Spec: cosiv1alpha1.BucketClaimSpec{BucketClassName: "seaweedfs", Protocols: []cosiv1alpha1.Protocol{cosiv1alpha1.ProtocolS3}},
Status: cosiv1alpha1.BucketClaimStatus{BucketReady: true, BucketName: "cosi-" + string(uid)},
}
}
func bucketWorkloadWithSize(name, owner string, uid types.UID, size string) *cozyv1alpha1.Workload {
w := legacyBucketWorkloadOwnedBy(uid)
w.Name = name
w.OwnerReferences[0].Name = owner
w.Status.Resources = map[string]resource.Quantity{resourceS3StorageBytes: resource.MustParse(size)}
return w
}
func workloadSize(t *testing.T, c client.Client, name string) (resource.Quantity, *cozyv1alpha1.Workload) {
t.Helper()
w := &cozyv1alpha1.Workload{}
if err := c.Get(context.TODO(), types.NamespacedName{Name: name, Namespace: "tenant-demo"}, w); err != nil {
t.Fatalf("Workload %s: %v", name, err)
}
return w.Status.Resources[resourceS3StorageBytes], w
}
// For a claim without the prefix the legacy and the new name coincide, so
// its own Workload must not be mistaken for a legacy one.
func TestReconcileBucketClaim_UnprefixedClaimKeepsItsWorkload(t *testing.T) {
s := newTestScheme()
monitor, claim := bucketMonitorAndClaim("my-bucket", "uid-mine")
c := fake.NewClientBuilder().WithScheme(s).
WithObjects(monitor, claim, bucketWorkloadWithSize("bucket-my-bucket", "my-bucket", "uid-mine", "50Gi")).
WithStatusSubresource(monitor).Build()
reconciler := &WorkloadMonitorReconciler{Client: c, Scheme: s}
req := reconcile.Request{NamespacedName: types.NamespacedName{Name: "my-bucket", Namespace: "tenant-demo"}}
for range 2 {
if _, err := reconciler.Reconcile(context.TODO(), req); err != nil {
t.Fatalf("Reconcile returned error: %v", err)
}
}
if got, _ := workloadSize(t, c, "bucket-my-bucket"); got.Cmp(resource.MustParse("50Gi")) != 0 {
t.Errorf("size = %s, want 50Gi", got.String())
}
}
func TestReconcileBucketApp_RecordedSizeWinsOverLegacy(t *testing.T) {
c := reconcileBucketApp(t,
legacyBucketWorkloadOwnedBy("claim-uid"),
bucketWorkloadWithSize("bucket-cozy-backups", "bucket-cozy-backups", "claim-uid", "700Gi"))
if got, _ := workloadSize(t, c, "bucket-cozy-backups"); got.Cmp(resource.MustParse("700Gi")) != 0 {
t.Errorf("size = %s, want the recorded 700Gi rather than the legacy 689Gi", got.String())
}
}
// Claims "x" and "bucket-x" both map to Workload "bucket-x"; the second one
// must not take over the first one's Workload.
func TestReconcileBucketClaim_DoesNotAdoptAnotherClaimsWorkload(t *testing.T) {
s := newTestScheme()
monitor, claim := bucketMonitorAndClaim("bucket-x", "uid-bx")
c := fake.NewClientBuilder().WithScheme(s).
WithObjects(monitor, claim, bucketWorkloadWithSize("bucket-x", "x", "uid-x", "10Gi")).
WithStatusSubresource(monitor).Build()
recorder := record.NewFakeRecorder(10)
reconciler := &WorkloadMonitorReconciler{Client: c, Scheme: s, Recorder: recorder}
req := reconcile.Request{NamespacedName: types.NamespacedName{Name: "bucket-x", Namespace: "tenant-demo"}}
if _, err := reconciler.Reconcile(context.TODO(), req); err != nil {
t.Fatalf("Reconcile returned error: %v", err)
}
got, w := workloadSize(t, c, "bucket-x")
if len(w.OwnerReferences) != 1 || w.OwnerReferences[0].UID != "uid-x" {
t.Errorf("owners = %v, want only uid-x", w.OwnerReferences)
}
if got.Cmp(resource.MustParse("10Gi")) != 0 {
t.Errorf("size = %s, want 10Gi untouched", got.String())
}
select {
case e := <-recorder.Events:
if !strings.Contains(e, "BucketWorkloadNameTaken") {
t.Errorf("event = %q, want BucketWorkloadNameTaken", e)
}
default:
t.Error("expected a BucketWorkloadNameTaken event")
}
// the bucket is still metered, under its legacy name
if _, own := workloadSize(t, c, "bucket-bucket-x"); len(own.OwnerReferences) != 1 || own.OwnerReferences[0].UID != "uid-bx" {
t.Errorf("bucket-bucket-x owners = %v, want only uid-bx", own.OwnerReferences)
}
}
// While the new name is held by another owner, the bucket's legacy Workload
// keeps being updated rather than freezing at its last size.
func TestReconcileBucketClaim_KeepsMeteringUnderLegacyNameWhileNewNameIsHeld(t *testing.T) {
s := newTestScheme()
monitor, claim := bucketMonitorAndClaim("bucket-x", "uid-bx")
legacy := bucketWorkloadWithSize("bucket-bucket-x", "bucket-x", "uid-bx", "689Gi")
c := fake.NewClientBuilder().WithScheme(s).
WithObjects(monitor, claim, legacy, bucketWorkloadWithSize("bucket-x", "x", "uid-x", "10Gi")).
WithStatusSubresource(monitor).Build()
reconciler := &WorkloadMonitorReconciler{Client: c, Scheme: s}
// the second pass has nothing left to change, so nothing masks a delete
for range 2 {
if err := reconciler.reconcileBucketClaimForMonitor(context.TODO(), monitor, *claim, nil); err != nil {
t.Fatalf("reconcileBucketClaimForMonitor returned error: %v", err)
}
}
got, w := workloadSize(t, c, "bucket-bucket-x")
if w.Labels[workloadMonitorLabel] != "bucket-x" {
t.Errorf("legacy Workload not updated: labels = %v", w.Labels)
}
if got.Cmp(resource.MustParse("689Gi")) != 0 || len(w.OwnerReferences) != 1 || w.OwnerReferences[0].UID != "uid-bx" {
t.Errorf("legacy Workload changed: owners=%v size=%s", w.OwnerReferences, got.String())
}
if foreign, fw := workloadSize(t, c, "bucket-x"); foreign.Cmp(resource.MustParse("10Gi")) != 0 || fw.OwnerReferences[0].UID != "uid-x" {
t.Errorf("held Workload bucket-x changed: owners=%v size=%s", fw.OwnerReferences, foreign.String())
}
}
// A claim without the prefix has no other name to fall back to.
func TestReconcileBucketClaim_UnprefixedClaimWithHeldNameIsSkipped(t *testing.T) {
s := newTestScheme()
monitor, claim := bucketMonitorAndClaim("my-bucket", "uid-mine")
c := fake.NewClientBuilder().WithScheme(s).
WithObjects(monitor, claim, bucketWorkloadWithSize("bucket-my-bucket", "bucket-my-bucket", "uid-other", "3Gi")).
WithStatusSubresource(monitor).Build()
recorder := record.NewFakeRecorder(10)
reconciler := &WorkloadMonitorReconciler{Client: c, Scheme: s, Recorder: recorder}
if err := reconciler.reconcileBucketClaimForMonitor(context.TODO(), monitor, *claim, nil); err != nil {
t.Fatalf("reconcileBucketClaimForMonitor returned error: %v", err)
}
if got, w := workloadSize(t, c, "bucket-my-bucket"); got.Cmp(resource.MustParse("3Gi")) != 0 || w.OwnerReferences[0].UID != "uid-other" {
t.Errorf("held Workload changed: owners=%v size=%s", w.OwnerReferences, got.String())
}
list := &cozyv1alpha1.WorkloadList{}
if err := c.List(context.TODO(), list); err != nil {
t.Fatal(err)
}
if len(list.Items) != 1 {
t.Errorf("got %d Workloads, want only the held one", len(list.Items))
}
select {
case e := <-recorder.Events:
if !strings.Contains(e, "not metered") {
t.Errorf("event = %q, want one saying the bucket is not metered", e)
}
default:
t.Error("expected a BucketWorkloadNameTaken event")
}
}
// The legacy Workload can disappear between the lookup and the delete, for
// example through owner garbage collection; that must not fail the reconcile.
func TestReconcileBucketApp_LegacyAlreadyGone(t *testing.T) {
s := newTestScheme()
monitor, claim := bucketMonitorAndClaim("bucket-cozy-backups", "claim-uid")
c := fake.NewClientBuilder().WithScheme(s).
WithObjects(monitor, claim, legacyBucketWorkloadOwnedBy("claim-uid")).
WithStatusSubresource(monitor).
WithInterceptorFuncs(interceptor.Funcs{
Delete: func(ctx context.Context, c client.WithWatch, obj client.Object, opts ...client.DeleteOption) error {
if obj.GetName() == "bucket-bucket-cozy-backups" {
if err := c.Delete(ctx, obj, opts...); err != nil {
return err
}
return apierrors.NewNotFound(schema.GroupResource{Group: "cozystack.io", Resource: "workloads"}, obj.GetName())
}
return c.Delete(ctx, obj, opts...)
},
}).Build()
reconciler := &WorkloadMonitorReconciler{Client: c, Scheme: s}
// Reconcile only logs per-claim errors, so check the per-claim result
if err := reconciler.reconcileBucketClaimForMonitor(context.TODO(), monitor, *claim, nil); err != nil {
t.Fatalf("reconcileBucketClaimForMonitor returned error: %v", err)
}
if got, _ := workloadSize(t, c, "bucket-cozy-backups"); got.Cmp(resource.MustParse("689Gi")) != 0 {
t.Errorf("size = %s, want 689Gi carried over", got.String())
}
}
// A claim's own Workload, with its ownerRef as a live cluster's cache records
// it (Kind set), must keep being updated rather than look like another claim's.
func TestReconcileBucketClaim_UpdatesItsOwnWorkload(t *testing.T) {
s := newTestScheme()
monitor, claim := bucketMonitorAndClaim("bucket-x", "uid-x")
own := bucketWorkloadWithSize("bucket-x", "bucket-x", "uid-x", "1Gi")
c := fake.NewClientBuilder().WithScheme(s).
WithObjects(monitor, claim, own).
WithStatusSubresource(monitor).Build()
recorder := record.NewFakeRecorder(10)
reconciler := &WorkloadMonitorReconciler{Client: c, Scheme: s, Recorder: recorder}
req := reconcile.Request{NamespacedName: types.NamespacedName{Name: "bucket-x", Namespace: "tenant-demo"}}
if _, err := reconciler.Reconcile(context.TODO(), req); err != nil {
t.Fatalf("Reconcile returned error: %v", err)
}
_, w := workloadSize(t, c, "bucket-x")
if got := w.Labels[workloadMonitorLabel]; got != "bucket-x" {
t.Errorf("monitor label = %q, want the Workload updated by its own claim", got)
}
select {
case e := <-recorder.Events:
t.Errorf("unexpected event %q for the claim's own Workload", e)
default:
}
}
// A Workload that lost its owner references belongs to nobody, so the claim
// whose name it carries takes it over instead of being blocked by it forever.
func TestReconcileBucketClaim_AdoptsOwnerlessWorkload(t *testing.T) {
s := newTestScheme()
monitor, claim := bucketMonitorAndClaim("bucket-x", "uid-x")
orphan := bucketWorkloadWithSize("bucket-x", "bucket-x", "uid-x", "1Gi")
orphan.OwnerReferences = nil
c := fake.NewClientBuilder().WithScheme(s).
WithObjects(monitor, claim, orphan).
WithStatusSubresource(monitor).Build()
reconciler := &WorkloadMonitorReconciler{Client: c, Scheme: s}
req := reconcile.Request{NamespacedName: types.NamespacedName{Name: "bucket-x", Namespace: "tenant-demo"}}
if _, err := reconciler.Reconcile(context.TODO(), req); err != nil {
t.Fatalf("Reconcile returned error: %v", err)
}
_, w := workloadSize(t, c, "bucket-x")
if len(w.OwnerReferences) != 1 || w.OwnerReferences[0].UID != "uid-x" {
t.Errorf("owners = %v, want the claim uid-x", w.OwnerReferences)
}
}
// A controller rolled back to the doubled prefix attaches claim bucket-x to
// bucket-bucket-x, which by then is the Bucket app "bucket-x"'s renamed
// Workload. On the next upgrade bucket-x must let go of it, not delete it.
func TestReconcileBucketApp_ReleasesSharedLegacyWorkload(t *testing.T) {
s := newTestScheme()
monitor, claim := bucketMonitorAndClaim("bucket-x", "uid-x")
shared := bucketWorkloadWithSize("bucket-bucket-x", "bucket-bucket-x", "uid-bx", "100Gi")
shared.OwnerReferences = append(shared.OwnerReferences, metav1.OwnerReference{
APIVersion: "objectstorage.k8s.io/v1alpha1", Kind: "BucketClaim", Name: "bucket-x", UID: "uid-x",
})
c := fake.NewClientBuilder().WithScheme(s).
WithObjects(monitor, claim, shared).
WithStatusSubresource(monitor).Build()
reconciler := &WorkloadMonitorReconciler{Client: c, Scheme: s}
if err := reconciler.reconcileBucketClaimForMonitor(context.TODO(), monitor, *claim, nil); err != nil {
t.Fatalf("reconcileBucketClaimForMonitor returned error: %v", err)
}
got, w := workloadSize(t, c, "bucket-bucket-x")
if len(w.OwnerReferences) != 1 || w.OwnerReferences[0].UID != "uid-bx" {
t.Errorf("bucket-bucket-x owners = %v, want only uid-bx", w.OwnerReferences)
}
if got.Cmp(resource.MustParse("100Gi")) != 0 {
t.Errorf("bucket-bucket-x size = %s, want 100Gi kept", got.String())
}
if _, own := workloadSize(t, c, "bucket-x"); own.Status.Resources != nil {
if _, ok := own.Status.Resources[resourceS3StorageBytes]; ok {
t.Errorf("bucket-x took the other bucket's size: %v", own.Status.Resources)
}
}
}
// The legacy Workload can be replaced under the same name between the lookup
// and the delete; the replacement must survive.
func TestReconcileBucketApp_KeepsWorkloadThatReplacedLegacy(t *testing.T) {
s := newTestScheme()
monitor, claim := bucketMonitorAndClaim("bucket-cozy-backups", "claim-uid")
legacy := legacyBucketWorkloadOwnedBy("claim-uid")
legacy.UID = "legacy-uid"
c := fake.NewClientBuilder().WithScheme(s).
WithObjects(monitor, claim, legacy).
WithStatusSubresource(monitor).
WithInterceptorFuncs(interceptor.Funcs{
Delete: func(ctx context.Context, c client.WithWatch, obj client.Object, opts ...client.DeleteOption) error {
if obj.GetName() == "bucket-bucket-cozy-backups" {
if err := c.Delete(ctx, obj.DeepCopyObject().(client.Object)); err != nil {
return err
}
replacement := bucketWorkloadWithSize("bucket-bucket-cozy-backups", "bucket-bucket-cozy-backups", "other-uid", "5Gi")
replacement.UID = "replacement-uid"
if err := c.Create(ctx, replacement); err != nil {
return err
}
}
return c.Delete(ctx, obj, opts...)
},
}).Build()
reconciler := &WorkloadMonitorReconciler{Client: c, Scheme: s}
if err := reconciler.reconcileBucketClaimForMonitor(context.TODO(), monitor, *claim, nil); err != nil {
t.Fatalf("reconcileBucketClaimForMonitor returned error: %v", err)
}
got, w := workloadSize(t, c, "bucket-bucket-cozy-backups")
if len(w.OwnerReferences) != 1 || w.OwnerReferences[0].UID != "other-uid" || got.Cmp(resource.MustParse("5Gi")) != 0 {
t.Errorf("replacement Workload changed: owners=%v size=%s", w.OwnerReferences, got.String())
}
}
// When both the new and the legacy name are held by other owners, neither is
// taken over.
func TestReconcileBucketClaim_BothNamesHeldIsSkipped(t *testing.T) {
s := newTestScheme()
monitor, claim := bucketMonitorAndClaim("bucket-x", "uid-bx")
c := fake.NewClientBuilder().WithScheme(s).
WithObjects(monitor, claim,
bucketWorkloadWithSize("bucket-x", "x", "uid-x", "10Gi"),
bucketWorkloadWithSize("bucket-bucket-x", "other", "uid-other", "20Gi")).
WithStatusSubresource(monitor).Build()
reconciler := &WorkloadMonitorReconciler{Client: c, Scheme: s}
if err := reconciler.reconcileBucketClaimForMonitor(context.TODO(), monitor, *claim, nil); err != nil {
t.Fatalf("reconcileBucketClaimForMonitor returned error: %v", err)
}
for name, owner := range map[string]types.UID{"bucket-x": "uid-x", "bucket-bucket-x": "uid-other"} {
if _, w := workloadSize(t, c, name); len(w.OwnerReferences) != 1 || w.OwnerReferences[0].UID != owner {
t.Errorf("%s owners = %v, want only %s", name, w.OwnerReferences, owner)
}
}
}
func orphanedLegacyWorkload(monitor string) *cozyv1alpha1.Workload {
w := legacyBucketWorkloadOwnedBy("gone-uid")
w.OwnerReferences = nil
w.Labels = map[string]string{workloadMonitorLabel: monitor}
return w
}
// A claim deleted with --cascade=orphan and recreated leaves its legacy
// Workload without owners; it must not be metered next to the new one.
func TestReconcileBucketApp_RemovesOrphanedLegacyWorkload(t *testing.T) {
c := reconcileBucketApp(t, orphanedLegacyWorkload("bucket-cozy-backups"))
err := c.Get(context.TODO(), types.NamespacedName{Name: "bucket-bucket-cozy-backups", Namespace: "tenant-demo"}, &cozyv1alpha1.Workload{})
if !apierrors.IsNotFound(err) {
t.Errorf("expected the orphaned legacy Workload to be deleted, got error %v", err)
}
if _, w := workloadSize(t, c, "bucket-cozy-backups"); len(w.Status.Resources) != 0 {
t.Errorf("sizes carried over from an orphaned legacy Workload: %v", w.Status.Resources)
}
}
func TestReconcileBucketApp_KeepsOrphanOfAnotherMonitor(t *testing.T) {
c := reconcileBucketApp(t, orphanedLegacyWorkload("some-other-monitor"))
if err := c.Get(context.TODO(), types.NamespacedName{Name: "bucket-bucket-cozy-backups", Namespace: "tenant-demo"}, &cozyv1alpha1.Workload{}); err != nil {
t.Errorf("expected another monitor's orphan to stay, got error %v", err)
}
}
// Sizes recorded on a Workload the claim did not own yet were not written for
// it, so the legacy Workload's sizes take precedence.
func TestReconcileBucketApp_LegacySizeWinsOverOwnerlessNewWorkload(t *testing.T) {
stale := bucketWorkloadWithSize("bucket-cozy-backups", "bucket-cozy-backups", "claim-uid", "10Gi")
stale.OwnerReferences = nil
c := reconcileBucketApp(t, legacyBucketWorkloadOwnedBy("claim-uid"), stale)
if got, _ := workloadSize(t, c, "bucket-cozy-backups"); got.Cmp(resource.MustParse("689Gi")) != 0 {
t.Errorf("size = %s, want the legacy 689Gi over the stale 10Gi", got.String())
}
}
Reference in new issue View Git Blame Copy Permalink