diff --git a/README.md b/README.md index 12b15ff..76646cb 100644 --- a/README.md +++ b/README.md @@ -75,7 +75,32 @@ for a fuller example including the S3 backend. | `engine` | Storage engine tuning: `flushInterval`, `readCacheSize`, `decodeCacheSize`, `decodeMemoryLimit`, `aggregateStats`. | | `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 merged over the generated config (top-level keys win) — for fields the CRD does not model (auth, retention policy, prometheus tuning, …). | +| `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). | + +### Reserved `extraConfig` paths + +`extraConfig` is merged recursively, so it can add keys the CRD does not model: + +```yaml +extraConfig: + storage: + policy: + recompress: {after: 3d, level: 19} # keeps backend/dir/cluster +``` + +The paths the operator renders from the spec are **reserved**: an `extraConfig` that sets one is +rejected, and the CR goes `Degraded` with reason `InvalidSpec` naming the offending path and the +spec field to use instead. + +| Reserved path | Use instead | +|---|---| +| `metrics_backend`, `traces_backend`, `logs_backend` | not configurable — signals are always served from the embedded storage engine | +| `profiles_backend` | `spec.signals.profiles` | +| `storage.backend` | `spec.storage.backend` | +| `storage.dir`, `storage.wal_dir` | `spec.storage.dir` | +| `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` | ### Status diff --git a/api/v1alpha1/oteldbcluster_types.go b/api/v1alpha1/oteldbcluster_types.go index 353f4e6..cd5f219 100644 --- a/api/v1alpha1/oteldbcluster_types.go +++ b/api/v1alpha1/oteldbcluster_types.go @@ -125,9 +125,18 @@ type OtelDBClusterSpec struct { // +optional SecurityContext *corev1.SecurityContext `json:"securityContext,omitempty"` - // ExtraConfig is arbitrary additional oteldb config 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, ...). Keys here override generated ones. + // 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 + // storage.policy can be added without discarding the generated storage block; any other value + // overrides the generated one. + // + // The paths the operator owns are reserved and rejected with a Degraded/InvalidSpec condition + // 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. // +optional // +kubebuilder:pruning:PreserveUnknownFields ExtraConfig *runtime.RawExtension `json:"extraConfig,omitempty"` diff --git a/config/crd/bases/db.oteldb.io_oteldbclusters.yaml b/config/crd/bases/db.oteldb.io_oteldbclusters.yaml index 61c4ca9..f8e2cbd 100644 --- a/config/crd/bases/db.oteldb.io_oteldbclusters.yaml +++ b/config/crd/bases/db.oteldb.io_oteldbclusters.yaml @@ -1068,9 +1068,18 @@ spec: type: object extraConfig: description: |- - ExtraConfig is arbitrary additional oteldb config 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, ...). Keys here override generated ones. + 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 + storage.policy can be added without discarding the generated storage block; any other value + overrides the generated one. + + The paths the operator owns are reserved and rejected with a Degraded/InvalidSpec condition + 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. type: object x-kubernetes-preserve-unknown-fields: true image: diff --git a/go.mod b/go.mod index d422a2b..52c4616 100644 --- a/go.mod +++ b/go.mod @@ -5,6 +5,7 @@ go 1.26.0 require ( github.com/onsi/ginkgo/v2 v2.27.4 github.com/onsi/gomega v1.39.0 + github.com/stretchr/testify v1.11.1 k8s.io/api v0.36.0 k8s.io/apimachinery v0.36.0 k8s.io/client-go v0.36.0 diff --git a/internal/controller/builders_test.go b/internal/controller/builders_test.go index 7f0808e..157d5cc 100644 --- a/internal/controller/builders_test.go +++ b/internal/controller/builders_test.go @@ -17,8 +17,10 @@ limitations under the License. package controller import ( + "errors" "testing" + "github.com/stretchr/testify/require" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -154,6 +156,46 @@ func TestRenderConfigExtraConfigOverrides(t *testing.T) { } } +// Regression: a shallow merge of extraConfig used to replace the whole storage block, silently +// downgrading the backend to memory and dropping the node out of the cluster. +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}}}}`), + } + 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") + + require.Equal(t, "file", storage["backend"]) + require.Equal(t, "/var/lib/oteldb", storage["dir"]) + cluster, ok := storage["cluster"].(map[string]any) + require.True(t, ok, "cluster block missing") + 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"]) +} + +func TestRenderConfigExtraConfigReservedPath(t *testing.T) { + cr := testCluster() + cr.Spec.ExtraConfig = &runtime.RawExtension{ + Raw: []byte(`{"storage":{"cluster":{"etcd":["http://other:2379"]}}}`), + } + _, err := renderConfig(cr, cr.Spec.Etcd.Endpoints) + require.ErrorContains(t, err, "storage.cluster") + + var invalid validationError + require.True(t, errors.As(err, &invalid), "reserved paths must fail as a spec validation error") +} + func TestBuildStatefulSetIdentityAndVolumes(t *testing.T) { cr := testCluster() sts := buildStatefulSet(cr, "deadbeef") diff --git a/internal/controller/config.go b/internal/controller/config.go index 226d967..ecc8583 100644 --- a/internal/controller/config.go +++ b/internal/controller/config.go @@ -19,7 +19,6 @@ package controller import ( "encoding/json" "fmt" - "maps" "sigs.k8s.io/yaml" @@ -41,31 +40,31 @@ func renderConfig(cr *dbv1alpha1.OtelDBCluster, etcdEndpoints []string) (string, // Every enabled signal is served from the embedded clustered storage engine; no ClickHouse. // A disabled signal drops both its backend and its API bind, so the API is not served at all. if metricsEnabled(cr) { - cfg["metrics_backend"] = valStorage + cfg[keyMetricsBackend] = valStorage cfg["prometheus"] = map[string]any{keyBind: "0.0.0.0:9090"} } if tracesEnabled(cr) { - cfg["traces_backend"] = valStorage + cfg[keyTracesBackend] = valStorage cfg["tempo"] = map[string]any{keyBind: "0.0.0.0:3200"} } if logsEnabled(cr) { - cfg["logs_backend"] = valStorage + cfg[keyLogsBackend] = valStorage cfg["loki"] = map[string]any{keyBind: "0.0.0.0:3100"} } if profilesEnabled(cr) { - cfg["profiles_backend"] = valStorage + cfg[keyProfilesBackend] = valStorage cfg["pyroscope"] = map[string]any{keyBind: "0.0.0.0:4040"} } storage := map[string]any{ - "backend": string(backendOf(cr)), - "dir": dirOf(cr), + keyBackend: string(backendOf(cr)), + keyDir: dirOf(cr), } // The cluster block. Only the deployment-wide settings live here; id/addr/zone come from env. cluster := map[string]any{ - "etcd": etcdEndpoints, - "port": int(peerPortOf(cr)), + keyEtcd: etcdEndpoints, + keyPort: int(peerPortOf(cr)), } if rf := cr.Spec.Cluster.ReplicationFactor; rf != nil { cluster["rf"] = int(*rf) @@ -83,7 +82,7 @@ func renderConfig(cr *dbv1alpha1.OtelDBCluster, etcdEndpoints []string) (string, if backendOf(cr) == dbv1alpha1.StorageBackendS3 { s3 := cr.Spec.Storage.S3 if s3 == nil || s3.Bucket == "" { - return "", fmt.Errorf("storage.s3.bucket is required when storage.backend is s3") + return "", invalidSpec("spec.storage.s3.bucket is required when spec.storage.backend is s3") } m := map[string]any{ "bucket": s3.Bucket, @@ -122,13 +121,18 @@ func renderConfig(cr *dbv1alpha1.OtelDBCluster, etcdEndpoints []string) (string, cfg["storage"] = storage - // Merge user-supplied ExtraConfig over the generated config (top-level keys win). + // Merge user-supplied ExtraConfig over the generated config. The merge is recursive so that, + // say, storage.policy can be added without discarding the generated storage block; paths the + // operator owns are rejected rather than merged (see reservedConfigPaths). if raw := cr.Spec.ExtraConfig; raw != nil && len(raw.Raw) > 0 { var extra map[string]any if err := json.Unmarshal(raw.Raw, &extra); err != nil { - return "", fmt.Errorf("parse extraConfig: %w", err) + return "", invalidSpec("parse spec.extraConfig: %s", err) } - maps.Copy(cfg, extra) + if err := validateExtraConfig(extra); err != nil { + return "", err + } + deepMerge(cfg, extra) } out, err := yaml.Marshal(cfg) diff --git a/internal/controller/extraconfig.go b/internal/controller/extraconfig.go new file mode 100644 index 0000000..4655ae3 --- /dev/null +++ b/internal/controller/extraconfig.go @@ -0,0 +1,116 @@ +/* +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 ( + "fmt" + "slices" + "strings" +) + +const embeddedStorageHint = "the operator always serves signals from the embedded storage engine" + +// reservedConfigPaths are the config paths the operator renders and owns: setting them from +// spec.extraConfig would silently fight the spec (and, for storage.backend/cluster, cost durability +// or cluster membership). Each maps to the spec field that must be used instead. +// +// A reserved path also covers everything below it, so "storage.cluster" rejects +// "storage.cluster.etcd" too. +var reservedConfigPaths = map[string]string{ + keyMetricsBackend: embeddedStorageHint, + keyTracesBackend: embeddedStorageHint, + keyLogsBackend: embeddedStorageHint, + keyProfilesBackend: "use spec.signals.profiles", + + "storage.backend": "use spec.storage.backend", + "storage.dir": "use spec.storage.dir", + "storage.wal_dir": "use spec.storage.dir", + "storage.s3": "use spec.storage.s3", + "storage.cluster": "use spec.cluster and spec.etcd.endpoints", + "storage.flush_interval": "use spec.engine.flushInterval", + "storage.read_cache_bytes": "use spec.engine.readCacheSize", + "storage.decode_cache_bytes": "use spec.engine.decodeCacheSize", + "storage.decode_memory_bytes": "use spec.engine.decodeMemoryLimit", + "storage.aggregate_stats": "use spec.engine.aggregateStats", +} + +// validationError marks a spec problem that no amount of retrying can fix: the reconcile is +// reported on the CR status and not requeued until the spec changes. +type validationError struct { + err error +} + +func (e validationError) Error() string { return e.err.Error() } +func (e validationError) Unwrap() error { return e.err } + +func invalidSpec(format string, args ...any) error { + return validationError{err: fmt.Errorf(format, args...)} +} + +// validateExtraConfig rejects extraConfig that targets operator-owned paths. +func validateExtraConfig(extra map[string]any) error { + var found []string + collectReservedPaths(extra, "", &found) + if len(found) == 0 { + return nil + } + slices.Sort(found) + + msgs := make([]string, 0, len(found)) + for _, p := range found { + msgs = append(msgs, fmt.Sprintf("%s (%s)", p, reservedConfigPaths[p])) + } + return invalidSpec("spec.extraConfig sets reserved config %s: %s", + plural(len(found), "path", "paths"), strings.Join(msgs, ", ")) +} + +func collectReservedPaths(m map[string]any, prefix string, found *[]string) { + for k, v := range m { + path := k + if prefix != "" { + path = prefix + "." + k + } + if _, reserved := reservedConfigPaths[path]; reserved { + *found = append(*found, path) + continue // Everything below a reserved path is reserved as well. + } + if sub, ok := v.(map[string]any); ok { + collectReservedPaths(sub, path, found) + } + } +} + +// deepMerge recursively merges src into dst: nested maps are merged key by key, any other value +// (including a nil, a list, or a map replacing a scalar) replaces the one in dst. +func deepMerge(dst, src map[string]any) { + for k, sv := range src { + if sm, ok := sv.(map[string]any); ok { + if dm, ok := dst[k].(map[string]any); ok { + deepMerge(dm, sm) + continue + } + } + dst[k] = sv + } +} + +func plural(n int, one, many string) string { + if n == 1 { + return one + } + return many +} diff --git a/internal/controller/extraconfig_test.go b/internal/controller/extraconfig_test.go new file mode 100644 index 0000000..0e2964d --- /dev/null +++ b/internal/controller/extraconfig_test.go @@ -0,0 +1,170 @@ +/* +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" + + "github.com/stretchr/testify/require" +) + +func TestDeepMerge(t *testing.T) { + tests := []struct { + name string + dst map[string]any + src map[string]any + want map[string]any + }{ + { + name: "disjoint keys", + dst: map[string]any{"a": 1}, + src: map[string]any{"b": 2}, + want: map[string]any{"a": 1, "b": 2}, + }, + { + name: "leaf wins", + dst: map[string]any{"a": 1}, + src: map[string]any{"a": 2}, + want: map[string]any{"a": 2}, + }, + { + name: "nested maps merge", + dst: map[string]any{"storage": map[string]any{"backend": "file", "dir": "/data"}}, + src: map[string]any{"storage": map[string]any{"policy": "keep"}}, + want: map[string]any{"storage": map[string]any{"backend": "file", "dir": "/data", "policy": "keep"}}, + }, + { + name: "nested merge recurses", + dst: map[string]any{"storage": map[string]any{ + "cluster": map[string]any{"rf": 3, "port": 7946}, + }}, + src: map[string]any{"storage": map[string]any{ + "cluster": map[string]any{"rf": 5}, + }}, + want: map[string]any{"storage": map[string]any{ + "cluster": map[string]any{"rf": 5, "port": 7946}, + }}, + }, + { + name: "map replaces scalar", + dst: map[string]any{"a": 1}, + src: map[string]any{"a": map[string]any{"b": 2}}, + want: map[string]any{"a": map[string]any{"b": 2}}, + }, + { + name: "scalar replaces map", + dst: map[string]any{"a": map[string]any{"b": 2}}, + src: map[string]any{"a": "x"}, + want: map[string]any{"a": "x"}, + }, + { + name: "lists replace, never merge", + dst: map[string]any{"etcd": []any{"a", "b"}}, + src: map[string]any{"etcd": []any{"c"}}, + want: map[string]any{"etcd": []any{"c"}}, + }, + { + name: "explicit null replaces", + dst: map[string]any{"a": 1}, + src: map[string]any{"a": nil}, + want: map[string]any{"a": nil}, + }, + { + name: "empty source keeps destination", + dst: map[string]any{"a": 1}, + src: map[string]any{}, + want: map[string]any{"a": 1}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + deepMerge(tt.dst, tt.src) + require.Equal(t, tt.want, tt.dst) + }) + } +} + +func TestValidateExtraConfig(t *testing.T) { + tests := []struct { + name string + extra map[string]any + wantErr string // substring; empty means the config must be accepted + }{ + { + name: "unrelated keys", + extra: map[string]any{"ttl": "720h", "prometheus": map[string]any{"bind": ":9091"}}, + }, + { + name: "non-reserved storage key", + extra: map[string]any{"storage": map[string]any{"policy": map[string]any{"recompress": "3d"}}}, + }, + { + name: "storage backend", + extra: map[string]any{"storage": map[string]any{"backend": "memory"}}, + wantErr: "storage.backend (use spec.storage.backend)", + }, + { + name: "storage dir", + extra: map[string]any{"storage": map[string]any{"dir": "/tmp"}}, + wantErr: "storage.dir", + }, + { + name: "cluster subtree", + extra: map[string]any{"storage": map[string]any{"cluster": map[string]any{"etcd": []any{"http://evil:2379"}}}}, + wantErr: "storage.cluster (use spec.cluster and spec.etcd.endpoints)", + }, + { + name: "s3 subtree", + extra: map[string]any{"storage": map[string]any{"s3": map[string]any{"bucket": "other"}}}, + wantErr: "storage.s3", + }, + { + name: "engine tuning", + extra: map[string]any{"storage": map[string]any{"flush_interval": "1m"}}, + wantErr: "storage.flush_interval (use spec.engine.flushInterval)", + }, + { + name: "signal backend", + extra: map[string]any{"metrics_backend": "clickhouse"}, + wantErr: "metrics_backend", + }, + { + name: "several paths are all reported", + extra: map[string]any{"storage": map[string]any{ + "backend": "memory", + "dir": "/tmp", + }}, + wantErr: "reserved config paths: storage.backend", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := validateExtraConfig(tt.extra) + if tt.wantErr == "" { + require.NoError(t, err) + return + } + require.Error(t, err) + require.ErrorContains(t, err, tt.wantErr) + var invalid validationError + require.True(t, errors.As(err, &invalid), "must be reported as a spec validation error") + }) + } +} diff --git a/internal/controller/naming.go b/internal/controller/naming.go index 6e57fca..2005630 100644 --- a/internal/controller/naming.go +++ b/internal/controller/naming.go @@ -36,6 +36,19 @@ const ( envAWSSecretAccessKey = "AWS_SECRET_ACCESS_KEY" ) +// oteldb config keys shared by the renderer and the reserved-path guard. +const ( + keyMetricsBackend = "metrics_backend" + keyTracesBackend = "traces_backend" + keyLogsBackend = "logs_backend" + keyProfilesBackend = "profiles_backend" + + keyBackend = "backend" + keyDir = "dir" + keyEtcd = "etcd" + keyPort = "port" +) + // Container ports exposed by every oteldb node. Names must be <= 15 chars (k8s port-name limit). const ( portOTLPGRPC = 4317 diff --git a/internal/controller/oteldbcluster_controller.go b/internal/controller/oteldbcluster_controller.go index f42546e..fbb6448 100644 --- a/internal/controller/oteldbcluster_controller.go +++ b/internal/controller/oteldbcluster_controller.go @@ -18,6 +18,7 @@ package controller import ( "context" + "errors" "fmt" appsv1 "k8s.io/api/apps/v1" @@ -61,8 +62,15 @@ func (r *OtelDBClusterReconciler) Reconcile(ctx context.Context, req ctrl.Reques } if err := r.reconcile(ctx, cr); err != nil { + // An invalid spec cannot be fixed by retrying: report it and wait for the next spec change. + var invalid validationError + if errors.As(err, &invalid) { + log.Error(err, "Rejected invalid OtelDBCluster spec") + r.setDegraded(ctx, cr, "InvalidSpec", err) + return ctrl.Result{}, nil + } log.Error(err, "reconcile failed") - r.setDegraded(ctx, cr, err) + r.setDegraded(ctx, cr, "ReconcileError", err) return ctrl.Result{}, err } @@ -75,7 +83,7 @@ func (r *OtelDBClusterReconciler) Reconcile(ctx context.Context, req ctrl.Reques func (r *OtelDBClusterReconciler) reconcile(ctx context.Context, cr *dbv1alpha1.OtelDBCluster) error { endpoints := cr.Spec.Etcd.Endpoints if len(endpoints) == 0 { - return fmt.Errorf("spec.etcd.endpoints is required (bring your own etcd)") + return invalidSpec("spec.etcd.endpoints is required (bring your own etcd)") } cm, err := buildConfigMap(cr, endpoints) @@ -182,9 +190,9 @@ func (r *OtelDBClusterReconciler) updateStatus(ctx context.Context, cr *dbv1alph return r.Status().Update(ctx, cr) } -func (r *OtelDBClusterReconciler) setDegraded(ctx context.Context, cr *dbv1alpha1.OtelDBCluster, cause error) { +func (r *OtelDBClusterReconciler) setDegraded(ctx context.Context, cr *dbv1alpha1.OtelDBCluster, reason string, cause error) { cr.Status.Phase = dbv1alpha1.PhaseDegraded - setCondition(cr, dbv1alpha1.ConditionDegraded, metav1.ConditionTrue, "ReconcileError", cause.Error()) + setCondition(cr, dbv1alpha1.ConditionDegraded, metav1.ConditionTrue, reason, cause.Error()) _ = r.Status().Update(ctx, cr) }