From 7ab5d7384117fd49266aeb7e29cd1acb08ba4f59 Mon Sep 17 00:00:00 2001 From: tdakkota Date: Tue, 28 Jul 2026 14:49:17 +0300 Subject: [PATCH 1/2] feat(api): add spec.retention and spec.limits Model oteldb's storage.policy.retention (max_age, max_bytes) and storage.policy.limits (ingest rate, in-flight bytes, series cardinality, part size) instead of leaving them to extraConfig. Both paths join reservedConfigPaths; the rest of storage.policy stays mergeable. Co-Authored-By: Claude Opus 5 (1M context) --- README.md | 8 + api/v1alpha1/oteldbcluster_types.go | 61 ++++- config/samples/db_v1alpha1_oteldbcluster.yaml | 14 + internal/controller/config.go | 8 + internal/controller/extraconfig.go | 4 + internal/controller/policy.go | 117 ++++++++ internal/controller/policy_test.go | 249 ++++++++++++++++++ 7 files changed, 459 insertions(+), 2 deletions(-) create mode 100644 internal/controller/policy.go create mode 100644 internal/controller/policy_test.go diff --git a/README.md b/README.md index 76646cb..69da294 100644 --- a/README.md +++ b/README.md @@ -73,6 +73,9 @@ for a fuller example including the S3 backend. | `cluster.staticZone` | Fixed failure-domain label for the cluster's nodes (ring zone-spreading). | | `signals` | Which signals to serve (all default on). Disabling one drops its backend, its API bind and its ports; disabling all is rejected. | | `engine` | Storage engine tuning: `flushInterval`, `readCacheSize`, `decodeCacheSize`, `decodeMemoryLimit`, `aggregateStats`. | +| `retention.maxAge` | How long data is kept (e.g. `720h`). Empty retains forever. Enforced at merge time by dropping whole partitions, so data can outlive the window briefly. | +| `retention.maxBytes` | Retained-bytes budget. **Accepted but not enforced yet** by the storage engine ([oteldb/storage#224](https://github.com/oteldb/storage/issues/224)) — use `maxAge` to bound disk growth. | +| `limits` | Per-node admission control: `ingestBytesPerSecond`, `maxInFlightBytes`, `maxSeries`, `maxSeriesSoft`, `maxPartSize`. Over-budget writes are shed as OTLP partial success rather than buffered. | | `service.type` / `annotations` | Client Service exposing the query/ingest APIs. | | `resources`, `nodeSelector`, `affinity`, `tolerations`, `topologySpreadConstraints`, `podSecurityContext`, `securityContext`, `podAnnotations`, `podLabels`, `serviceAccountName` | Standard pod scheduling/security knobs. | | `extraConfig` | Arbitrary raw oteldb config **deep-merged** over the generated config — for fields the CRD does not model (auth, retention policy, prometheus tuning, …). Nested objects merge key by key (`storage.policy` does not wipe `storage.backend`); operator-owned paths are [reserved](#reserved-extraconfig-paths). | @@ -101,6 +104,11 @@ spec field to use instead. | `storage.s3` | `spec.storage.s3` | | `storage.cluster` (whole subtree) | `spec.cluster`, `spec.etcd.endpoints` | | `storage.flush_interval`, `storage.read_cache_bytes`, `storage.decode_cache_bytes`, `storage.decode_memory_bytes`, `storage.aggregate_stats` | `spec.engine` | +| `storage.policy.retention` | `spec.retention` | +| `storage.policy.limits` | `spec.limits` | + +The rest of `storage.policy` — `precision`, `downsample`, `recompress` — is not modelled by the +CRD and stays mergeable, as in the example above. ### Status diff --git a/api/v1alpha1/oteldbcluster_types.go b/api/v1alpha1/oteldbcluster_types.go index cd5f219..df0389c 100644 --- a/api/v1alpha1/oteldbcluster_types.go +++ b/api/v1alpha1/oteldbcluster_types.go @@ -78,6 +78,14 @@ type OtelDBClusterSpec struct { // +optional Engine EngineSpec `json:"engine,omitempty"` + // Retention bounds how long ingested data is kept. Empty retains forever. + // +optional + Retention RetentionSpec `json:"retention,omitempty"` + + // Limits are the per-node admission-control limits. Empty means unlimited. + // +optional + Limits LimitsSpec `json:"limits,omitempty"` + // Service configures the client-facing Service that exposes the query and ingest APIs. // +optional Service ServiceSpec `json:"service,omitempty"` @@ -135,8 +143,10 @@ type OtelDBClusterSpec struct { // instead of being merged: metrics_backend, traces_backend, logs_backend, profiles_backend, // storage.backend, storage.dir, storage.wal_dir, storage.s3, storage.cluster (and everything // below it), storage.flush_interval, storage.read_cache_bytes, storage.decode_cache_bytes, - // storage.decode_memory_bytes and storage.aggregate_stats. Configure those through - // spec.storage, spec.cluster, spec.etcd, spec.signals and spec.engine. + // storage.decode_memory_bytes, storage.aggregate_stats, storage.policy.retention and + // storage.policy.limits. Configure those through spec.storage, spec.cluster, spec.etcd, + // spec.signals, spec.engine, spec.retention and spec.limits. The rest of storage.policy + // (precision, downsample, recompress) stays mergeable. // +optional // +kubebuilder:pruning:PreserveUnknownFields ExtraConfig *runtime.RawExtension `json:"extraConfig,omitempty"` @@ -324,6 +334,53 @@ type EngineSpec struct { AggregateStats *bool `json:"aggregateStats,omitempty"` } +// RetentionSpec bounds how long data is kept. Enforcement happens at merge time and drops whole +// partitions — never individual rows — so data can outlive the window until the partition holding +// it has fully expired. +type RetentionSpec struct { + // MaxAge is the maximum age of retained data (e.g. "720h"). Empty retains forever. + // +optional + MaxAge *metav1.Duration `json:"maxAge,omitempty"` + + // MaxBytes is the total retained-bytes budget across every signal on a node. + // + // oteldb accepts it, but the storage engine does not enforce it yet (oteldb/storage#224), so + // setting it alone bounds nothing today. Use MaxAge to bound disk growth. + // +optional + MaxBytes *resource.Quantity `json:"maxBytes,omitempty"` +} + +// LimitsSpec are the per-node admission-control limits. They shed over-budget writes and report +// them as OTLP partial success (RESOURCE_EXHAUSTED), so an overload degrades rather than OOMs. +type LimitsSpec struct { + // IngestBytesPerSecond caps the ingest rate, bursting to one second of budget. + // +optional + IngestBytesPerSecond *resource.Quantity `json:"ingestBytesPerSecond,omitempty"` + + // MaxInFlightBytes caps the unflushed in-flight bytes buffered before backpressure sheds. + // +optional + MaxInFlightBytes *resource.Quantity `json:"maxInFlightBytes,omitempty"` + + // MaxSeries is the hard active-series ceiling: a sample minting a new series past it is shed. + // Existing series are unaffected. + // +kubebuilder:validation:Minimum=0 + // +optional + MaxSeries *int64 `json:"maxSeries,omitempty"` + + // MaxSeriesSoft is a soft cardinality budget (metrics only): past it a new series' samples go + // to a synthetic per-metric overflow series instead of being shed, until MaxSeries is reached. + // It must not exceed MaxSeries, and needs MaxSeries set to have any effect. + // +kubebuilder:validation:Minimum=0 + // +optional + MaxSeriesSoft *int64 `json:"maxSeriesSoft,omitempty"` + + // MaxPartSize caps an immutable part's approximate uncompressed size; flush and merge split + // their output to respect it. It is structural: fixed when a node's engine is first created, + // so changing it does not affect existing data. + // +optional + MaxPartSize *resource.Quantity `json:"maxPartSize,omitempty"` +} + // ServiceSpec configures the client-facing Service. type ServiceSpec struct { // Type of the client Service. diff --git a/config/samples/db_v1alpha1_oteldbcluster.yaml b/config/samples/db_v1alpha1_oteldbcluster.yaml index a5bfb9f..726a345 100644 --- a/config/samples/db_v1alpha1_oteldbcluster.yaml +++ b/config/samples/db_v1alpha1_oteldbcluster.yaml @@ -36,6 +36,20 @@ spec: traces: true profiles: true + # Keep 30 days of data. Enforced at merge time by dropping whole partitions, so data can + # outlive the window until the partition holding it has fully expired. + retention: + maxAge: 720h + # maxBytes is accepted but not enforced by the storage engine yet (oteldb/storage#224). + + # Per-node admission control: over-budget writes are shed and reported as OTLP partial + # success, so an overload degrades instead of OOMing. + limits: + maxSeries: 2000000 + maxSeriesSoft: 1500000 # past this, new series fold into a per-metric overflow series + maxInFlightBytes: 1Gi + maxPartSize: 256Mi + resources: requests: cpu: "1" diff --git a/internal/controller/config.go b/internal/controller/config.go index ecc8583..b0114db 100644 --- a/internal/controller/config.go +++ b/internal/controller/config.go @@ -32,6 +32,9 @@ func renderConfig(cr *dbv1alpha1.OtelDBCluster, etcdEndpoints []string) (string, if err := validateSignals(cr); err != nil { return "", err } + if err := validatePolicy(cr); err != nil { + return "", err + } cfg := map[string]any{ "health_check": map[string]any{keyBind: "0.0.0.0:13133"}, @@ -119,6 +122,11 @@ func renderConfig(cr *dbv1alpha1.OtelDBCluster, etcdEndpoints []string) (string, storage["aggregate_stats"] = *eng.AggregateStats } + // Retention and admission-control limits ride the per-tenant storage policy. + if policy := renderPolicy(cr); policy != nil { + storage[keyPolicy] = policy + } + cfg["storage"] = storage // Merge user-supplied ExtraConfig over the generated config. The merge is recursive so that, diff --git a/internal/controller/extraconfig.go b/internal/controller/extraconfig.go index 4655ae3..fbcac30 100644 --- a/internal/controller/extraconfig.go +++ b/internal/controller/extraconfig.go @@ -46,6 +46,10 @@ var reservedConfigPaths = map[string]string{ "storage.decode_cache_bytes": "use spec.engine.decodeCacheSize", "storage.decode_memory_bytes": "use spec.engine.decodeMemoryLimit", "storage.aggregate_stats": "use spec.engine.aggregateStats", + + // The rest of storage.policy (precision, downsample, recompress) stays mergeable. + "storage.policy.retention": "use spec.retention", + "storage.policy.limits": "use spec.limits", } // validationError marks a spec problem that no amount of retrying can fix: the reconcile is diff --git a/internal/controller/policy.go b/internal/controller/policy.go new file mode 100644 index 0000000..9ba817d --- /dev/null +++ b/internal/controller/policy.go @@ -0,0 +1,117 @@ +/* +Copyright 2026. + +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 controller + +import ( + "k8s.io/apimachinery/pkg/api/resource" + + dbv1alpha1 "github.com/oteldb/operator/api/v1alpha1" +) + +// oteldb storage.policy config keys. +const ( + keyPolicy = "policy" + keyRetention = "retention" + keyLimits = "limits" +) + +// renderPolicy builds the storage.policy block from spec.retention and spec.limits, or returns nil +// when neither is set — oteldb installs no tenancy resolver for an absent policy, which is the +// library default (retain forever, no limits). +func renderPolicy(cr *dbv1alpha1.OtelDBCluster) map[string]any { + policy := map[string]any{} + + if retention := renderRetention(cr.Spec.Retention); len(retention) > 0 { + policy[keyRetention] = retention + } + if limits := renderLimits(cr.Spec.Limits); len(limits) > 0 { + policy[keyLimits] = limits + } + + if len(policy) == 0 { + return nil + } + return policy +} + +func renderRetention(spec dbv1alpha1.RetentionSpec) map[string]any { + m := map[string]any{} + if spec.MaxAge != nil { + m["max_age"] = spec.MaxAge.Duration.String() + } + if spec.MaxBytes != nil { + m["max_bytes"] = spec.MaxBytes.Value() + } + return m +} + +func renderLimits(spec dbv1alpha1.LimitsSpec) map[string]any { + m := map[string]any{} + if spec.IngestBytesPerSecond != nil { + m["ingest_bytes_per_second"] = spec.IngestBytesPerSecond.Value() + } + if spec.MaxInFlightBytes != nil { + m["max_in_flight_bytes"] = spec.MaxInFlightBytes.Value() + } + if spec.MaxSeries != nil { + m["max_series"] = *spec.MaxSeries + } + if spec.MaxSeriesSoft != nil { + m["max_series_soft"] = *spec.MaxSeriesSoft + } + if spec.MaxPartSize != nil { + m["max_part_size"] = spec.MaxPartSize.Value() + } + return m +} + +// validatePolicy rejects retention and limit values oteldb would silently treat as "unset" or that +// contradict each other. A negative quantity is always a mistake; the zero value is the documented +// "unlimited", so it is left alone. +func validatePolicy(cr *dbv1alpha1.OtelDBCluster) error { + retention := cr.Spec.Retention + if retention.MaxAge != nil && retention.MaxAge.Duration < 0 { + return invalidSpec("spec.retention.maxAge must not be negative, got %s", retention.MaxAge.Duration) + } + limits := cr.Spec.Limits + for _, q := range []struct { + field string + value *resource.Quantity + }{ + {"spec.retention.maxBytes", retention.MaxBytes}, + {"spec.limits.ingestBytesPerSecond", limits.IngestBytesPerSecond}, + {"spec.limits.maxInFlightBytes", limits.MaxInFlightBytes}, + {"spec.limits.maxPartSize", limits.MaxPartSize}, + } { + if q.value != nil && q.value.Sign() < 0 { + return invalidSpec("%s must not be negative, got %s", q.field, q.value.String()) + } + } + + // A soft budget above the hard ceiling never engages: the hard limit sheds first, so the + // overflow series the soft budget promises are never minted. + if limits.MaxSeriesSoft != nil && *limits.MaxSeriesSoft > 0 { + if limits.MaxSeries == nil || *limits.MaxSeries <= 0 { + return invalidSpec("spec.limits.maxSeriesSoft needs spec.limits.maxSeries to be set") + } + if *limits.MaxSeriesSoft > *limits.MaxSeries { + return invalidSpec("spec.limits.maxSeriesSoft (%d) must not exceed spec.limits.maxSeries (%d)", + *limits.MaxSeriesSoft, *limits.MaxSeries) + } + } + return nil +} diff --git a/internal/controller/policy_test.go b/internal/controller/policy_test.go new file mode 100644 index 0000000..0d7a465 --- /dev/null +++ b/internal/controller/policy_test.go @@ -0,0 +1,249 @@ +/* +Copyright 2026. + +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 controller + +import ( + "errors" + "testing" + "time" + + "github.com/stretchr/testify/require" + "k8s.io/apimachinery/pkg/api/resource" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/utils/ptr" + "sigs.k8s.io/yaml" + + dbv1alpha1 "github.com/oteldb/operator/api/v1alpha1" +) + +// renderStorage renders cr and returns its storage block. +func renderStorage(t *testing.T, cr *dbv1alpha1.OtelDBCluster) map[string]any { + t.Helper() + out, err := renderConfig(cr, cr.Spec.Etcd.Endpoints) + require.NoError(t, err) + + var cfg map[string]any + require.NoError(t, yaml.Unmarshal([]byte(out), &cfg)) + storage, ok := cfg["storage"].(map[string]any) + require.True(t, ok, "storage block missing:\n%s", out) + return storage +} + +func TestRenderPolicyAbsentByDefault(t *testing.T) { + storage := renderStorage(t, testCluster()) + require.NotContains(t, storage, "policy", + "no policy block should be rendered when neither retention nor limits is set") +} + +func TestRenderPolicyRetention(t *testing.T) { + cr := testCluster() + cr.Spec.Retention = dbv1alpha1.RetentionSpec{ + MaxAge: &metav1.Duration{Duration: 720 * time.Hour}, + MaxBytes: ptr.To(resource.MustParse("500Gi")), + } + + policy, ok := renderStorage(t, cr)["policy"].(map[string]any) + require.True(t, ok, "policy block missing") + require.Equal(t, map[string]any{ + "max_age": "720h0m0s", + "max_bytes": float64(500 * 1024 * 1024 * 1024), + }, policy["retention"]) + require.NotContains(t, policy, "limits", "limits must stay absent when unset") +} + +func TestRenderPolicyLimits(t *testing.T) { + cr := testCluster() + cr.Spec.Limits = dbv1alpha1.LimitsSpec{ + IngestBytesPerSecond: ptr.To(resource.MustParse("50Mi")), + MaxInFlightBytes: ptr.To(resource.MustParse("1Gi")), + MaxSeries: ptr.To[int64](1_000_000), + MaxSeriesSoft: ptr.To[int64](800_000), + MaxPartSize: ptr.To(resource.MustParse("256Mi")), + } + + policy, ok := renderStorage(t, cr)["policy"].(map[string]any) + require.True(t, ok, "policy block missing") + require.Equal(t, map[string]any{ + "ingest_bytes_per_second": float64(50 * 1024 * 1024), + "max_in_flight_bytes": float64(1024 * 1024 * 1024), + "max_series": float64(1_000_000), + "max_series_soft": float64(800_000), + "max_part_size": float64(256 * 1024 * 1024), + }, policy["limits"]) + require.NotContains(t, policy, "retention", "retention must stay absent when unset") +} + +// The policy must not disturb the rest of the storage block, which is where the shallow-merge bug +// (issue #1) did its damage. +func TestRenderPolicyKeepsStorageBlock(t *testing.T) { + cr := testCluster() + cr.Spec.Retention.MaxAge = &metav1.Duration{Duration: time.Hour} + + storage := renderStorage(t, cr) + require.Equal(t, "file", storage["backend"]) + require.Equal(t, defaultDataDir, storage["dir"]) + require.Contains(t, storage, "cluster") +} + +func TestRenderPolicyExtraConfigMergesSiblings(t *testing.T) { + cr := testCluster() + cr.Spec.Retention.MaxAge = &metav1.Duration{Duration: 24 * time.Hour} + cr.Spec.ExtraConfig = &runtime.RawExtension{ + Raw: []byte(`{"storage":{"policy":{"recompress":{"after":"72h","level":19}}}}`), + } + + policy, ok := renderStorage(t, cr)["policy"].(map[string]any) + require.True(t, ok, "policy block missing") + require.Equal(t, map[string]any{"max_age": "24h0m0s"}, policy["retention"], + "extraConfig must not displace the generated retention") + require.Equal(t, map[string]any{"after": "72h", "level": float64(19)}, policy["recompress"]) +} + +func TestValidatePolicy(t *testing.T) { + tests := []struct { + name string + retention dbv1alpha1.RetentionSpec + limits dbv1alpha1.LimitsSpec + wantErr string + }{ + { + name: "empty is valid", + }, + { + name: "zero max age is retain forever", + retention: dbv1alpha1.RetentionSpec{MaxAge: &metav1.Duration{}}, + }, + { + name: "negative max age", + retention: dbv1alpha1.RetentionSpec{MaxAge: &metav1.Duration{Duration: -time.Hour}}, + wantErr: "spec.retention.maxAge must not be negative", + }, + { + name: "negative max bytes", + retention: dbv1alpha1.RetentionSpec{MaxBytes: ptr.To(resource.MustParse("-1Gi"))}, + wantErr: "spec.retention.maxBytes must not be negative", + }, + { + name: "negative ingest rate", + limits: dbv1alpha1.LimitsSpec{IngestBytesPerSecond: ptr.To(resource.MustParse("-1"))}, + wantErr: "spec.limits.ingestBytesPerSecond must not be negative", + }, + { + name: "negative max part size", + limits: dbv1alpha1.LimitsSpec{MaxPartSize: ptr.To(resource.MustParse("-256Mi"))}, + wantErr: "spec.limits.maxPartSize must not be negative", + }, + { + name: "soft budget without hard ceiling", + limits: dbv1alpha1.LimitsSpec{MaxSeriesSoft: ptr.To[int64](1000)}, + wantErr: "spec.limits.maxSeriesSoft needs spec.limits.maxSeries", + }, + { + name: "soft budget above hard ceiling", + limits: dbv1alpha1.LimitsSpec{ + MaxSeries: ptr.To[int64](1000), + MaxSeriesSoft: ptr.To[int64](2000), + }, + wantErr: "must not exceed spec.limits.maxSeries", + }, + { + name: "soft budget equal to hard ceiling", + limits: dbv1alpha1.LimitsSpec{ + MaxSeries: ptr.To[int64](1000), + MaxSeriesSoft: ptr.To[int64](1000), + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cr := testCluster() + cr.Spec.Retention = tt.retention + cr.Spec.Limits = tt.limits + + err := validatePolicy(cr) + if tt.wantErr == "" { + require.NoError(t, err) + return + } + require.ErrorContains(t, err, tt.wantErr) + + // A bad policy is a spec problem, so it must not be requeued. + var invalid validationError + require.ErrorAs(t, err, &invalid) + + // renderConfig rejects it too, rather than emitting a config oteldb would refuse. + _, err = renderConfig(cr, cr.Spec.Etcd.Endpoints) + require.ErrorContains(t, err, tt.wantErr) + }) + } +} + +func TestValidateExtraConfigReservedPolicyPaths(t *testing.T) { + tests := []struct { + name string + extra map[string]any + wantErr string + }{ + { + name: "retention is reserved", + extra: map[string]any{"storage": map[string]any{ + "policy": map[string]any{"retention": map[string]any{"max_age": "1h"}}, + }}, + wantErr: "storage.policy.retention (use spec.retention)", + }, + { + name: "limits is reserved", + extra: map[string]any{"storage": map[string]any{ + "policy": map[string]any{"limits": map[string]any{"max_series": 10}}, + }}, + wantErr: "storage.policy.limits (use spec.limits)", + }, + { + name: "a key below a reserved path is reserved too", + extra: map[string]any{"storage": map[string]any{ + "policy": map[string]any{"retention": map[string]any{ + "max_bytes": map[string]any{"nested": true}, + }}, + }}, + wantErr: "storage.policy.retention", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := validateExtraConfig(tt.extra) + require.ErrorContains(t, err, tt.wantErr) + + var invalid validationError + require.True(t, errors.As(err, &invalid), "must be reported as a spec validation error") + }) + } +} + +// The siblings of the reserved policy keys stay open, so the CRD's coverage of retention/limits +// does not lock users out of the rest of storage.policy. +func TestValidateExtraConfigPolicySiblingsAllowed(t *testing.T) { + for _, key := range []string{"precision", "downsample", "recompress"} { + t.Run(key, func(t *testing.T) { + require.NoError(t, validateExtraConfig(map[string]any{ + "storage": map[string]any{"policy": map[string]any{key: "whatever"}}, + })) + }) + } +} From c203b26977feeaff63ebd0aeab2a78e6443cddbb Mon Sep 17 00:00:00 2001 From: tdakkota Date: Tue, 28 Jul 2026 14:49:17 +0300 Subject: [PATCH 2/2] chore(crd): regenerate manifests for retention and limits Co-Authored-By: Claude Opus 5 (1M context) --- api/v1alpha1/zz_generated.deepcopy.go | 67 +++++++++++++++++ .../bases/db.oteldb.io_oteldbclusters.yaml | 72 ++++++++++++++++++- 2 files changed, 137 insertions(+), 2 deletions(-) diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go index 39b227c..96757a4 100644 --- a/api/v1alpha1/zz_generated.deepcopy.go +++ b/api/v1alpha1/zz_generated.deepcopy.go @@ -111,6 +111,46 @@ func (in *EtcdSpec) DeepCopy() *EtcdSpec { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *LimitsSpec) DeepCopyInto(out *LimitsSpec) { + *out = *in + if in.IngestBytesPerSecond != nil { + in, out := &in.IngestBytesPerSecond, &out.IngestBytesPerSecond + x := (*in).DeepCopy() + *out = &x + } + if in.MaxInFlightBytes != nil { + in, out := &in.MaxInFlightBytes, &out.MaxInFlightBytes + x := (*in).DeepCopy() + *out = &x + } + if in.MaxSeries != nil { + in, out := &in.MaxSeries, &out.MaxSeries + *out = new(int64) + **out = **in + } + if in.MaxSeriesSoft != nil { + in, out := &in.MaxSeriesSoft, &out.MaxSeriesSoft + *out = new(int64) + **out = **in + } + if in.MaxPartSize != nil { + in, out := &in.MaxPartSize, &out.MaxPartSize + x := (*in).DeepCopy() + *out = &x + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new LimitsSpec. +func (in *LimitsSpec) DeepCopy() *LimitsSpec { + if in == nil { + return nil + } + out := new(LimitsSpec) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *OtelDBCluster) DeepCopyInto(out *OtelDBCluster) { *out = *in @@ -188,6 +228,8 @@ func (in *OtelDBClusterSpec) DeepCopyInto(out *OtelDBClusterSpec) { in.Cluster.DeepCopyInto(&out.Cluster) in.Signals.DeepCopyInto(&out.Signals) in.Engine.DeepCopyInto(&out.Engine) + in.Retention.DeepCopyInto(&out.Retention) + in.Limits.DeepCopyInto(&out.Limits) in.Service.DeepCopyInto(&out.Service) in.Resources.DeepCopyInto(&out.Resources) if in.PodAnnotations != nil { @@ -284,6 +326,31 @@ func (in *OtelDBClusterStatus) DeepCopy() *OtelDBClusterStatus { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *RetentionSpec) DeepCopyInto(out *RetentionSpec) { + *out = *in + if in.MaxAge != nil { + in, out := &in.MaxAge, &out.MaxAge + *out = new(metav1.Duration) + **out = **in + } + if in.MaxBytes != nil { + in, out := &in.MaxBytes, &out.MaxBytes + x := (*in).DeepCopy() + *out = &x + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new RetentionSpec. +func (in *RetentionSpec) DeepCopy() *RetentionSpec { + if in == nil { + return nil + } + out := new(RetentionSpec) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *S3CredentialsSecret) DeepCopyInto(out *S3CredentialsSecret) { *out = *in diff --git a/config/crd/bases/db.oteldb.io_oteldbclusters.yaml b/config/crd/bases/db.oteldb.io_oteldbclusters.yaml index f8e2cbd..07cc56e 100644 --- a/config/crd/bases/db.oteldb.io_oteldbclusters.yaml +++ b/config/crd/bases/db.oteldb.io_oteldbclusters.yaml @@ -1078,8 +1078,10 @@ spec: instead of being merged: metrics_backend, traces_backend, logs_backend, profiles_backend, storage.backend, storage.dir, storage.wal_dir, storage.s3, storage.cluster (and everything below it), storage.flush_interval, storage.read_cache_bytes, storage.decode_cache_bytes, - storage.decode_memory_bytes and storage.aggregate_stats. Configure those through - spec.storage, spec.cluster, spec.etcd, spec.signals and spec.engine. + storage.decode_memory_bytes, storage.aggregate_stats, storage.policy.retention and + storage.policy.limits. Configure those through spec.storage, spec.cluster, spec.etcd, + spec.signals, spec.engine, spec.retention and spec.limits. The rest of storage.policy + (precision, downsample, recompress) stays mergeable. type: object x-kubernetes-preserve-unknown-fields: true image: @@ -1113,6 +1115,52 @@ spec: type: object x-kubernetes-map-type: atomic type: array + limits: + description: Limits are the per-node admission-control limits. Empty + means unlimited. + properties: + ingestBytesPerSecond: + anyOf: + - type: integer + - type: string + description: IngestBytesPerSecond caps the ingest rate, bursting + to one second of budget. + pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$ + x-kubernetes-int-or-string: true + maxInFlightBytes: + anyOf: + - type: integer + - type: string + description: MaxInFlightBytes caps the unflushed in-flight bytes + buffered before backpressure sheds. + pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$ + x-kubernetes-int-or-string: true + maxPartSize: + anyOf: + - type: integer + - type: string + description: |- + MaxPartSize caps an immutable part's approximate uncompressed size; flush and merge split + their output to respect it. It is structural: fixed when a node's engine is first created, + so changing it does not affect existing data. + pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$ + x-kubernetes-int-or-string: true + maxSeries: + description: |- + MaxSeries is the hard active-series ceiling: a sample minting a new series past it is shed. + Existing series are unaffected. + format: int64 + minimum: 0 + type: integer + maxSeriesSoft: + description: |- + MaxSeriesSoft is a soft cardinality budget (metrics only): past it a new series' samples go + to a synthetic per-metric overflow series instead of being shed, until MaxSeries is reached. + It must not exceed MaxSeries, and needs MaxSeries set to have any effect. + format: int64 + minimum: 0 + type: integer + type: object logLevel: description: LogLevel sets OTEL_LOG_LEVEL for the oteldb process (e.g. DEBUG, INFO, WARN, ERROR). @@ -1433,6 +1481,26 @@ spec: More info: https://kubernetes.io/docs/concepts/configuration/manage-resources-containers/ type: object type: object + retention: + description: Retention bounds how long ingested data is kept. Empty + retains forever. + properties: + maxAge: + description: MaxAge is the maximum age of retained data (e.g. + "720h"). Empty retains forever. + type: string + maxBytes: + anyOf: + - type: integer + - type: string + description: |- + MaxBytes is the total retained-bytes budget across every signal on a node. + + oteldb accepts it, but the storage engine does not enforce it yet (oteldb/storage#224), so + setting it alone bounds nothing today. Use MaxAge to bound disk growth. + pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$ + x-kubernetes-int-or-string: true + type: object securityContext: description: SecurityContext for the oteldb container. properties: