diff --git a/api/core/v1alpha1/logset_types.go b/api/core/v1alpha1/logset_types.go index 3404b3ea..b32e4fa8 100644 --- a/api/core/v1alpha1/logset_types.go +++ b/api/core/v1alpha1/logset_types.go @@ -85,20 +85,31 @@ func (l *LogSetSpec) GetStoreFailureTimeout() metav1.Duration { } func (l *LogSetSpec) GetPVCRetentionPolicy() PVCRetentionPolicy { - if l.PVCRetentionPolicy == nil { - l.setDefaultRetentionPolicy() + if l.PVCRetentionPolicy != nil { + return *l.PVCRetentionPolicy } - return *l.PVCRetentionPolicy + // inherit from s3 policy if only s3 is set (e.g. old objects without pvcRetentionPolicy) + if l.SharedStorage.S3 != nil && l.SharedStorage.S3.S3RetentionPolicy != nil { + return *l.SharedStorage.S3.S3RetentionPolicy + } + return PVCRetentionPolicyDelete } func (l *LogSetSpec) GetS3RetentionPolicy() *PVCRetentionPolicy { if l.SharedStorage.S3 == nil { return nil } - if l.SharedStorage.S3.S3RetentionPolicy == nil { - l.setDefaultRetentionPolicy() + if l.SharedStorage.S3.S3RetentionPolicy != nil { + p := *l.SharedStorage.S3.S3RetentionPolicy + return &p + } + // inherit from pvc policy if only pvc is set (e.g. old objects without s3RetentionPolicy) + if l.PVCRetentionPolicy != nil { + p := *l.PVCRetentionPolicy + return &p } - return l.SharedStorage.S3.S3RetentionPolicy + defaultPolicy := PVCRetentionPolicyDelete + return &defaultPolicy } type InitialConfig struct { diff --git a/api/core/v1alpha1/logset_types_test.go b/api/core/v1alpha1/logset_types_test.go index d78e4141..9797cd71 100644 --- a/api/core/v1alpha1/logset_types_test.go +++ b/api/core/v1alpha1/logset_types_test.go @@ -1,4 +1,4 @@ -// Copyright 2025 Matrix Origin +// Copyright 2025-2026 Matrix Origin // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -105,3 +105,112 @@ func TestSetDefaultRetentionPolicy(t *testing.T) { } } } + +func TestGetRetentionPolicy(t *testing.T) { + del := PVCRetentionPolicyDelete + retain := PVCRetentionPolicyRetain + s3Tpl := func(policy *PVCRetentionPolicy) SharedStorageProvider { + return SharedStorageProvider{S3: &S3Provider{S3RetentionPolicy: policy}} + } + + testCases := []struct { + name string + logset LogSetSpec + pvcPolicy PVCRetentionPolicy + s3Policy *PVCRetentionPolicy + }{ + { + name: "both nil with s3 returns delete", + logset: LogSetSpec{SharedStorage: s3Tpl(nil)}, + pvcPolicy: del, + s3Policy: &del, + }, + { + name: "pvc only delete inherits to s3 getter", + logset: LogSetSpec{PVCRetentionPolicy: &del, SharedStorage: s3Tpl(nil)}, + pvcPolicy: del, + s3Policy: &del, + }, + { + name: "s3 only retain inherits to pvc getter", + logset: LogSetSpec{SharedStorage: s3Tpl(&retain)}, + pvcPolicy: retain, + s3Policy: &retain, + }, + { + name: "both set delete and retain independently", + logset: LogSetSpec{ + PVCRetentionPolicy: &del, + SharedStorage: s3Tpl(&retain), + }, + pvcPolicy: del, + s3Policy: &retain, + }, + { + name: "both set retain and delete independently", + logset: LogSetSpec{ + PVCRetentionPolicy: &retain, + SharedStorage: s3Tpl(&del), + }, + pvcPolicy: retain, + s3Policy: &del, + }, + { + name: "both set retain", + logset: LogSetSpec{ + PVCRetentionPolicy: &retain, + SharedStorage: s3Tpl(&retain), + }, + pvcPolicy: retain, + s3Policy: &retain, + }, + { + name: "both set delete", + logset: LogSetSpec{ + PVCRetentionPolicy: &del, + SharedStorage: s3Tpl(&del), + }, + pvcPolicy: del, + s3Policy: &del, + }, + { + name: "no s3 returns nil s3 policy", + logset: LogSetSpec{PVCRetentionPolicy: &retain}, + pvcPolicy: retain, + s3Policy: nil, + }, + { + name: "no s3 and no pvc returns delete", + logset: LogSetSpec{}, + pvcPolicy: del, + s3Policy: nil, + }, + } + + for _, c := range testCases { + t.Run(c.name, func(t *testing.T) { + spec := c.logset + origPVC := spec.PVCRetentionPolicy + var origS3 *PVCRetentionPolicy + if spec.SharedStorage.S3 != nil { + origS3 = spec.SharedStorage.S3.S3RetentionPolicy + } + + assert.Equal(t, c.pvcPolicy, spec.GetPVCRetentionPolicy()) + + s3Policy := spec.GetS3RetentionPolicy() + if c.s3Policy == nil { + assert.Nil(t, s3Policy) + } else { + assert.NotNil(t, s3Policy) + assert.Equal(t, *c.s3Policy, *s3Policy) + } + + // getters must not mutate spec + assert.Equal(t, origPVC, spec.PVCRetentionPolicy) + if spec.SharedStorage.S3 != nil { + assert.Equal(t, origS3, spec.SharedStorage.S3.S3RetentionPolicy) + } + }) + } +} diff --git a/go.mod b/go.mod index ac15f8f2..23e39753 100644 --- a/go.mod +++ b/go.mod @@ -23,7 +23,6 @@ require ( github.com/openkruise/kruise-api v1.4.0 github.com/prometheus/client_golang v1.17.0 github.com/samber/lo v1.38.1 - github.com/stretchr/testify v1.9.0 go.uber.org/multierr v1.11.0 go.uber.org/zap v1.24.0 golang.org/x/exp v0.0.0-20231006140011-7918f672742d @@ -161,6 +160,7 @@ require ( github.com/shirou/gopsutil/v3 v3.23.12 // indirect github.com/sirupsen/logrus v1.9.3 // indirect github.com/spf13/pflag v1.0.5 // indirect + github.com/stretchr/testify v1.9.0 // indirect github.com/tidwall/btree v1.6.0 // indirect github.com/tklauser/go-sysconf v0.3.12 // indirect github.com/tklauser/numcpus v0.6.1 // indirect diff --git a/pkg/controllers/bucketclaim/controller.go b/pkg/controllers/bucketclaim/controller.go index cf60c0ee..596a00bb 100644 --- a/pkg/controllers/bucketclaim/controller.go +++ b/pkg/controllers/bucketclaim/controller.go @@ -1,4 +1,4 @@ -// Copyright 2025 Matrix Origin +// Copyright 2025-2026 Matrix Origin // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -71,6 +71,16 @@ func (bca *Actor) Finalize(ctx *recon.Context[*v1alpha1.BucketClaim]) (bool, err return false, ctx.Update(bucket) } + // if bucket is Released (s3RetentionPolicy=Retain), skip data deletion and release the finalizer directly + if bucket.Status.State == v1alpha1.StatusReleased { + ctx.Log.Info(fmt.Sprintf("skip data deletion for released bucket %v (s3RetentionPolicy=Retain)", client.ObjectKeyFromObject(ctx.Obj))) + controllerutil.RemoveFinalizer(bucket, v1alpha1.BucketDataFinalizer) + if err := ctx.Update(bucket); err != nil { + return false, err + } + return true, nil + } + // if AnnAnyInstanceRunning is not set, indicates that there is no running pod instance found for its mo cluster // so there is no need to start a job to finalize bucket data. // NOTE: generally LogSet start successfully is a precondition of starting DN and CN, if no running pod found means that diff --git a/pkg/controllers/bucketclaim/controller_test.go b/pkg/controllers/bucketclaim/controller_test.go index e687f006..ce48e8ad 100644 --- a/pkg/controllers/bucketclaim/controller_test.go +++ b/pkg/controllers/bucketclaim/controller_test.go @@ -1,4 +1,4 @@ -// Copyright 2025 Matrix Origin +// Copyright 2025-2026 Matrix Origin // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -15,12 +15,71 @@ package bucketclaim import ( + "context" "testing" + "time" + "github.com/golang/mock/gomock" + "github.com/matrixorigin/controller-runtime/pkg/fake" "github.com/matrixorigin/matrixone-operator/api/core/v1alpha1" - "github.com/stretchr/testify/assert" + batchv1 "k8s.io/api/batch/v1" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + utilruntime "k8s.io/apimachinery/pkg/util/runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + + . "github.com/onsi/gomega" ) +func TestActor_Finalize_ReleasedSkipsS3CleanupJob(t *testing.T) { + g := NewGomegaWithT(t) + + now := metav1.NewTime(time.Now()) + bucket := &v1alpha1.BucketClaim{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-bucket", + Namespace: "default", + Finalizers: []string{v1alpha1.BucketDataFinalizer}, + DeletionTimestamp: &now, + Annotations: map[string]string{ + v1alpha1.AnnAnyInstanceRunning: "true", + }, + }, + Spec: v1alpha1.BucketClaimSpec{ + S3: &v1alpha1.S3Provider{Path: "minio-bucket/test"}, + }, + Status: v1alpha1.BucketClaimStatus{ + State: v1alpha1.StatusReleased, + }, + } + + scheme := runtime.NewScheme() + utilruntime.Must(v1alpha1.AddToScheme(scheme)) + utilruntime.Must(batchv1.AddToScheme(scheme)) + utilruntime.Must(corev1.AddToScheme(scheme)) + + cli := fake.KubeClientBuilder().WithScheme(scheme).WithObjects(bucket).Build() + + mockCtrl := gomock.NewController(t) + eventEmitter := fake.NewMockEventEmitter(mockCtrl) + ctx := fake.NewContext(bucket, cli, eventEmitter) + + actor := New() + ok, err := actor.Finalize(ctx) + g.Expect(err).To(Succeed()) + g.Expect(ok).To(BeTrue()) + + updated := &v1alpha1.BucketClaim{} + err = cli.Get(context.TODO(), client.ObjectKeyFromObject(bucket), updated) + g.Expect(apierrors.IsNotFound(err)).To(BeTrue()) + + jobs := &batchv1.JobList{} + g.Expect(cli.List(context.TODO(), jobs, client.InNamespace(bucket.Namespace))).To(Succeed()) + g.Expect(jobs.Items).To(BeEmpty()) +} + func TestParseEndpoint(t *testing.T) { s3 := v1alpha1.S3Provider{ Path: "minio-mo/test", @@ -56,6 +115,7 @@ else fi ` endpoint, err := parseEntrypoint(&s3) - assert.Nil(t, err) - assert.Equal(t, expect, endpoint) + g := NewGomegaWithT(t) + g.Expect(err).To(Succeed()) + g.Expect(endpoint).To(Equal(expect)) } diff --git a/test/e2e/bucket_test.go b/test/e2e/bucket_test.go index 4555793c..9d87904d 100644 --- a/test/e2e/bucket_test.go +++ b/test/e2e/bucket_test.go @@ -1,4 +1,4 @@ -// Copyright 2025 Matrix Origin +// Copyright 2025-2026 Matrix Origin // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -104,6 +104,107 @@ var _ = Describe("Matrix BucketClaim test", func() { }, waitBucketStatusTimeout, time.Second*2).Should(Succeed()) }) + It("Should keep S3 data when deleting a Released BucketClaim with pvc=Delete and s3=Retain", func() { + By("create logset with pvc=Delete and s3=Retain") + minioSecret := e2eutil.MinioSecret(env.Namespace) + Expect(kubeCli.Create(ctx, minioSecret)).To(Succeed()) + + minioProvider := e2eutil.MinioShareStorage(minioSecret.Name) + policyDelete := v1alpha1.PVCRetentionPolicyDelete + policyRetain := v1alpha1.PVCRetentionPolicyRetain + minioProvider.S3.S3RetentionPolicy = &policyRetain + ls := e2eutil.NewLogSetTpl(env.Namespace, fmt.Sprintf("%s:%s", moImageRepo, moVersion)) + ls.Spec.PVCRetentionPolicy = &policyDelete + ls.Spec.SharedStorage = minioProvider + Expect(kubeCli.Create(ctx, ls)).To(Succeed()) + + var bucket *v1alpha1.BucketClaim + var err error + Eventually(func() error { + bucket, err = v1alpha1.ClaimedBucket(kubeCli, minioProvider.S3) + if err != nil || bucket == nil { + return fmt.Errorf("wait bucket creating for logset %v, %v", client.ObjectKeyFromObject(ls), err) + } + expectedStatus := v1alpha1.BucketClaimStatus{ + BindTo: v1alpha1.BucketBindToMark(ls.ObjectMeta), + State: v1alpha1.StatusInUse, + } + if !reflect.DeepEqual(expectedStatus, bucket.Status) { + return fmt.Errorf("bucket status is not inuse, current %v", bucket.Status) + } + return nil + }, waitBucketStatusTimeout, time.Second*2).Should(Succeed()) + + By("put object to s3 path") + object, err := e2eminio.PutObject(minioProvider.S3.Path) + Expect(err).Should(BeNil()) + exist, err := e2eminio.IsObjectExist(object) + Expect(err).Should(BeNil()) + Expect(exist).Should(BeTrue()) + + By("wait logset available to set any-instance-running annotation") + Eventually(func() error { + if err := kubeCli.Get(ctx, client.ObjectKeyFromObject(ls), ls); err != nil { + return err + } + if len(ls.Status.AvailableStores) > 0 { + return nil + } + return fmt.Errorf("wait logset pod in running state") + }, createLogSetTimeout, pollInterval).Should(Succeed()) + Eventually(func() error { + if err := kubeCli.Get(ctx, client.ObjectKeyFromObject(bucket), bucket); err != nil { + return err + } + if bucket.Annotations[v1alpha1.AnnAnyInstanceRunning] == "" { + return fmt.Errorf("wait any-instance-running annotation on bucket") + } + return nil + }, waitBucketStatusTimeout, time.Second*2).Should(Succeed()) + + By("tear down logset cluster") + Expect(kubeCli.Delete(ctx, ls)).To(Succeed()) + Eventually(func() error { + return waitLogSetDeleted(ls) + }, teardownClusterTimeout, pollInterval).Should(Succeed()) + + By("bucket should in released state") + Eventually(func() error { + if err := kubeCli.Get(ctx, client.ObjectKeyFromObject(bucket), bucket); err != nil { + return err + } + if bucket.DeletionTimestamp != nil { + return fmt.Errorf("bucket should not be deleted before explicit bucketclaim deletion") + } + expectedStatus := v1alpha1.BucketClaimStatus{ + BindTo: "", + State: v1alpha1.StatusReleased, + } + if !reflect.DeepEqual(expectedStatus, bucket.Status) { + return fmt.Errorf("bucket status is not released, current %v", bucket.Status) + } + return nil + }, waitBucketStatusTimeout, time.Second*2).Should(Succeed()) + + By("delete released bucket claim with finalizer intact") + Expect(bucket.Finalizers).To(ContainElement(v1alpha1.BucketDataFinalizer)) + Expect(kubeCli.Delete(ctx, bucket)).To(Succeed()) + + By("wait bucket claim been deleted") + Eventually(func() error { + err := kubeCli.Get(ctx, client.ObjectKeyFromObject(bucket), bucket) + if apierrors.IsNotFound(err) { + return nil + } + return fmt.Errorf("bucket should be deleted") + }, waitBucketStatusTimeout, time.Second*2).Should(Succeed()) + + By("s3 object should still exist") + exist, err = e2eminio.IsObjectExist(object) + Expect(err).Should(BeNil()) + Expect(exist).Should(BeTrue()) + }) + It("Should bucket been deleted use delete retain policy", func() { By("create logset cluster with minio provider") minioSecret := e2eutil.MinioSecret(env.Namespace)