diff --git a/.gitignore b/.gitignore index e41669a1..9609ee3f 100644 --- a/.gitignore +++ b/.gitignore @@ -51,3 +51,4 @@ node_modules/ .chart-packages/ .cursor +.envrc diff --git a/.licenserc.yaml b/.licenserc.yaml index 46b56079..5d96dd35 100644 --- a/.licenserc.yaml +++ b/.licenserc.yaml @@ -39,5 +39,6 @@ header: - '**/*.json' - '**/.helmignore' - 'testbin/**' + - '.envrc' comment: on-failure diff --git a/api/v1alpha1/pulsartopic_types.go b/api/v1alpha1/pulsartopic_types.go index b285fbec..e14e7ddb 100644 --- a/api/v1alpha1/pulsartopic_types.go +++ b/api/v1alpha1/pulsartopic_types.go @@ -85,13 +85,17 @@ type PulsarTopicSpec struct { // RetentionTime specifies the minimum time to retain messages on the topic. // Should be set in conjunction with RetentionSize for effective retention policy. - // Retention Quota must exceed configured backlog quota for topic + // Retention Quota must exceed configured backlog quota for topic. + // Use "-1" for infinite retention time. + // Valid formats: "1h", "30m", "5s", "-1" // +optional RetentionTime *utils.Duration `json:"retentionTime,omitempty"` // RetentionSize specifies the maximum size of backlog retained on the topic. // Should be set in conjunction with RetentionTime for effective retention policy. - // Retention Quota must exceed configured backlog quota for topic + // Retention Quota must exceed configured backlog quota for topic. + // Use "-1" for infinite retention size. + // Valid formats: "1Gi", "500Mi", "100M", "-1" // +optional RetentionSize *resource.Quantity `json:"retentionSize,omitempty"` diff --git a/config/crd/bases/resource.streamnative.io_pulsartopics.yaml b/config/crd/bases/resource.streamnative.io_pulsartopics.yaml index a8e366fc..73a5a0b2 100644 --- a/config/crd/bases/resource.streamnative.io_pulsartopics.yaml +++ b/config/crd/bases/resource.streamnative.io_pulsartopics.yaml @@ -434,14 +434,18 @@ spec: description: |- RetentionSize specifies the maximum size of backlog retained on the topic. Should be set in conjunction with RetentionTime for effective retention policy. - Retention Quota must exceed configured backlog quota for topic + Retention Quota must exceed configured backlog quota for topic. + Use "-1" for infinite retention size. + Valid formats: "1Gi", "500Mi", "100M", "-1" pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$ x-kubernetes-int-or-string: true retentionTime: description: |- RetentionTime specifies the minimum time to retain messages on the topic. Should be set in conjunction with RetentionSize for effective retention policy. - Retention Quota must exceed configured backlog quota for topic + Retention Quota must exceed configured backlog quota for topic. + Use "-1" for infinite retention time. + Valid formats: "1h", "30m", "5s", "-1" type: string schemaCompatibilityStrategy: description: |- diff --git a/docs/pulsar_namespace.md b/docs/pulsar_namespace.md index 64a91ef8..966037d6 100644 --- a/docs/pulsar_namespace.md +++ b/docs/pulsar_namespace.md @@ -16,8 +16,8 @@ The `PulsarNamespace` resource defines a namespace in a Pulsar cluster. It allow | `maxConsumersPerTopic` | Maximum number of consumers allowed on a single topic in the namespace. | No | | `maxConsumersPerSubscription` | Maximum number of consumers allowed on a single subscription in the namespace. | No | | `messageTTL` | Time to Live (TTL) for messages in the namespace. Messages older than this TTL will be automatically marked as consumed. | No | -| `retentionTime` | Minimum time to retain messages in the namespace. Should be set in conjunction with RetentionSize for effective retention policy. | No | -| `retentionSize` | Maximum size of backlog retained in the namespace. Should be set in conjunction with RetentionTime for effective retention policy. | No | +| `retentionTime` | Minimum time to retain messages in the namespace. Should be set in conjunction with RetentionSize for effective retention policy. Use "-1" for infinite retention time. | No | +| `retentionSize` | Maximum size of backlog retained in the namespace. Should be set in conjunction with RetentionTime for effective retention policy. Use "-1" for infinite retention size. | No | | `backlogQuotaLimitTime` | Time limit for message backlog. Messages older than this limit will be removed or handled according to the retention policy. | No | | `backlogQuotaLimitSize` | Size limit for message backlog. When the limit is reached, older messages will be removed or handled according to the retention policy. | No | | `backlogQuotaRetentionPolicy` | Retention policy for messages when backlog quota is exceeded. Options: "producer_request_hold", "producer_exception", or "consumer_backlog_eviction". | No | @@ -393,6 +393,48 @@ The `replicationClusters` and `geoReplicationRefs` fields serve different purpos Choose `replicationClusters` for simpler, intra-instance replication, and `geoReplicationRefs` for more complex, inter-instance geo-replication scenarios. These fields are mutually exclusive; use only one depending on your replication requirements. +## Infinite Retention Configuration + +The `retentionTime` and `retentionSize` fields support infinite retention by using the special value `"-1"`. This is equivalent to passing -1 to Pulsar admin APIs and provides unlimited retention capabilities for all topics within the namespace. + +### Infinite Retention Time + +To set infinite retention time for the namespace, use the value `"-1"` for the `retentionTime` field: + +```yaml +spec: + retentionTime: "-1" # Messages will be retained indefinitely regardless of age + retentionSize: "100Gi" # Still limited by size +``` + +### Infinite Retention Size + +To set infinite retention size for the namespace, use the value `"-1"` for the `retentionSize` field: + +```yaml +spec: + retentionTime: "30d" # Still limited by time + retentionSize: "-1" # No size limit for message retention +``` + +### Complete Infinite Retention + +For completely unlimited retention (both time and size), set both fields to `"-1"`: + +```yaml +spec: + retentionTime: "-1" # Infinite time retention + retentionSize: "-1" # Infinite size retention +``` + +**Important Notes:** +- The `"-1"` value is case-sensitive and must be quoted in YAML +- Infinite retention should be used carefully as it can lead to unlimited storage consumption +- Retention quota must exceed configured backlog quota for the namespace +- These settings apply to all topics within the namespace by default +- Individual topics can override namespace-level retention settings +- Consider the storage and cost implications before enabling infinite retention + ## Create A Pulsar Namespace 1. Define a namespace named `test-tenant/testns` by using the YAML file and save the YAML file `namespace.yaml`. @@ -466,8 +508,8 @@ spec: # maxProducersPerTopic: 2 # maxConsumersPerTopic: 2 # maxConsumersPerSubscription: 2 - # retentionTime: 20h - # retentionSize: 2Gi + # retentionTime: 20h # or "-1" for infinite retention time + # retentionSize: 2Gi # or "-1" for infinite retention size # lifecyclePolicy: CleanUpAfterDeletion # topicAutoCreationConfig: # allow: true diff --git a/docs/pulsar_topic.md b/docs/pulsar_topic.md index b33eefa5..5e1888a6 100644 --- a/docs/pulsar_topic.md +++ b/docs/pulsar_topic.md @@ -19,8 +19,8 @@ The `PulsarTopic` resource defines a topic in a Pulsar cluster. It allows you to | `messageTTL` | Time to Live (TTL) for messages in the topic. Messages older than this TTL will be automatically marked as consumed. | No | | `maxUnAckedMessagesPerConsumer` | Maximum number of unacknowledged messages allowed per consumer. | No | | `maxUnAckedMessagesPerSubscription` | Maximum number of unacknowledged messages allowed per subscription. | No | -| `retentionTime` | Minimum time to retain messages in the topic. Should be set in conjunction with retentionSize for effective retention policy. | No | -| `retentionSize` | Maximum size of backlog retained in the topic. Should be set in conjunction with retentionTime for effective retention policy. | No | +| `retentionTime` | Minimum time to retain messages in the topic. Should be set in conjunction with retentionSize for effective retention policy. Use "-1" for infinite retention time. | No | +| `retentionSize` | Maximum size of backlog retained in the topic. Should be set in conjunction with retentionTime for effective retention policy. Use "-1" for infinite retention size. | No | | `backlogQuotaLimitTime` | Time limit for message backlog. Messages older than this limit will be removed or handled according to the retention policy. | No | | `backlogQuotaLimitSize` | Size limit for message backlog. When the limit is reached, older messages will be removed or handled according to the retention policy. | No | | `backlogQuotaRetentionPolicy` | Retention policy for messages when backlog quota is exceeded. Options: "producer_request_hold", "producer_exception", or "consumer_backlog_eviction". | No | @@ -68,6 +68,45 @@ The `replicationClusters` and `geoReplicationRefs` fields serve different purpos Choose `replicationClusters` for simpler, intra-instance replication, and `geoReplicationRefs` for more complex, inter-instance geo-replication scenarios. These fields are mutually exclusive; use only one depending on your replication requirements. +## Infinite Retention Configuration + +The `retentionTime` and `retentionSize` fields support infinite retention by using the special value `"-1"`. This is equivalent to passing -1 to Pulsar admin APIs and provides unlimited retention capabilities. + +### Infinite Retention Time + +To set infinite retention time, use the value `"-1"` for the `retentionTime` field: + +```yaml +spec: + retentionTime: "-1" # Messages will be retained indefinitely regardless of age + retentionSize: "10Gi" # Still limited by size +``` + +### Infinite Retention Size + +To set infinite retention size, use the value `"-1"` for the `retentionSize` field: + +```yaml +spec: + retentionTime: "7d" # Still limited by time + retentionSize: "-1" # No size limit for message retention +``` + +### Complete Infinite Retention + +For completely unlimited retention (both time and size), set both fields to `"-1"`: + +```yaml +spec: + retentionTime: "-1" # Infinite time retention + retentionSize: "-1" # Infinite size retention +``` + +**Important Notes:** +- The `"-1"` value is case-sensitive and must be quoted in YAML +- Infinite retention should be used carefully as it can lead to unlimited storage consumption +- Retention quota must exceed configured backlog quota for the topic +- Consider the storage and cost implications before enabling infinite retention ## Create A Pulsar Topic @@ -89,8 +128,8 @@ spec: # messageTTL: # maxUnAckedMessagesPerConsumer: # maxUnAckedMessagesPerSubscription: -# retentionTime: 20h -# retentionSize: 2Gi +# retentionTime: 20h # or "-1" for infinite retention time +# retentionSize: 2Gi # or "-1" for infinite retention size # backlogQuotaLimitTime: 24h # backlogQuotaLimitSize: 1Gi # backlogQuotaRetentionPolicy: producer_request_hold diff --git a/pkg/admin/impl.go b/pkg/admin/impl.go index 8dad13a3..700805c6 100644 --- a/pkg/admin/impl.go +++ b/pkg/admin/impl.go @@ -284,14 +284,22 @@ func (p *PulsarAdminClient) applyTopicPolicies(topicName *utils.TopicName, param retentionTime := -1 retentionSize := -1 if params.RetentionTime != nil { - t, err := params.RetentionTime.Parse() - if err != nil { - return err + if params.RetentionTime.IsInfinite() { + retentionTime = -1 // Infinite retention time + } else { + t, err := params.RetentionTime.Parse() + if err != nil { + return err + } + retentionTime = int(t.Minutes()) } - retentionTime = int(t.Minutes()) } if params.RetentionSize != nil { - retentionSize = int(params.RetentionSize.ScaledValue(resource.Mega)) + if rutils.IsInfiniteQuantity(params.RetentionSize) { + retentionSize = -1 // Infinite retention size + } else { + retentionSize = int(params.RetentionSize.ScaledValue(resource.Mega)) + } } retentionPolicy := utils.NewRetentionPolicies(retentionTime, retentionSize) err = p.adminClient.Topics().SetRetention(*topicName, retentionPolicy) @@ -646,14 +654,22 @@ func (p *PulsarAdminClient) applyNamespacePolicies(completeNSName string, params retentionTime := -1 retentionSize := -1 if params.RetentionTime != nil { - t, err := params.RetentionTime.Parse() - if err != nil { - return err + if params.RetentionTime.IsInfinite() { + retentionTime = -1 // Infinite retention time + } else { + t, err := params.RetentionTime.Parse() + if err != nil { + return err + } + retentionTime = int(t.Minutes()) } - retentionTime = int(t.Minutes()) } if params.RetentionSize != nil { - retentionSize = int(params.RetentionSize.ScaledValue(resource.Mega)) + if rutils.IsInfiniteQuantity(params.RetentionSize) { + retentionSize = -1 // Infinite retention size + } else { + retentionSize = int(params.RetentionSize.ScaledValue(resource.Mega)) + } } retentionPolicy := utils.NewRetentionPolicies(retentionTime, retentionSize) err = p.adminClient.Namespaces().SetRetention(completeNSName, retentionPolicy) diff --git a/pkg/utils/duration.go b/pkg/utils/duration.go index e7bb4785..aeb9028a 100644 --- a/pkg/utils/duration.go +++ b/pkg/utils/duration.go @@ -15,21 +15,74 @@ package utils import ( + "fmt" "time" str2duration "github.com/xhit/go-str2duration/v2" ) // Duration represents a elapsed time in string. +// Supports standard duration formats (e.g., "1h", "30m", "5s") and special value "-1" for infinite duration. type Duration string +const ( + // InfiniteDurationValue represents the special value for infinite duration + InfiniteDurationValue = "-1" +) + // Parse parses a duration from string. // Valid time units are "ns", "us" (or "µs"), "ms", "s", "m", "h", "d", "w". +// Special value "-1" represents infinite duration and returns -1 nanosecond. func (d *Duration) Parse() (time.Duration, error) { - var res time.Duration - res, err := str2duration.ParseDuration(string(*d)) + s := string(*d) + + // Handle infinite duration special case + if s == InfiniteDurationValue { + return time.Duration(-1), nil + } + + // Parse normal duration + res, err := str2duration.ParseDuration(s) if err != nil { - return res, err + return 0, fmt.Errorf("invalid duration format '%s': %w", s, err) } + + // Ensure duration is positive for non-infinite values + if res < 0 { + return 0, fmt.Errorf("duration must be positive or -1 for infinite, got: %s", s) + } + return res, nil } + +// IsInfinite returns true if the duration represents infinite duration ("-1"). +func (d *Duration) IsInfinite() bool { + return string(*d) == InfiniteDurationValue +} + +// ToSeconds returns the duration in seconds. +// Returns -1 for infinite duration, or the actual seconds for finite duration. +func (d *Duration) ToSeconds() (int64, error) { + if d.IsInfinite() { + return -1, nil + } + + duration, err := d.Parse() + if err != nil { + return 0, err + } + + return int64(duration.Seconds()), nil +} + +// String returns the string representation of the duration. +func (d Duration) String() string { + return string(d) +} + +// Validate validates the duration format. +// Accepts standard duration formats and the special value "-1". +func (d *Duration) Validate() error { + _, err := d.Parse() + return err +} diff --git a/pkg/utils/duration_test.go b/pkg/utils/duration_test.go new file mode 100644 index 00000000..e54b4641 --- /dev/null +++ b/pkg/utils/duration_test.go @@ -0,0 +1,170 @@ +// Copyright 2025 StreamNative +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package utils + +import ( + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("Duration", func() { + Context("Parse", func() { + It("should parse normal duration formats correctly", func() { + testCases := []struct { + input string + expected time.Duration + }{ + {"1h", time.Hour}, + {"30m", 30 * time.Minute}, + {"5s", 5 * time.Second}, + {"1d", 24 * time.Hour}, + {"2w", 14 * 24 * time.Hour}, + } + + for _, tc := range testCases { + duration := Duration(tc.input) + result, err := duration.Parse() + Expect(err).NotTo(HaveOccurred()) + Expect(result).To(Equal(tc.expected)) + } + }) + + It("should parse infinite duration (-1) correctly", func() { + duration := Duration("-1") + result, err := duration.Parse() + Expect(err).NotTo(HaveOccurred()) + Expect(result).To(Equal(time.Duration(-1))) + }) + + It("should return error for invalid duration formats", func() { + testCases := []string{ + "invalid", + "", + "abc", + "-2", // Only -1 is allowed for negative values + "-5m", // Negative durations other than -1 are not allowed + } + + for _, tc := range testCases { + duration := Duration(tc) + _, err := duration.Parse() + Expect(err).To(HaveOccurred()) + } + }) + }) + + Context("IsInfinite", func() { + It("should return true for -1 value", func() { + duration := Duration("-1") + Expect(duration.IsInfinite()).To(BeTrue()) + }) + + It("should return false for normal duration values", func() { + testCases := []string{ + "1h", + "30m", + "5s", + "0", + } + + for _, tc := range testCases { + duration := Duration(tc) + Expect(duration.IsInfinite()).To(BeFalse()) + } + }) + }) + + Context("ToSeconds", func() { + It("should return -1 for infinite duration", func() { + duration := Duration("-1") + seconds, err := duration.ToSeconds() + Expect(err).NotTo(HaveOccurred()) + Expect(seconds).To(Equal(int64(-1))) + }) + + It("should return correct seconds for normal durations", func() { + testCases := []struct { + input string + expected int64 + }{ + {"1h", 3600}, + {"30m", 1800}, + {"5s", 5}, + } + + for _, tc := range testCases { + duration := Duration(tc.input) + seconds, err := duration.ToSeconds() + Expect(err).NotTo(HaveOccurred()) + Expect(seconds).To(Equal(tc.expected)) + } + }) + + It("should return error for invalid duration formats", func() { + duration := Duration("invalid") + _, err := duration.ToSeconds() + Expect(err).To(HaveOccurred()) + }) + }) + + Context("Validate", func() { + It("should validate normal duration formats", func() { + testCases := []string{ + "1h", + "30m", + "5s", + "-1", + } + + for _, tc := range testCases { + duration := Duration(tc) + err := duration.Validate() + Expect(err).NotTo(HaveOccurred()) + } + }) + + It("should return error for invalid duration formats", func() { + testCases := []string{ + "invalid", + "", + "abc", + "-2", + } + + for _, tc := range testCases { + duration := Duration(tc) + err := duration.Validate() + Expect(err).To(HaveOccurred()) + } + }) + }) + + Context("String", func() { + It("should return original string representation", func() { + testCases := []string{ + "1h", + "30m", + "-1", + } + + for _, tc := range testCases { + duration := Duration(tc) + Expect(duration.String()).To(Equal(tc)) + } + }) + }) +}) diff --git a/pkg/utils/quantity.go b/pkg/utils/quantity.go new file mode 100644 index 00000000..7d8befef --- /dev/null +++ b/pkg/utils/quantity.go @@ -0,0 +1,65 @@ +// Copyright 2025 StreamNative +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package utils + +import ( + "k8s.io/apimachinery/pkg/api/resource" +) + +const ( + // InfiniteQuantityValue represents the special value for infinite quantity + InfiniteQuantityValue = "-1" +) + +// IsInfiniteQuantity returns true if the quantity represents infinite size ("-1"). +func IsInfiniteQuantity(q *resource.Quantity) bool { + if q == nil { + return false + } + return q.String() == InfiniteQuantityValue +} + +// QuantityToBytes returns the quantity value in bytes. +// Returns -1 for infinite quantity, or the actual bytes for finite quantity. +func QuantityToBytes(q *resource.Quantity) int64 { + if IsInfiniteQuantity(q) { + return -1 + } + + // Convert quantity to bytes + value, ok := q.AsInt64() + if !ok { + // If can't convert to int64, use approximation + return int64(q.AsApproximateFloat64()) + } + + return value +} + +// NewInfiniteQuantity creates a new quantity with infinite value ("-1"). +func NewInfiniteQuantity() *resource.Quantity { + q := resource.MustParse(InfiniteQuantityValue) + return &q +} + +// ValidateQuantityValue validates if a string is a valid quantity or "-1". +func ValidateQuantityValue(s string) error { + if s == InfiniteQuantityValue { + return nil + } + + _, err := resource.ParseQuantity(s) + return err +} diff --git a/pkg/utils/quantity_test.go b/pkg/utils/quantity_test.go new file mode 100644 index 00000000..1d001006 --- /dev/null +++ b/pkg/utils/quantity_test.go @@ -0,0 +1,118 @@ +// Copyright 2025 StreamNative +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package utils + +import ( + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "k8s.io/apimachinery/pkg/api/resource" +) + +var _ = Describe("Quantity Utils", func() { + Context("IsInfiniteQuantity", func() { + It("should return true for -1 quantity", func() { + q := resource.MustParse("-1") + Expect(IsInfiniteQuantity(&q)).To(BeTrue()) + }) + + It("should return false for normal quantities", func() { + testCases := []string{ + "1Gi", + "500Mi", + "100M", + "0", + } + + for _, tc := range testCases { + q := resource.MustParse(tc) + Expect(IsInfiniteQuantity(&q)).To(BeFalse()) + } + }) + + It("should return false for nil quantity", func() { + Expect(IsInfiniteQuantity(nil)).To(BeFalse()) + }) + }) + + Context("QuantityToBytes", func() { + It("should return -1 for infinite quantity", func() { + q := resource.MustParse("-1") + bytes := QuantityToBytes(&q) + Expect(bytes).To(Equal(int64(-1))) + }) + + It("should return correct bytes for normal quantities", func() { + testCases := []struct { + input string + expected int64 + }{ + {"1Ki", 1024}, + {"1Mi", 1024 * 1024}, + {"1Gi", 1024 * 1024 * 1024}, + {"100", 100}, + } + + for _, tc := range testCases { + q := resource.MustParse(tc.input) + bytes := QuantityToBytes(&q) + Expect(bytes).To(Equal(tc.expected)) + } + }) + }) + + Context("NewInfiniteQuantity", func() { + It("should create a quantity with -1 value", func() { + q := NewInfiniteQuantity() + Expect(q).NotTo(BeNil()) + Expect(q.String()).To(Equal("-1")) + Expect(IsInfiniteQuantity(q)).To(BeTrue()) + }) + }) + + Context("ValidateQuantityValue", func() { + It("should validate -1 as infinite quantity", func() { + err := ValidateQuantityValue("-1") + Expect(err).NotTo(HaveOccurred()) + }) + + It("should validate normal quantity formats", func() { + testCases := []string{ + "1Gi", + "500Mi", + "100M", + "1Ki", + "0", + } + + for _, tc := range testCases { + err := ValidateQuantityValue(tc) + Expect(err).NotTo(HaveOccurred()) + } + }) + + It("should return error for invalid quantity formats", func() { + testCases := []string{ + "invalid", + "abc", + "1.5.2", + } + + for _, tc := range testCases { + err := ValidateQuantityValue(tc) + Expect(err).To(HaveOccurred()) + } + }) + }) +}) diff --git a/tests/operator/resources_test.go b/tests/operator/resources_test.go index 0af19689..4bb85011 100644 --- a/tests/operator/resources_test.go +++ b/tests/operator/resources_test.go @@ -28,11 +28,13 @@ import ( v1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" "k8s.io/utils/pointer" v1alphav1 "github.com/streamnative/pulsar-resources-operator/api/v1alpha1" + rutils "github.com/streamnative/pulsar-resources-operator/pkg/utils" "github.com/streamnative/pulsar-resources-operator/tests/utils" ) @@ -1172,6 +1174,289 @@ var _ = Describe("Resources", func() { }) }) + Context("PulsarTopic Infinite Retention Policies", Ordered, func() { + var ( + infiniteRetentionTopic *v1alphav1.PulsarTopic + infiniteRetentionTopicName string = "test-infinite-retention-topic" + infiniteRetentionFullTopicName string = "persistent://cloud/stage/infinite-retention-test" + + infiniteTimeTopic *v1alphav1.PulsarTopic + infiniteTimeTopicName string = "test-infinite-time-topic" + infiniteTimeFullTopicName string = "persistent://cloud/stage/infinite-time-test" + + infiniteSizeTopic *v1alphav1.PulsarTopic + infiniteSizeTopicName string = "test-infinite-size-topic" + infiniteSizeFullTopicName string = "persistent://cloud/stage/infinite-size-test" + + finiteRetentionTopic *v1alphav1.PulsarTopic + finiteRetentionTopicName string = "test-finite-retention-topic" + finiteRetentionFullTopicName string = "persistent://cloud/stage/finite-retention-test" + ) + + BeforeAll(func() { + // Create topic with both infinite retention time and size + infiniteRetentionTopic = utils.MakePulsarTopicWithInfiniteRetention( + namespaceName, + infiniteRetentionTopicName, + infiniteRetentionFullTopicName, + pconnName, + lifecyclePolicy, + ) + + // Create topic with infinite retention time only + infiniteTimeTopic = utils.MakePulsarTopicWithInfiniteRetentionTime( + namespaceName, + infiniteTimeTopicName, + infiniteTimeFullTopicName, + pconnName, + lifecyclePolicy, + ) + + // Create topic with infinite retention size only + infiniteSizeTopic = utils.MakePulsarTopicWithInfiniteRetentionSize( + namespaceName, + infiniteSizeTopicName, + infiniteSizeFullTopicName, + pconnName, + lifecyclePolicy, + ) + + // Create topic with finite retention for comparison + finiteRetentionTopic = utils.MakePulsarTopicWithFiniteRetention( + namespaceName, + finiteRetentionTopicName, + finiteRetentionFullTopicName, + pconnName, + lifecyclePolicy, + ) + }) + + It("should create topic with infinite retention (both time and size) successfully", func() { + err := k8sClient.Create(ctx, infiniteRetentionTopic) + Expect(err == nil || apierrors.IsAlreadyExists(err)).Should(BeTrue()) + }) + + It("should be ready", func() { + Eventually(func() bool { + t := &v1alphav1.PulsarTopic{} + tns := types.NamespacedName{Namespace: namespaceName, Name: infiniteRetentionTopicName} + Expect(k8sClient.Get(ctx, tns, t)).Should(Succeed()) + return v1alphav1.IsPulsarResourceReady(t) + }, "20s", "100ms").Should(BeTrue()) + }) + + It("should have correct infinite retention configuration (both time and size)", func() { + topic := &v1alphav1.PulsarTopic{} + tns := types.NamespacedName{Namespace: namespaceName, Name: infiniteRetentionTopicName} + Expect(k8sClient.Get(ctx, tns, topic)).Should(Succeed()) + + // Verify infinite retention time + Expect(topic.Spec.RetentionTime).ShouldNot(BeNil()) + Expect(string(*topic.Spec.RetentionTime)).Should(Equal("-1")) + Expect(topic.Spec.RetentionTime.IsInfinite()).Should(BeTrue()) + + // Verify infinite retention size + Expect(topic.Spec.RetentionSize).ShouldNot(BeNil()) + Expect(topic.Spec.RetentionSize.String()).Should(Equal("-1")) + Expect(rutils.IsInfiniteQuantity(topic.Spec.RetentionSize)).Should(BeTrue()) + }) + + It("should create topic with infinite retention time only successfully", func() { + err := k8sClient.Create(ctx, infiniteTimeTopic) + Expect(err == nil || apierrors.IsAlreadyExists(err)).Should(BeTrue()) + }) + + It("should be ready", func() { + Eventually(func() bool { + t := &v1alphav1.PulsarTopic{} + tns := types.NamespacedName{Namespace: namespaceName, Name: infiniteTimeTopicName} + Expect(k8sClient.Get(ctx, tns, t)).Should(Succeed()) + return v1alphav1.IsPulsarResourceReady(t) + }, "20s", "100ms").Should(BeTrue()) + }) + + It("should have correct infinite retention time and finite size configuration", func() { + topic := &v1alphav1.PulsarTopic{} + tns := types.NamespacedName{Namespace: namespaceName, Name: infiniteTimeTopicName} + Expect(k8sClient.Get(ctx, tns, topic)).Should(Succeed()) + + // Verify infinite retention time + Expect(topic.Spec.RetentionTime).ShouldNot(BeNil()) + Expect(string(*topic.Spec.RetentionTime)).Should(Equal("-1")) + Expect(topic.Spec.RetentionTime.IsInfinite()).Should(BeTrue()) + + // Verify finite retention size + Expect(topic.Spec.RetentionSize).ShouldNot(BeNil()) + Expect(topic.Spec.RetentionSize.String()).Should(Equal("10Gi")) + Expect(rutils.IsInfiniteQuantity(topic.Spec.RetentionSize)).Should(BeFalse()) + }) + + It("should create topic with infinite retention size only successfully", func() { + err := k8sClient.Create(ctx, infiniteSizeTopic) + Expect(err == nil || apierrors.IsAlreadyExists(err)).Should(BeTrue()) + }) + + It("should be ready", func() { + Eventually(func() bool { + t := &v1alphav1.PulsarTopic{} + tns := types.NamespacedName{Namespace: namespaceName, Name: infiniteSizeTopicName} + Expect(k8sClient.Get(ctx, tns, t)).Should(Succeed()) + return v1alphav1.IsPulsarResourceReady(t) + }, "20s", "100ms").Should(BeTrue()) + }) + + It("should have correct finite retention time and infinite size configuration", func() { + topic := &v1alphav1.PulsarTopic{} + tns := types.NamespacedName{Namespace: namespaceName, Name: infiniteSizeTopicName} + Expect(k8sClient.Get(ctx, tns, topic)).Should(Succeed()) + + // Verify finite retention time + Expect(topic.Spec.RetentionTime).ShouldNot(BeNil()) + Expect(string(*topic.Spec.RetentionTime)).Should(Equal("7d")) + Expect(topic.Spec.RetentionTime.IsInfinite()).Should(BeFalse()) + + // Verify infinite retention size + Expect(topic.Spec.RetentionSize).ShouldNot(BeNil()) + Expect(topic.Spec.RetentionSize.String()).Should(Equal("-1")) + Expect(rutils.IsInfiniteQuantity(topic.Spec.RetentionSize)).Should(BeTrue()) + }) + + It("should create topic with finite retention for comparison successfully", func() { + err := k8sClient.Create(ctx, finiteRetentionTopic) + Expect(err == nil || apierrors.IsAlreadyExists(err)).Should(BeTrue()) + }) + + It("should be ready", func() { + Eventually(func() bool { + t := &v1alphav1.PulsarTopic{} + tns := types.NamespacedName{Namespace: namespaceName, Name: finiteRetentionTopicName} + Expect(k8sClient.Get(ctx, tns, t)).Should(Succeed()) + return v1alphav1.IsPulsarResourceReady(t) + }, "20s", "100ms").Should(BeTrue()) + }) + + It("should have correct finite retention configuration", func() { + topic := &v1alphav1.PulsarTopic{} + tns := types.NamespacedName{Namespace: namespaceName, Name: finiteRetentionTopicName} + Expect(k8sClient.Get(ctx, tns, topic)).Should(Succeed()) + + // Verify finite retention time + Expect(topic.Spec.RetentionTime).ShouldNot(BeNil()) + Expect(string(*topic.Spec.RetentionTime)).Should(Equal("24h")) + Expect(topic.Spec.RetentionTime.IsInfinite()).Should(BeFalse()) + + // Verify finite retention size + Expect(topic.Spec.RetentionSize).ShouldNot(BeNil()) + Expect(topic.Spec.RetentionSize.String()).Should(Equal("5Gi")) + Expect(rutils.IsInfiniteQuantity(topic.Spec.RetentionSize)).Should(BeFalse()) + }) + + It("should update finite retention to infinite retention successfully", func() { + topic := &v1alphav1.PulsarTopic{} + tns := types.NamespacedName{Namespace: namespaceName, Name: finiteRetentionTopicName} + Expect(k8sClient.Get(ctx, tns, topic)).Should(Succeed()) + + // Update to infinite retention + infiniteTime := rutils.Duration("-1") + infiniteSize := resource.MustParse("-1") + topic.Spec.RetentionTime = &infiniteTime + topic.Spec.RetentionSize = &infiniteSize + err := k8sClient.Update(ctx, topic) + Expect(err).Should(Succeed()) + }) + + It("should be ready after update to infinite retention", func() { + Eventually(func() bool { + t := &v1alphav1.PulsarTopic{} + tns := types.NamespacedName{Namespace: namespaceName, Name: finiteRetentionTopicName} + Expect(k8sClient.Get(ctx, tns, t)).Should(Succeed()) + return v1alphav1.IsPulsarResourceReady(t) + }, "20s", "100ms").Should(BeTrue()) + }) + + It("should have updated infinite retention configuration", func() { + topic := &v1alphav1.PulsarTopic{} + tns := types.NamespacedName{Namespace: namespaceName, Name: finiteRetentionTopicName} + Expect(k8sClient.Get(ctx, tns, topic)).Should(Succeed()) + + // Verify updated infinite retention time + Expect(topic.Spec.RetentionTime).ShouldNot(BeNil()) + Expect(string(*topic.Spec.RetentionTime)).Should(Equal("-1")) + Expect(topic.Spec.RetentionTime.IsInfinite()).Should(BeTrue()) + + // Verify updated infinite retention size + Expect(topic.Spec.RetentionSize).ShouldNot(BeNil()) + Expect(topic.Spec.RetentionSize.String()).Should(Equal("-1")) + Expect(rutils.IsInfiniteQuantity(topic.Spec.RetentionSize)).Should(BeTrue()) + }) + + It("should update infinite retention back to finite retention successfully", func() { + topic := &v1alphav1.PulsarTopic{} + tns := types.NamespacedName{Namespace: namespaceName, Name: finiteRetentionTopicName} + Expect(k8sClient.Get(ctx, tns, topic)).Should(Succeed()) + + // Update back to finite retention + finiteTime := rutils.Duration("12h") + finiteSize := resource.MustParse("2Gi") + topic.Spec.RetentionTime = &finiteTime + topic.Spec.RetentionSize = &finiteSize + err := k8sClient.Update(ctx, topic) + Expect(err).Should(Succeed()) + }) + + It("should be ready after update back to finite retention", func() { + Eventually(func() bool { + t := &v1alphav1.PulsarTopic{} + tns := types.NamespacedName{Namespace: namespaceName, Name: finiteRetentionTopicName} + Expect(k8sClient.Get(ctx, tns, t)).Should(Succeed()) + return v1alphav1.IsPulsarResourceReady(t) + }, "20s", "100ms").Should(BeTrue()) + }) + + It("should have updated finite retention configuration", func() { + topic := &v1alphav1.PulsarTopic{} + tns := types.NamespacedName{Namespace: namespaceName, Name: finiteRetentionTopicName} + Expect(k8sClient.Get(ctx, tns, topic)).Should(Succeed()) + + // Verify updated finite retention time + Expect(topic.Spec.RetentionTime).ShouldNot(BeNil()) + Expect(string(*topic.Spec.RetentionTime)).Should(Equal("12h")) + Expect(topic.Spec.RetentionTime.IsInfinite()).Should(BeFalse()) + + // Verify updated finite retention size + Expect(topic.Spec.RetentionSize).ShouldNot(BeNil()) + Expect(topic.Spec.RetentionSize.String()).Should(Equal("2Gi")) + Expect(rutils.IsInfiniteQuantity(topic.Spec.RetentionSize)).Should(BeFalse()) + }) + + AfterAll(func() { + // Clean up all test topics + topics := []*v1alphav1.PulsarTopic{ + infiniteRetentionTopic, + infiniteTimeTopic, + infiniteSizeTopic, + finiteRetentionTopic, + } + topicNames := []string{ + infiniteRetentionTopicName, + infiniteTimeTopicName, + infiniteSizeTopicName, + finiteRetentionTopicName, + } + + for i, topic := range topics { + if topic != nil { + Eventually(func(g Gomega) { + t := &v1alphav1.PulsarTopic{} + tns := types.NamespacedName{Namespace: namespaceName, Name: topicNames[i]} + g.Expect(k8sClient.Get(ctx, tns, t)).Should(Succeed()) + g.Expect(k8sClient.Delete(ctx, t)).Should(Succeed()) + }).Should(Succeed()) + } + } + }) + }) + Context("PulsarNamespace Rate Limiting", Ordered, func() { var ( rateLimitingNamespace *v1alphav1.PulsarNamespace diff --git a/tests/utils/spec.go b/tests/utils/spec.go index cb89097d..9396b620 100644 --- a/tests/utils/spec.go +++ b/tests/utils/spec.go @@ -749,3 +749,87 @@ func MakePulsarTopicWithSchemaCompatibilityStrategy(namespace, name, topicName, }, } } + +// MakePulsarTopicWithInfiniteRetention creates a PulsarTopic with infinite retention policies (both time and size) +func MakePulsarTopicWithInfiniteRetention(namespace, name, topicName, connectionName string, policy v1alpha1.PulsarResourceLifeCyclePolicy) *v1alpha1.PulsarTopic { + infiniteTime := rsutils.Duration("-1") + infiniteSize := resource.MustParse("-1") + return &v1alpha1.PulsarTopic{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: namespace, + Name: name, + }, + Spec: v1alpha1.PulsarTopicSpec{ + ConnectionRef: corev1.LocalObjectReference{ + Name: connectionName, + }, + Name: topicName, + LifecyclePolicy: policy, + RetentionTime: &infiniteTime, + RetentionSize: &infiniteSize, + }, + } +} + +// MakePulsarTopicWithInfiniteRetentionTime creates a PulsarTopic with infinite retention time only +func MakePulsarTopicWithInfiniteRetentionTime(namespace, name, topicName, connectionName string, policy v1alpha1.PulsarResourceLifeCyclePolicy) *v1alpha1.PulsarTopic { + infiniteTime := rsutils.Duration("-1") + finiteSize := resource.MustParse("10Gi") + return &v1alpha1.PulsarTopic{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: namespace, + Name: name, + }, + Spec: v1alpha1.PulsarTopicSpec{ + ConnectionRef: corev1.LocalObjectReference{ + Name: connectionName, + }, + Name: topicName, + LifecyclePolicy: policy, + RetentionTime: &infiniteTime, + RetentionSize: &finiteSize, + }, + } +} + +// MakePulsarTopicWithInfiniteRetentionSize creates a PulsarTopic with infinite retention size only +func MakePulsarTopicWithInfiniteRetentionSize(namespace, name, topicName, connectionName string, policy v1alpha1.PulsarResourceLifeCyclePolicy) *v1alpha1.PulsarTopic { + finiteTime := rsutils.Duration("7d") + infiniteSize := resource.MustParse("-1") + return &v1alpha1.PulsarTopic{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: namespace, + Name: name, + }, + Spec: v1alpha1.PulsarTopicSpec{ + ConnectionRef: corev1.LocalObjectReference{ + Name: connectionName, + }, + Name: topicName, + LifecyclePolicy: policy, + RetentionTime: &finiteTime, + RetentionSize: &infiniteSize, + }, + } +} + +// MakePulsarTopicWithFiniteRetention creates a PulsarTopic with finite retention policies for comparison +func MakePulsarTopicWithFiniteRetention(namespace, name, topicName, connectionName string, policy v1alpha1.PulsarResourceLifeCyclePolicy) *v1alpha1.PulsarTopic { + finiteTime := rsutils.Duration("24h") + finiteSize := resource.MustParse("5Gi") + return &v1alpha1.PulsarTopic{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: namespace, + Name: name, + }, + Spec: v1alpha1.PulsarTopicSpec{ + ConnectionRef: corev1.LocalObjectReference{ + Name: connectionName, + }, + Name: topicName, + LifecyclePolicy: policy, + RetentionTime: &finiteTime, + RetentionSize: &finiteSize, + }, + } +}