diff --git a/README.md b/README.md index 232fe90..d24a1f4 100644 --- a/README.md +++ b/README.md @@ -76,14 +76,18 @@ for a fuller example including the S3 backend. | `policy.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. | | `policy.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. | | `policy.limits` | Per-node admission control: `ingestBytesPerSecond`, `maxInFlightBytes`, `maxSeries`, `maxSeriesSoft`, `maxPartSize`. Over-budget writes are shed as OTLP partial success rather than buffered. | +| `policy.downsample[]` | Merge-time age-tiered rollup: `{after, interval, agg}`. Samples past `after` collapse to one per `interval` bucket. **Lossy and irreversible.** | +| `policy.precision[]` | Age-tiered lossy float precision: `{after, bits}`. Parts past `after` keep only `bits` mantissa bits. **Lossy and irreversible.** | +| `policy.recompress` | `{after, level}`. Rewrites fully-cold parts with a higher-ratio Zstandard profile. Decode-transparent and lossless. | | `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). | +| `extraConfig` | Arbitrary raw oteldb config **deep-merged** over the generated config — for fields the CRD does not model (auth, prometheus tuning, …). Nested objects merge key by key (`storage.policy` does not wipe `storage.backend`); operator-owned paths are [reserved](#reserved-extraconfig-paths). | -> `spec.policy` maps onto oteldb's `storage.policy`, which gained `retention` and `limits` after -> **v0.48.0**. Older oteldb builds — including the operator's current default image — ignore -> unknown config keys silently, so on those the policy is accepted by the API server and has no -> effect. Pin a newer `spec.image` before relying on it. +> `spec.policy` maps onto oteldb's `storage.policy`. `downsample`, `precision` and `recompress` +> work against oteldb v0.48.0; `retention` and `limits` landed upstream **after** it. Older oteldb +> builds — including the operator's current default image — ignore unknown config keys silently, +> so on those the two newer blocks are accepted by the API server and have no effect. Pin a newer +> `spec.image` before relying on them. ### Reserved `extraConfig` paths @@ -91,9 +95,8 @@ for a fuller example including the S3 backend. ```yaml extraConfig: - storage: - policy: - recompress: {after: 3d, level: 19} # keeps backend/dir/cluster + auth: + tenant_header: X-Scope-OrgID # merged in; keeps backend/dir/cluster ``` The paths the operator renders from the spec are **reserved**: an `extraConfig` that sets one is @@ -109,11 +112,9 @@ 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.policy.retention` | -| `storage.policy.limits` | `spec.policy.limits` | +| `storage.policy.retention`, `storage.policy.limits`, `storage.policy.downsample`, `storage.policy.precision`, `storage.policy.recompress` | `spec.policy` | -The rest of `storage.policy` — `precision`, `downsample`, `recompress` — is not modelled by the -CRD and stays mergeable, as in the example above. +`storage.policy` is now modelled in full, so the whole block is reserved. ### Status diff --git a/api/v1alpha1/oteldbcluster_types.go b/api/v1alpha1/oteldbcluster_types.go index 660cf1a..ec72423 100644 --- a/api/v1alpha1/oteldbcluster_types.go +++ b/api/v1alpha1/oteldbcluster_types.go @@ -78,9 +78,9 @@ type OtelDBClusterSpec struct { // +optional Engine EngineSpec `json:"engine,omitempty"` - // Policy is the per-tenant storage policy: retention and admission-control limits. It maps - // onto oteldb's storage.policy block. Empty leaves the engine at its defaults (retain - // forever, no limits). + // Policy is the per-tenant storage policy: retention, admission-control limits, and the + // merge-time downsample/precision/recompress tiers. It maps onto oteldb's storage.policy + // block. Empty leaves the engine at its defaults (retain forever, no limits, lossless, raw). // +optional Policy PolicySpec `json:"policy,omitempty"` @@ -133,7 +133,7 @@ type OtelDBClusterSpec struct { // ExtraConfig is arbitrary additional oteldb config deeply merged over the generated config, as // a top-level YAML/JSON object. Use it to set fields the CRD does not model directly (auth, - // prometheus tuning, retention policy, ...). Nested objects are merged key by key, so + // prometheus tuning, ...). Nested objects are merged key by key, so // storage.policy can be added without discarding the generated storage block; any other value // overrides the generated one. // @@ -141,10 +141,9 @@ 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, storage.aggregate_stats, storage.policy.retention and - // storage.policy.limits. Configure those through spec.storage, spec.cluster, spec.etcd, - // spec.signals, spec.engine and spec.policy. The rest of storage.policy - // (precision, downsample, recompress) stays mergeable. + // storage.decode_memory_bytes, storage.aggregate_stats and storage.policy (modelled in full, so + // the whole block is reserved). Configure those through spec.storage, spec.cluster, spec.etcd, + // spec.signals, spec.engine and spec.policy. // +optional // +kubebuilder:pruning:PreserveUnknownFields ExtraConfig *runtime.RawExtension `json:"extraConfig,omitempty"` @@ -334,12 +333,9 @@ type EngineSpec struct { // PolicySpec is the per-tenant storage policy, mapping 1:1 onto oteldb's storage.policy block. // -// Retention and limits landed in oteldb's config after v0.48.0. Older builds ignore unknown config -// keys silently, so against those this policy is accepted and does nothing — pin a newer Image -// before relying on it. -// -// The merge-time policies oteldb also supports there — precision, downsample and recompress — are -// not modelled yet (see oteldb/operator#3); they stay reachable through spec.extraConfig. +// Downsample, Precision and Recompress work against oteldb v0.48.0. Retention and Limits landed in +// oteldb's config after it; older builds ignore unknown config keys silently, so against those +// those two are accepted and do nothing — pin a newer Image before relying on them. type PolicySpec struct { // Retention bounds how long ingested data is kept. Empty retains forever. // +optional @@ -348,6 +344,80 @@ type PolicySpec struct { // Limits are the per-node admission-control limits. Empty means unlimited. // +optional Limits LimitsSpec `json:"limits,omitempty"` + + // Downsample is the age-tiered merge-time rollup: samples older than a tier's After are + // replaced by one representative per Interval-wide bucket. Empty keeps data raw. + // + // This rewrites data in place and cannot be undone: lowering a tier's After re-processes + // existing parts at the next merge, and the replaced samples are gone. + // +optional + // +listType=atomic + Downsample []DownsampleTierSpec `json:"downsample,omitempty"` + + // Precision is the age-tiered lossy float-compression policy: the value column of parts older + // than a tier's After is re-encoded to keep only Bits mantissa bits. Empty stays lossless. + // + // This rewrites data in place and cannot be undone: discarded mantissa bits are not + // recoverable, and lowering a tier's After re-processes existing parts at the next merge. + // +optional + // +listType=atomic + Precision []PrecisionTierSpec `json:"precision,omitempty"` + + // Recompress rewrites fully-cold parts with a higher-ratio Zstandard profile at merge, trading + // merge CPU for storage. It is decode-transparent and lossless. Nil disables it. + // +optional + Recompress *RecompressSpec `json:"recompress,omitempty"` +} + +// DownsampleTierSpec is one age band of the downsampling policy. Tiers are order-independent: a +// sample is rolled up by the coarsest tier whose After it has exceeded, and samples younger than +// every tier stay raw. Buckets align to absolute multiples of Interval, so repeated merges are +// stable. +type DownsampleTierSpec struct { + // After is the age past which this tier applies, relative to merge time (e.g. "24h"). + // +required + After metav1.Duration `json:"after"` + + // Interval is the rollup bucket width (e.g. "5m"). It must be positive. + // +required + Interval metav1.Duration `json:"interval"` + + // Agg combines the samples in a bucket. Defaults to "last". + // +kubebuilder:validation:Enum=last;first;min;max;sum;avg;count + // +optional + Agg string `json:"agg,omitempty"` +} + +// PrecisionTierSpec is one age band of the lossy float-precision policy. Tiers are +// order-independent: a part takes the most aggressive tier whose After it has exceeded. The +// encoder keeps whichever of the lossy and lossless encodings is smaller, so a tier can only help +// size. +type PrecisionTierSpec struct { + // After is the age past which this tier applies, relative to merge time (e.g. "168h"). + // +required + After metav1.Duration `json:"after"` + + // Bits is the number of significant mantissa bits retained. Fewer bits compress better and + // lose more accuracy. + // +kubebuilder:validation:Minimum=1 + // +kubebuilder:validation:Maximum=63 + // +required + Bits int32 `json:"bits"` +} + +// RecompressSpec configures cold-part recompression. +type RecompressSpec struct { + // After is the age past which a fully-cold part is recompressed at merge. It must be positive + // — the block exists only to enable recompression. + // +required + After metav1.Duration `json:"after"` + + // Level is the Zstandard level: 1 is fastest, 19 is the best ratio. Empty uses the best-ratio + // default. + // +kubebuilder:validation:Minimum=1 + // +kubebuilder:validation:Maximum=19 + // +optional + Level *int32 `json:"level,omitempty"` } // RetentionSpec bounds how long data is kept. Enforcement happens at merge time and drops whole diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go index 0d98095..a50e399 100644 --- a/api/v1alpha1/zz_generated.deepcopy.go +++ b/api/v1alpha1/zz_generated.deepcopy.go @@ -51,6 +51,23 @@ func (in *ClusterSpec) DeepCopy() *ClusterSpec { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *DownsampleTierSpec) DeepCopyInto(out *DownsampleTierSpec) { + *out = *in + out.After = in.After + out.Interval = in.Interval +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new DownsampleTierSpec. +func (in *DownsampleTierSpec) DeepCopy() *DownsampleTierSpec { + if in == nil { + return nil + } + out := new(DownsampleTierSpec) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *EngineSpec) DeepCopyInto(out *EngineSpec) { *out = *in @@ -330,6 +347,21 @@ func (in *PolicySpec) DeepCopyInto(out *PolicySpec) { *out = *in in.Retention.DeepCopyInto(&out.Retention) in.Limits.DeepCopyInto(&out.Limits) + if in.Downsample != nil { + in, out := &in.Downsample, &out.Downsample + *out = make([]DownsampleTierSpec, len(*in)) + copy(*out, *in) + } + if in.Precision != nil { + in, out := &in.Precision, &out.Precision + *out = make([]PrecisionTierSpec, len(*in)) + copy(*out, *in) + } + if in.Recompress != nil { + in, out := &in.Recompress, &out.Recompress + *out = new(RecompressSpec) + (*in).DeepCopyInto(*out) + } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new PolicySpec. @@ -342,6 +374,43 @@ func (in *PolicySpec) DeepCopy() *PolicySpec { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *PrecisionTierSpec) DeepCopyInto(out *PrecisionTierSpec) { + *out = *in + out.After = in.After +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new PrecisionTierSpec. +func (in *PrecisionTierSpec) DeepCopy() *PrecisionTierSpec { + if in == nil { + return nil + } + out := new(PrecisionTierSpec) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *RecompressSpec) DeepCopyInto(out *RecompressSpec) { + *out = *in + out.After = in.After + if in.Level != nil { + in, out := &in.Level, &out.Level + *out = new(int32) + **out = **in + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new RecompressSpec. +func (in *RecompressSpec) DeepCopy() *RecompressSpec { + if in == nil { + return nil + } + out := new(RecompressSpec) + in.DeepCopyInto(out) + 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 diff --git a/config/crd/bases/db.oteldb.io_oteldbclusters.yaml b/config/crd/bases/db.oteldb.io_oteldbclusters.yaml index fbb3085..f32e5a6 100644 --- a/config/crd/bases/db.oteldb.io_oteldbclusters.yaml +++ b/config/crd/bases/db.oteldb.io_oteldbclusters.yaml @@ -1070,7 +1070,7 @@ spec: description: |- ExtraConfig is arbitrary additional oteldb config deeply merged over the generated config, as a top-level YAML/JSON object. Use it to set fields the CRD does not model directly (auth, - prometheus tuning, retention policy, ...). Nested objects are merged key by key, so + prometheus tuning, ...). Nested objects are merged key by key, so storage.policy can be added without discarding the generated storage block; any other value overrides the generated one. @@ -1078,10 +1078,9 @@ 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, storage.aggregate_stats, storage.policy.retention and - storage.policy.limits. Configure those through spec.storage, spec.cluster, spec.etcd, - spec.signals, spec.engine and spec.policy. The rest of storage.policy - (precision, downsample, recompress) stays mergeable. + storage.decode_memory_bytes, storage.aggregate_stats and storage.policy (modelled in full, so + the whole block is reserved). Configure those through spec.storage, spec.cluster, spec.etcd, + spec.signals, spec.engine and spec.policy. type: object x-kubernetes-preserve-unknown-fields: true image: @@ -1369,10 +1368,50 @@ spec: type: object policy: description: |- - Policy is the per-tenant storage policy: retention and admission-control limits. It maps - onto oteldb's storage.policy block. Empty leaves the engine at its defaults (retain - forever, no limits). + Policy is the per-tenant storage policy: retention, admission-control limits, and the + merge-time downsample/precision/recompress tiers. It maps onto oteldb's storage.policy + block. Empty leaves the engine at its defaults (retain forever, no limits, lossless, raw). properties: + downsample: + description: |- + Downsample is the age-tiered merge-time rollup: samples older than a tier's After are + replaced by one representative per Interval-wide bucket. Empty keeps data raw. + + This rewrites data in place and cannot be undone: lowering a tier's After re-processes + existing parts at the next merge, and the replaced samples are gone. + items: + description: |- + DownsampleTierSpec is one age band of the downsampling policy. Tiers are order-independent: a + sample is rolled up by the coarsest tier whose After it has exceeded, and samples younger than + every tier stay raw. Buckets align to absolute multiples of Interval, so repeated merges are + stable. + properties: + after: + description: After is the age past which this tier applies, + relative to merge time (e.g. "24h"). + type: string + agg: + description: Agg combines the samples in a bucket. Defaults + to "last". + enum: + - last + - first + - min + - max + - sum + - avg + - count + type: string + interval: + description: Interval is the rollup bucket width (e.g. "5m"). + It must be positive. + type: string + required: + - after + - interval + type: object + type: array + x-kubernetes-list-type: atomic limits: description: Limits are the per-node admission-control limits. Empty means unlimited. @@ -1419,6 +1458,59 @@ spec: minimum: 0 type: integer type: object + precision: + description: |- + Precision is the age-tiered lossy float-compression policy: the value column of parts older + than a tier's After is re-encoded to keep only Bits mantissa bits. Empty stays lossless. + + This rewrites data in place and cannot be undone: discarded mantissa bits are not + recoverable, and lowering a tier's After re-processes existing parts at the next merge. + items: + description: |- + PrecisionTierSpec is one age band of the lossy float-precision policy. Tiers are + order-independent: a part takes the most aggressive tier whose After it has exceeded. The + encoder keeps whichever of the lossy and lossless encodings is smaller, so a tier can only help + size. + properties: + after: + description: After is the age past which this tier applies, + relative to merge time (e.g. "168h"). + type: string + bits: + description: |- + Bits is the number of significant mantissa bits retained. Fewer bits compress better and + lose more accuracy. + format: int32 + maximum: 63 + minimum: 1 + type: integer + required: + - after + - bits + type: object + type: array + x-kubernetes-list-type: atomic + recompress: + description: |- + Recompress rewrites fully-cold parts with a higher-ratio Zstandard profile at merge, trading + merge CPU for storage. It is decode-transparent and lossless. Nil disables it. + properties: + after: + description: |- + After is the age past which a fully-cold part is recompressed at merge. It must be positive + — the block exists only to enable recompression. + type: string + level: + description: |- + Level is the Zstandard level: 1 is fastest, 19 is the best ratio. Empty uses the best-ratio + default. + format: int32 + maximum: 19 + minimum: 1 + type: integer + required: + - after + type: object retention: description: Retention bounds how long ingested data is kept. Empty retains forever. diff --git a/config/samples/db_v1alpha1_oteldbcluster.yaml b/config/samples/db_v1alpha1_oteldbcluster.yaml index 412c8ae..ce4635f 100644 --- a/config/samples/db_v1alpha1_oteldbcluster.yaml +++ b/config/samples/db_v1alpha1_oteldbcluster.yaml @@ -53,6 +53,20 @@ spec: maxInFlightBytes: 1Gi maxPartSize: 256Mi + # Merge-time coarsening of old data. These rewrite parts in place and CANNOT be undone — + # lowering an `after` re-processes existing data at the next merge. + downsample: + - {after: 24h, interval: 5m} # samples older than a day: one point per 5m bucket + - {after: 168h, interval: 1h, agg: avg} + precision: + - {after: 168h, bits: 32} # a week old: half the mantissa + - {after: 336h, bits: 16} + + # Lossless and decode-transparent: costs merge CPU, saves storage. + recompress: + after: 72h + level: 19 + resources: requests: cpu: "1" diff --git a/internal/controller/builders_test.go b/internal/controller/builders_test.go index 157d5cc..9e4a754 100644 --- a/internal/controller/builders_test.go +++ b/internal/controller/builders_test.go @@ -162,7 +162,7 @@ func TestRenderConfigExtraConfigPreservesStorageBlock(t *testing.T) { cr := testCluster() cr.Spec.Cluster.ReplicationFactor = ptr.To[int32](3) cr.Spec.ExtraConfig = &runtime.RawExtension{ - Raw: []byte(`{"storage":{"policy":{"recompress":{"after":"3d","level":19}}}}`), + Raw: []byte(`{"storage":{"log_query_parallelism":4}}`), } out, err := renderConfig(cr, cr.Spec.Etcd.Endpoints) require.NoError(t, err) @@ -179,9 +179,7 @@ func TestRenderConfigExtraConfigPreservesStorageBlock(t *testing.T) { require.Equal(t, []any{"http://etcd:2379"}, cluster["etcd"]) require.EqualValues(t, 3, cluster["rf"]) - policy, ok := storage["policy"].(map[string]any) - require.True(t, ok, "extraConfig policy not merged") - require.Equal(t, map[string]any{"after": "3d", "level": float64(19)}, policy["recompress"]) + require.EqualValues(t, 4, storage["log_query_parallelism"], "extraConfig key not merged") } func TestRenderConfigExtraConfigReservedPath(t *testing.T) { diff --git a/internal/controller/extraconfig.go b/internal/controller/extraconfig.go index 014b963..b94f3f8 100644 --- a/internal/controller/extraconfig.go +++ b/internal/controller/extraconfig.go @@ -47,9 +47,11 @@ var reservedConfigPaths = map[string]string{ "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.policy.retention", - "storage.policy.limits": "use spec.policy.limits", + "storage.policy.retention": "use spec.policy.retention", + "storage.policy.limits": "use spec.policy.limits", + "storage.policy.downsample": "use spec.policy.downsample", + "storage.policy.precision": "use spec.policy.precision", + "storage.policy.recompress": "use spec.policy.recompress", } // validationError marks a spec problem that no amount of retrying can fix: the reconcile is diff --git a/internal/controller/extraconfig_test.go b/internal/controller/extraconfig_test.go index 0e2964d..c4039f3 100644 --- a/internal/controller/extraconfig_test.go +++ b/internal/controller/extraconfig_test.go @@ -112,7 +112,7 @@ func TestValidateExtraConfig(t *testing.T) { }, { name: "non-reserved storage key", - extra: map[string]any{"storage": map[string]any{"policy": map[string]any{"recompress": "3d"}}}, + extra: map[string]any{"storage": map[string]any{"log_query_parallelism": 4}}, }, { name: "storage backend", diff --git a/internal/controller/policy.go b/internal/controller/policy.go index 1991ab1..1123f9b 100644 --- a/internal/controller/policy.go +++ b/internal/controller/policy.go @@ -17,6 +17,8 @@ limitations under the License. package controller import ( + "fmt" + "k8s.io/apimachinery/pkg/api/resource" dbv1alpha1 "github.com/oteldb/operator/api/v1alpha1" @@ -24,23 +26,42 @@ import ( // oteldb storage.policy config keys. const ( - keyPolicy = "policy" - keyRetention = "retention" - keyLimits = "limits" + keyPolicy = "policy" + keyRetention = "retention" + keyLimits = "limits" + keyDownsample = "downsample" + keyPrecision = "precision" + keyRecompress = "recompress" + keyAfter = "after" + keyInterval = "interval" ) -// 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). +// renderPolicy builds the storage.policy block from spec.policy, or returns nil when nothing is +// configured — oteldb installs no tenancy resolver for an absent policy, which is the library +// default (retain forever, no limits, lossless, no rollup). func renderPolicy(cr *dbv1alpha1.OtelDBCluster) map[string]any { + spec := cr.Spec.Policy policy := map[string]any{} - if retention := renderRetention(cr.Spec.Policy.Retention); len(retention) > 0 { + if retention := renderRetention(spec.Retention); len(retention) > 0 { policy[keyRetention] = retention } - if limits := renderLimits(cr.Spec.Policy.Limits); len(limits) > 0 { + if limits := renderLimits(spec.Limits); len(limits) > 0 { policy[keyLimits] = limits } + if tiers := renderDownsample(spec.Downsample); len(tiers) > 0 { + policy[keyDownsample] = tiers + } + if tiers := renderPrecision(spec.Precision); len(tiers) > 0 { + policy[keyPrecision] = tiers + } + if r := spec.Recompress; r != nil { + m := map[string]any{keyAfter: r.After.Duration.String()} + if r.Level != nil { + m["level"] = *r.Level + } + policy[keyRecompress] = m + } if len(policy) == 0 { return nil @@ -48,6 +69,32 @@ func renderPolicy(cr *dbv1alpha1.OtelDBCluster) map[string]any { return policy } +func renderDownsample(tiers []dbv1alpha1.DownsampleTierSpec) []any { + out := make([]any, 0, len(tiers)) + for _, t := range tiers { + m := map[string]any{ + keyAfter: t.After.Duration.String(), + keyInterval: t.Interval.Duration.String(), + } + if t.Agg != "" { + m["agg"] = t.Agg + } + out = append(out, m) + } + return out +} + +func renderPrecision(tiers []dbv1alpha1.PrecisionTierSpec) []any { + out := make([]any, 0, len(tiers)) + for _, t := range tiers { + out = append(out, map[string]any{ + keyAfter: t.After.Duration.String(), + "bits": t.Bits, + }) + } + return out +} + func renderRetention(spec dbv1alpha1.RetentionSpec) map[string]any { m := map[string]any{} if spec.MaxAge != nil { @@ -113,5 +160,70 @@ func validatePolicy(cr *dbv1alpha1.OtelDBCluster) error { *limits.MaxSeriesSoft, *limits.MaxSeries) } } + + return validateMergeTiers(cr.Spec.Policy) +} + +// validateMergeTiers rejects downsample/precision/recompress settings the engine would ignore. The +// tiers are lossy and irreversible, so a tier that silently does nothing is worth failing over: the +// user believes their old data is being coarsened when it is not. +func validateMergeTiers(spec dbv1alpha1.PolicySpec) error { + seen := map[string]int{} + for i, t := range spec.Downsample { + field := fmt.Sprintf("spec.policy.downsample[%d]", i) + if t.After.Duration < 0 { + return invalidSpec("%s.after must not be negative, got %s", field, t.After.Duration) + } + if t.Interval.Duration <= 0 { + return invalidSpec("%s.interval must be positive, got %s", field, t.Interval.Duration) + } + if prev, dup := seen[t.After.Duration.String()]; dup { + return invalidSpec("%s.after duplicates spec.policy.downsample[%d].after (%s): "+ + "a sample takes one tier, so the other is dead", field, prev, t.After.Duration) + } + seen[t.After.Duration.String()] = i + } + + seen = map[string]int{} + for i, t := range spec.Precision { + field := fmt.Sprintf("spec.policy.precision[%d]", i) + if t.After.Duration < 0 { + return invalidSpec("%s.after must not be negative, got %s", field, t.After.Duration) + } + if prev, dup := seen[t.After.Duration.String()]; dup { + return invalidSpec("%s.after duplicates spec.policy.precision[%d].after (%s): "+ + "a part takes one tier, so the other is dead", field, prev, t.After.Duration) + } + seen[t.After.Duration.String()] = i + } + + if r := spec.Recompress; r != nil && r.After.Duration <= 0 { + return invalidSpec("spec.policy.recompress.after must be positive, got %s; "+ + "remove the recompress block to disable it", r.After.Duration) + } + + // Coarsening past the retention window is merge work whose output is dropped before it can be + // read. Zero maxAge is "retain forever", so only a real window is checked. + if maxAge := spec.Retention.MaxAge; maxAge != nil && maxAge.Duration > 0 { + for i, t := range spec.Downsample { + if t.After.Duration >= maxAge.Duration { + return invalidSpec("spec.policy.downsample[%d].after (%s) is at or past "+ + "spec.policy.retention.maxAge (%s): the tier never applies before the data is dropped", + i, t.After.Duration, maxAge.Duration) + } + } + for i, t := range spec.Precision { + if t.After.Duration >= maxAge.Duration { + return invalidSpec("spec.policy.precision[%d].after (%s) is at or past "+ + "spec.policy.retention.maxAge (%s): the tier never applies before the data is dropped", + i, t.After.Duration, maxAge.Duration) + } + } + if r := spec.Recompress; r != nil && r.After.Duration >= maxAge.Duration { + return invalidSpec("spec.policy.recompress.after (%s) is at or past "+ + "spec.policy.retention.maxAge (%s): parts are dropped before they are recompressed", + r.After.Duration, maxAge.Duration) + } + } return nil } diff --git a/internal/controller/policy_test.go b/internal/controller/policy_test.go index 44a4214..c2805ca 100644 --- a/internal/controller/policy_test.go +++ b/internal/controller/policy_test.go @@ -100,18 +100,21 @@ func TestRenderPolicyKeepsStorageBlock(t *testing.T) { require.Contains(t, storage, "cluster") } -func TestRenderPolicyExtraConfigMergesSiblings(t *testing.T) { +// storage.policy is reserved in full, but its siblings under storage are not: an extraConfig key +// the CRD does not model still merges alongside a generated policy. +func TestRenderPolicyExtraConfigMergesStorageSiblings(t *testing.T) { cr := testCluster() cr.Spec.Policy.Retention.MaxAge = &metav1.Duration{Duration: 24 * time.Hour} cr.Spec.ExtraConfig = &runtime.RawExtension{ - Raw: []byte(`{"storage":{"policy":{"recompress":{"after":"72h","level":19}}}}`), + Raw: []byte(`{"storage":{"log_query_parallelism":4}}`), } - policy, ok := renderStorage(t, cr)["policy"].(map[string]any) + storage := renderStorage(t, cr) + require.EqualValues(t, 4, storage["log_query_parallelism"]) + + policy, ok := storage["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"]) + require.Equal(t, map[string]any{"max_age": "24h0m0s"}, policy["retention"]) } func TestValidatePolicy(t *testing.T) { @@ -238,12 +241,174 @@ func TestValidateExtraConfigReservedPolicyPaths(t *testing.T) { // 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"} { +// Every storage.policy key the CRD models is reserved, so extraConfig cannot fight spec.policy. +func TestValidateExtraConfigPolicyFullyReserved(t *testing.T) { + for _, key := range []string{"retention", "limits", "downsample", "precision", "recompress"} { t.Run(key, func(t *testing.T) { - require.NoError(t, validateExtraConfig(map[string]any{ + err := validateExtraConfig(map[string]any{ "storage": map[string]any{"policy": map[string]any{key: "whatever"}}, - })) + }) + require.ErrorContains(t, err, "storage.policy."+key+" (use spec.policy."+key+")") + }) + } +} + +func TestRenderPolicyMergeTiers(t *testing.T) { + cr := testCluster() + cr.Spec.Policy = dbv1alpha1.PolicySpec{ + Downsample: []dbv1alpha1.DownsampleTierSpec{ + {After: metav1.Duration{Duration: 6 * time.Hour}, Interval: metav1.Duration{Duration: time.Minute}}, + {After: metav1.Duration{Duration: 72 * time.Hour}, Interval: metav1.Duration{Duration: 5 * time.Minute}, Agg: "avg"}, + }, + Precision: []dbv1alpha1.PrecisionTierSpec{ + {After: metav1.Duration{Duration: 72 * time.Hour}, Bits: 32}, + }, + Recompress: &dbv1alpha1.RecompressSpec{ + After: metav1.Duration{Duration: 72 * time.Hour}, + Level: ptr.To[int32](19), + }, + } + + policy, ok := renderStorage(t, cr)["policy"].(map[string]any) + require.True(t, ok, "policy block missing") + require.Equal(t, []any{ + map[string]any{"after": "6h0m0s", "interval": "1m0s"}, + map[string]any{"after": "72h0m0s", "interval": "5m0s", "agg": "avg"}, + }, policy["downsample"], "an unset agg must be omitted so oteldb applies its own default") + require.Equal(t, []any{ + map[string]any{"after": "72h0m0s", "bits": float64(32)}, + }, policy["precision"]) + require.Equal(t, map[string]any{"after": "72h0m0s", "level": float64(19)}, policy["recompress"]) +} + +func TestRenderPolicyRecompressWithoutLevel(t *testing.T) { + cr := testCluster() + cr.Spec.Policy.Recompress = &dbv1alpha1.RecompressSpec{After: metav1.Duration{Duration: time.Hour}} + + policy := renderStorage(t, cr)["policy"].(map[string]any) + require.Equal(t, map[string]any{"after": "1h0m0s"}, policy["recompress"], + "an unset level must be omitted so oteldb picks its best-ratio default") +} + +func TestValidateMergeTiers(t *testing.T) { + hour := func(h int) metav1.Duration { return metav1.Duration{Duration: time.Duration(h) * time.Hour} } + min := func(m int) metav1.Duration { return metav1.Duration{Duration: time.Duration(m) * time.Minute} } + + tests := []struct { + name string + policy dbv1alpha1.PolicySpec + wantErr string + }{ + { + name: "empty is valid", + }, + { + name: "ordered tiers", + policy: dbv1alpha1.PolicySpec{ + Downsample: []dbv1alpha1.DownsampleTierSpec{ + {After: hour(6), Interval: min(1)}, + {After: hour(72), Interval: min(5)}, + }, + }, + }, + { + name: "tier order does not matter", + policy: dbv1alpha1.PolicySpec{ + Downsample: []dbv1alpha1.DownsampleTierSpec{ + {After: hour(72), Interval: min(5)}, + {After: hour(6), Interval: min(1)}, + }, + }, + }, + { + name: "zero downsample interval", + policy: dbv1alpha1.PolicySpec{ + Downsample: []dbv1alpha1.DownsampleTierSpec{{After: hour(6)}}, + }, + wantErr: "spec.policy.downsample[0].interval must be positive", + }, + { + name: "negative downsample after", + policy: dbv1alpha1.PolicySpec{ + Downsample: []dbv1alpha1.DownsampleTierSpec{{After: hour(-1), Interval: min(1)}}, + }, + wantErr: "spec.policy.downsample[0].after must not be negative", + }, + { + name: "duplicate downsample after", + policy: dbv1alpha1.PolicySpec{ + Downsample: []dbv1alpha1.DownsampleTierSpec{ + {After: hour(6), Interval: min(1)}, + {After: hour(6), Interval: min(5)}, + }, + }, + wantErr: "spec.policy.downsample[1].after duplicates spec.policy.downsample[0].after", + }, + { + name: "duplicate precision after", + policy: dbv1alpha1.PolicySpec{ + Precision: []dbv1alpha1.PrecisionTierSpec{ + {After: hour(72), Bits: 32}, + {After: hour(72), Bits: 16}, + }, + }, + wantErr: "spec.policy.precision[1].after duplicates spec.policy.precision[0].after", + }, + { + name: "zero recompress after", + policy: dbv1alpha1.PolicySpec{Recompress: &dbv1alpha1.RecompressSpec{}}, + wantErr: "spec.policy.recompress.after must be positive", + }, + { + name: "downsample tier past retention", + policy: dbv1alpha1.PolicySpec{ + Retention: dbv1alpha1.RetentionSpec{MaxAge: &metav1.Duration{Duration: 24 * time.Hour}}, + Downsample: []dbv1alpha1.DownsampleTierSpec{{After: hour(48), Interval: min(5)}}, + }, + wantErr: "spec.policy.downsample[0].after (48h0m0s) is at or past spec.policy.retention.maxAge", + }, + { + name: "precision tier past retention", + policy: dbv1alpha1.PolicySpec{ + Retention: dbv1alpha1.RetentionSpec{MaxAge: &metav1.Duration{Duration: 24 * time.Hour}}, + Precision: []dbv1alpha1.PrecisionTierSpec{{After: hour(24), Bits: 32}}, + }, + wantErr: "spec.policy.precision[0].after (24h0m0s) is at or past spec.policy.retention.maxAge", + }, + { + name: "recompress past retention", + policy: dbv1alpha1.PolicySpec{ + Retention: dbv1alpha1.RetentionSpec{MaxAge: &metav1.Duration{Duration: 24 * time.Hour}}, + Recompress: &dbv1alpha1.RecompressSpec{After: hour(48)}, + }, + wantErr: "spec.policy.recompress.after (48h0m0s) is at or past spec.policy.retention.maxAge", + }, + { + name: "retain forever does not bound the tiers", + policy: dbv1alpha1.PolicySpec{ + Retention: dbv1alpha1.RetentionSpec{MaxAge: &metav1.Duration{}}, + Downsample: []dbv1alpha1.DownsampleTierSpec{{After: hour(8760), Interval: min(60)}}, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cr := testCluster() + cr.Spec.Policy = tt.policy + + err := validatePolicy(cr) + if tt.wantErr == "" { + require.NoError(t, err) + return + } + require.ErrorContains(t, err, tt.wantErr) + + var invalid validationError + require.ErrorAs(t, err, &invalid, "must be reported as a spec validation error") + + _, err = renderConfig(cr, cr.Spec.Etcd.Endpoints) + require.ErrorContains(t, err, tt.wantErr) }) } }