diff --git a/api/formance.com/v1beta1/otelexporterendpoint_types.go b/api/formance.com/v1beta1/otelexporterendpoint_types.go new file mode 100644 index 000000000..455c5c754 --- /dev/null +++ b/api/formance.com/v1beta1/otelexporterendpoint_types.go @@ -0,0 +1,130 @@ +/* +Copyright 2023. + +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 v1beta1 + +import ( + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +// OtelExporterAuth configures per-signal authentication. +// Auth is per-signal so traces and metrics can use different credentials if needed. +type OtelExporterAuth struct { + // Type is the authentication type. + // +kubebuilder:validation:Enum=bearer + Type string `json:"type"` + // FromSecret references a Secret name. + // The controller creates a ResourceReference to replicate the secret into each target stack namespace. + // The source secret must have a "formance.com/stack" label set to "any" or a specific stack name. + // +kubebuilder:validation:MinLength=1 + FromSecret string `json:"fromSecret"` + // FromSecretKey is the key within the Secret that contains the token. Defaults to "token". + // +optional + // +kubebuilder:default="token" + FromSecretKey string `json:"fromSecretKey,omitempty"` +} + +// OtelSignalConfig configures a single signal type (traces or metrics). +// Each signal type has its own endpoint and authentication block, allowing +// different destinations or credentials per signal. +// Protocol is inferred from the URL scheme: grpc:// for gRPC, http:// or https:// for HTTP/protobuf (default). +type OtelSignalConfig struct { + // Endpoint URL for the signal (e.g., "http://my-collector:4318", "grpc://my-collector:4317"). + // Supported schemes: http, https, grpc. Bare host:port is treated as HTTP. + // Protocol is inferred from the URL scheme. HTTP/protobuf is the default for firewall compatibility. + // +kubebuilder:validation:MinLength=1 + // +kubebuilder:validation:Pattern=`^(https?://|grpc://|[a-zA-Z0-9])` + Endpoint string `json:"endpoint"` + + // Auth is the optional per-signal authentication configuration. + // +optional + Auth *OtelExporterAuth `json:"auth,omitempty"` +} + +// +kubebuilder:validation:XValidation:rule="has(self.traces) || has(self.metrics)",message="at least one signal (traces or metrics) must be configured" +type OtelExporterEndpointSpec struct { + // StackSelector is a standard Kubernetes LabelSelector (matchLabels/matchExpressions). + // One CRD can target all current and future stacks with a single selector. + // Matches the pattern established by Settings. + StackSelector *metav1.LabelSelector `json:"stackSelector"` + + // Traces configures the traces signal. At least one of traces or metrics must be set. + // Logs are intentionally out of scope. + // +optional + Traces *OtelSignalConfig `json:"traces,omitempty"` + + // Metrics configures the metrics signal. At least one of traces or metrics must be set. + // Logs are intentionally out of scope. + // +optional + Metrics *OtelSignalConfig `json:"metrics,omitempty"` + + // ResourceAttributes are injected into outgoing telemetry via a collector processor. + // +optional + ResourceAttributes map[string]string `json:"resourceAttributes,omitempty"` +} + +// OtelExporterEndpointStatus represents the observed state of an OtelExporterEndpoint. +type OtelExporterEndpointStatus struct { + Status `json:",inline"` + // Stacks is a sorted list of stack names currently targeted by this endpoint. + // Includes stacks with successful reconciliation and stacks with transient errors or pending cleanup. + // Used by the finalizer to find previously matched stacks during deletion. + // +optional + Stacks []string `json:"stacks,omitempty"` +} + +// OtelExporterEndpoint configures an OpenTelemetry collector proxy for exporting traces and metrics. +// Multiple OtelExporterEndpoints can target the same stacks — the collector fans out to all matching destinations. +// +// +kubebuilder:object:root=true +// +kubebuilder:subresource:status +// +kubebuilder:resource:scope=Cluster +// +kubebuilder:printcolumn:name="Ready",type=string,JSONPath=".status.ready",description="Is ready" +// +kubebuilder:printcolumn:name="Info",type=string,JSONPath=".status.info",description="Info" +type OtelExporterEndpoint struct { + metav1.TypeMeta `json:",inline"` + metav1.ObjectMeta `json:"metadata,omitempty"` + + Spec OtelExporterEndpointSpec `json:"spec,omitempty"` + Status OtelExporterEndpointStatus `json:"status,omitempty"` +} + +func (in *OtelExporterEndpoint) IsReady() bool { + return in.Status.Ready +} + +func (in *OtelExporterEndpoint) SetReady(b bool) { + in.Status.Ready = b +} + +func (in *OtelExporterEndpoint) SetError(s string) { + in.Status.Info = s +} + +func (in *OtelExporterEndpoint) GetConditions() *Conditions { + return &in.Status.Conditions +} + +// +kubebuilder:object:root=true +type OtelExporterEndpointList struct { + metav1.TypeMeta `json:",inline"` + metav1.ListMeta `json:"metadata,omitempty"` + Items []OtelExporterEndpoint `json:"items"` +} + +func init() { + SchemeBuilder.Register(&OtelExporterEndpoint{}, &OtelExporterEndpointList{}) +} diff --git a/api/formance.com/v1beta1/zz_generated.deepcopy.go b/api/formance.com/v1beta1/zz_generated.deepcopy.go index 552862474..0775d3a97 100644 --- a/api/formance.com/v1beta1/zz_generated.deepcopy.go +++ b/api/formance.com/v1beta1/zz_generated.deepcopy.go @@ -1516,6 +1516,158 @@ func (in *OrchestrationStatus) DeepCopy() *OrchestrationStatus { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *OtelExporterAuth) DeepCopyInto(out *OtelExporterAuth) { + *out = *in +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new OtelExporterAuth. +func (in *OtelExporterAuth) DeepCopy() *OtelExporterAuth { + if in == nil { + return nil + } + out := new(OtelExporterAuth) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *OtelExporterEndpoint) DeepCopyInto(out *OtelExporterEndpoint) { + *out = *in + out.TypeMeta = in.TypeMeta + in.ObjectMeta.DeepCopyInto(&out.ObjectMeta) + in.Spec.DeepCopyInto(&out.Spec) + in.Status.DeepCopyInto(&out.Status) +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new OtelExporterEndpoint. +func (in *OtelExporterEndpoint) DeepCopy() *OtelExporterEndpoint { + if in == nil { + return nil + } + out := new(OtelExporterEndpoint) + in.DeepCopyInto(out) + return out +} + +// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object. +func (in *OtelExporterEndpoint) DeepCopyObject() runtime.Object { + if c := in.DeepCopy(); c != nil { + return c + } + return nil +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *OtelExporterEndpointList) DeepCopyInto(out *OtelExporterEndpointList) { + *out = *in + out.TypeMeta = in.TypeMeta + in.ListMeta.DeepCopyInto(&out.ListMeta) + if in.Items != nil { + in, out := &in.Items, &out.Items + *out = make([]OtelExporterEndpoint, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new OtelExporterEndpointList. +func (in *OtelExporterEndpointList) DeepCopy() *OtelExporterEndpointList { + if in == nil { + return nil + } + out := new(OtelExporterEndpointList) + in.DeepCopyInto(out) + return out +} + +// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object. +func (in *OtelExporterEndpointList) DeepCopyObject() runtime.Object { + if c := in.DeepCopy(); c != nil { + return c + } + return nil +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *OtelExporterEndpointSpec) DeepCopyInto(out *OtelExporterEndpointSpec) { + *out = *in + if in.StackSelector != nil { + in, out := &in.StackSelector, &out.StackSelector + *out = new(metav1.LabelSelector) + (*in).DeepCopyInto(*out) + } + if in.Traces != nil { + in, out := &in.Traces, &out.Traces + *out = new(OtelSignalConfig) + (*in).DeepCopyInto(*out) + } + if in.Metrics != nil { + in, out := &in.Metrics, &out.Metrics + *out = new(OtelSignalConfig) + (*in).DeepCopyInto(*out) + } + if in.ResourceAttributes != nil { + in, out := &in.ResourceAttributes, &out.ResourceAttributes + *out = make(map[string]string, len(*in)) + for key, val := range *in { + (*out)[key] = val + } + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new OtelExporterEndpointSpec. +func (in *OtelExporterEndpointSpec) DeepCopy() *OtelExporterEndpointSpec { + if in == nil { + return nil + } + out := new(OtelExporterEndpointSpec) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *OtelExporterEndpointStatus) DeepCopyInto(out *OtelExporterEndpointStatus) { + *out = *in + in.Status.DeepCopyInto(&out.Status) + if in.Stacks != nil { + in, out := &in.Stacks, &out.Stacks + *out = make([]string, len(*in)) + copy(*out, *in) + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new OtelExporterEndpointStatus. +func (in *OtelExporterEndpointStatus) DeepCopy() *OtelExporterEndpointStatus { + if in == nil { + return nil + } + out := new(OtelExporterEndpointStatus) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *OtelSignalConfig) DeepCopyInto(out *OtelSignalConfig) { + *out = *in + if in.Auth != nil { + in, out := &in.Auth, &out.Auth + *out = new(OtelExporterAuth) + **out = **in + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new OtelSignalConfig. +func (in *OtelSignalConfig) DeepCopy() *OtelSignalConfig { + if in == nil { + return nil + } + out := new(OtelSignalConfig) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *Payments) DeepCopyInto(out *Payments) { *out = *in diff --git a/cmd/main.go b/cmd/main.go index 494a7ea9b..bcde35a70 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -63,6 +63,7 @@ func main() { licenceSecret string licenceNamespace string utilsVersion string + collectorImage string ) flag.StringVar(&metricsAddr, "metrics-bind-address", ":8080", "The address the metric endpoint binds to.") flag.StringVar(&probeAddr, "health-probe-bind-address", ":8081", "The address the probe endpoint binds to.") @@ -74,6 +75,7 @@ func main() { flag.StringVar(&licenceSecret, "licence-secret", "", "The licence secret that contains the token and the issuer") flag.StringVar(&licenceNamespace, "licence-namespace", "", "The namespace where the licence secret lives (defaults to operator namespace)") flag.StringVar(&utilsVersion, "utils-version", "latest", "The version of the operator utils image") + flag.StringVar(&collectorImage, "collector-image", core.DefaultCollectorImage, "The OTel Collector image for OtelExporterEndpoint resources") opts := zap.Options{ Development: false, } @@ -125,6 +127,7 @@ func main() { LicenceSecret: licenceSecret, LicenceNamespace: licenceNamespace, UtilsVersion: utilsVersion, + CollectorImage: collectorImage, } if licenceSecret != "" { diff --git a/config/crd/bases/formance.com_otelexporterendpoints.yaml b/config/crd/bases/formance.com_otelexporterendpoints.yaml new file mode 100644 index 000000000..0c716faeb --- /dev/null +++ b/config/crd/bases/formance.com_otelexporterendpoints.yaml @@ -0,0 +1,271 @@ +--- +apiVersion: apiextensions.k8s.io/v1 +kind: CustomResourceDefinition +metadata: + annotations: + controller-gen.kubebuilder.io/version: v0.18.0 + name: otelexporterendpoints.formance.com +spec: + group: formance.com + names: + kind: OtelExporterEndpoint + listKind: OtelExporterEndpointList + plural: otelexporterendpoints + singular: otelexporterendpoint + scope: Cluster + versions: + - additionalPrinterColumns: + - description: Is ready + jsonPath: .status.ready + name: Ready + type: string + - description: Info + jsonPath: .status.info + name: Info + type: string + name: v1beta1 + schema: + openAPIV3Schema: + description: |- + OtelExporterEndpoint configures an OpenTelemetry collector proxy for exporting traces and metrics. + Multiple OtelExporterEndpoints can target the same stacks — the collector fans out to all matching destinations. + properties: + apiVersion: + description: |- + APIVersion defines the versioned schema of this representation of an object. + Servers should convert recognized schemas to the latest internal value, and + may reject unrecognized values. + More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources + type: string + kind: + description: |- + Kind is a string value representing the REST resource this object represents. + Servers may infer this from the endpoint the client submits requests to. + Cannot be updated. + In CamelCase. + More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds + type: string + metadata: + type: object + spec: + properties: + metrics: + description: |- + Metrics configures the metrics signal. At least one of traces or metrics must be set. + Logs are intentionally out of scope. + properties: + auth: + description: Auth is the optional per-signal authentication configuration. + properties: + fromSecret: + description: |- + FromSecret references a Secret name. + The controller creates a ResourceReference to replicate the secret into each target stack namespace. + The source secret must have a "formance.com/stack" label set to "any" or a specific stack name. + minLength: 1 + type: string + fromSecretKey: + default: token + description: FromSecretKey is the key within the Secret that + contains the token. Defaults to "token". + type: string + type: + description: Type is the authentication type. + enum: + - bearer + type: string + required: + - fromSecret + - type + type: object + endpoint: + description: |- + Endpoint URL for the signal (e.g., "http://my-collector:4318", "grpc://my-collector:4317"). + Supported schemes: http, https, grpc. Bare host:port is treated as HTTP. + Protocol is inferred from the URL scheme. HTTP/protobuf is the default for firewall compatibility. + minLength: 1 + pattern: ^(https?://|grpc://|[a-zA-Z0-9]) + type: string + required: + - endpoint + type: object + resourceAttributes: + additionalProperties: + type: string + description: ResourceAttributes are injected into outgoing telemetry + via a collector processor. + type: object + stackSelector: + description: |- + StackSelector is a standard Kubernetes LabelSelector (matchLabels/matchExpressions). + One CRD can target all current and future stacks with a single selector. + Matches the pattern established by Settings. + properties: + matchExpressions: + description: matchExpressions is a list of label selector requirements. + The requirements are ANDed. + items: + description: |- + A label selector requirement is a selector that contains values, a key, and an operator that + relates the key and values. + properties: + key: + description: key is the label key that the selector applies + to. + type: string + operator: + description: |- + operator represents a key's relationship to a set of values. + Valid operators are In, NotIn, Exists and DoesNotExist. + type: string + values: + description: |- + values is an array of string values. If the operator is In or NotIn, + the values array must be non-empty. If the operator is Exists or DoesNotExist, + the values array must be empty. This array is replaced during a strategic + merge patch. + items: + type: string + type: array + x-kubernetes-list-type: atomic + required: + - key + - operator + type: object + type: array + x-kubernetes-list-type: atomic + matchLabels: + additionalProperties: + type: string + description: |- + matchLabels is a map of {key,value} pairs. A single {key,value} in the matchLabels + map is equivalent to an element of matchExpressions, whose key field is "key", the + operator is "In", and the values array contains only "value". The requirements are ANDed. + type: object + type: object + x-kubernetes-map-type: atomic + traces: + description: |- + Traces configures the traces signal. At least one of traces or metrics must be set. + Logs are intentionally out of scope. + properties: + auth: + description: Auth is the optional per-signal authentication configuration. + properties: + fromSecret: + description: |- + FromSecret references a Secret name. + The controller creates a ResourceReference to replicate the secret into each target stack namespace. + The source secret must have a "formance.com/stack" label set to "any" or a specific stack name. + minLength: 1 + type: string + fromSecretKey: + default: token + description: FromSecretKey is the key within the Secret that + contains the token. Defaults to "token". + type: string + type: + description: Type is the authentication type. + enum: + - bearer + type: string + required: + - fromSecret + - type + type: object + endpoint: + description: |- + Endpoint URL for the signal (e.g., "http://my-collector:4318", "grpc://my-collector:4317"). + Supported schemes: http, https, grpc. Bare host:port is treated as HTTP. + Protocol is inferred from the URL scheme. HTTP/protobuf is the default for firewall compatibility. + minLength: 1 + pattern: ^(https?://|grpc://|[a-zA-Z0-9]) + type: string + required: + - endpoint + type: object + required: + - stackSelector + type: object + x-kubernetes-validations: + - message: at least one signal (traces or metrics) must be configured + rule: has(self.traces) || has(self.metrics) + status: + description: OtelExporterEndpointStatus represents the observed state + of an OtelExporterEndpoint. + properties: + conditions: + items: + description: Condition contains details for one aspect of the current + state of this API Resource. + properties: + lastTransitionTime: + description: |- + lastTransitionTime is the last time the condition transitioned from one status to another. + This should be when the underlying condition changed. If that is not known, then using the time when the API field changed is acceptable. + format: date-time + type: string + message: + description: |- + message is a human readable message indicating details about the transition. + This may be an empty string. + maxLength: 32768 + type: string + observedGeneration: + description: |- + observedGeneration represents the .metadata.generation that the condition was set based upon. + For instance, if .metadata.generation is currently 12, but the .status.conditions[x].observedGeneration is 9, the condition is out of date + with respect to the current state of the instance. + format: int64 + minimum: 0 + type: integer + reason: + description: |- + reason contains a programmatic identifier indicating the reason for the condition's last transition. + Producers of specific condition types may define expected values and meanings for this field, + and whether the values are considered a guaranteed API. + The value should be a CamelCase string. + This field may not be empty. + maxLength: 1024 + pattern: ^([A-Za-z]([A-Za-z0-9_,:]*[A-Za-z0-9_])?)?$ + type: string + status: + description: status of the condition, one of True, False, Unknown. + enum: + - "True" + - "False" + - Unknown + type: string + type: + description: type of condition in CamelCase or in foo.example.com/CamelCase. + maxLength: 316 + pattern: ^([a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*/)?(([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9])$ + type: string + required: + - lastTransitionTime + - message + - status + - type + type: object + type: array + info: + description: Info can contain any additional like reconciliation errors + type: string + ready: + description: Ready indicates if the resource is seen as completely + reconciled + type: boolean + stacks: + description: |- + Stacks is a sorted list of stack names currently targeted by this endpoint. + Includes stacks with successful reconciliation and stacks with transient errors or pending cleanup. + Used by the finalizer to find previously matched stacks during deletion. + items: + type: string + type: array + type: object + type: object + served: true + storage: true + subresources: + status: {} diff --git a/config/crd/kustomization.yaml b/config/crd/kustomization.yaml index 65e9b2c58..cb501590e 100644 --- a/config/crd/kustomization.yaml +++ b/config/crd/kustomization.yaml @@ -26,6 +26,7 @@ resources: - bases/formance.com_brokerconsumers.yaml - bases/formance.com_brokers.yaml - bases/formance.com_transactionplanes.yaml +- bases/formance.com_otelexporterendpoints.yaml #+kubebuilder:scaffold:crdkustomizeresource diff --git a/config/rbac/role.yaml b/config/rbac/role.yaml index 9e2bae12a..bf8876117 100644 --- a/config/rbac/role.yaml +++ b/config/rbac/role.yaml @@ -94,6 +94,7 @@ rules: - ledgers - mcps - orchestrations + - otelexporterendpoints - payments - reconciliations - resourcereferences @@ -129,6 +130,7 @@ rules: - ledgers/finalizers - mcps/finalizers - orchestrations/finalizers + - otelexporterendpoints/finalizers - payments/finalizers - reconciliations/finalizers - resourcereferences/finalizers @@ -158,6 +160,7 @@ rules: - ledgers/status - mcps/status - orchestrations/status + - otelexporterendpoints/status - payments/status - reconciliations/status - resourcereferences/status diff --git a/config/samples/formance.com_v1beta1_otelexporterendpoint.yaml b/config/samples/formance.com_v1beta1_otelexporterendpoint.yaml new file mode 100644 index 000000000..2bd6b44f0 --- /dev/null +++ b/config/samples/formance.com_v1beta1_otelexporterendpoint.yaml @@ -0,0 +1,46 @@ +apiVersion: formance.com/v1beta1 +kind: OtelExporterEndpoint +metadata: + labels: + app.kubernetes.io/name: otelexporterendpoint + app.kubernetes.io/instance: otelexporterendpoint0 + app.kubernetes.io/part-of: operatorv2 + app.kubernetes.io/managed-by: kustomize + app.kubernetes.io/created-by: operatorv2 + name: otelexporterendpoint0 +spec: + stackSelector: + matchLabels: + formance.com/stack: any + traces: + endpoint: "http://otel-collector-opentelemetry-collector.formance.svc.cluster.local:4318" + metrics: + endpoint: "http://otel-collector-opentelemetry-collector.formance.svc.cluster.local:4318" +--- +apiVersion: formance.com/v1beta1 +kind: OtelExporterEndpoint +metadata: + labels: + app.kubernetes.io/name: otelexporterendpoint + app.kubernetes.io/instance: formance-support + app.kubernetes.io/part-of: operatorv2 + app.kubernetes.io/managed-by: kustomize + app.kubernetes.io/created-by: operatorv2 + name: formance-support +spec: + stackSelector: + matchExpressions: + - key: formance.com/stack + operator: Exists + traces: + endpoint: "https://otel-support.internal.frmnc.net/v1/traces" + auth: + type: bearer + fromSecret: "formance-licence" + metrics: + endpoint: "https://otel-support.internal.frmnc.net/v1/metrics" + auth: + type: bearer + fromSecret: "formance-licence" + resourceAttributes: + cluster.id: "abc-123" diff --git a/crd-doc-templates/type.tpl b/crd-doc-templates/type.tpl index 4d2a3eb3b..2a9e314c6 100644 --- a/crd-doc-templates/type.tpl +++ b/crd-doc-templates/type.tpl @@ -2,6 +2,7 @@ {{- $type := .Type -}} {{- $recurse := .Recurse -}} {{- $prefix := .Prefix -}} +{{- $ctx := . -}} {{ repeat ($prefix | int) "#" }} {{ $type.Name }} @@ -43,13 +44,16 @@ {{- if $recurse }} {{- if eq $type.Kind 4}} -{{- $dummy := set . "Fields" $type.UnderlyingType.Fields }} +{{- $dummy := set $ctx "Fields" $type.UnderlyingType.Fields }} {{- else }} -{{- $dummy := set . "Fields" $type.Fields }} +{{- $dummy := set $ctx "Fields" $type.Fields }} {{- end }} -{{- range $k, $field := .Fields }} -{{- if hasPrefix "github.com/formancehq/operator/api" $field.Type.Package }} +{{- range $k, $field := $ctx.Fields }} +{{- if hasPrefix "github.com/formancehq/operator/v3/api" $field.Type.Package }} {{- if has "Type: string" $field.Type.Validation }}{{ continue }}{{ end }} +{{- $seenKey := printf "_seen_%s" $field.Type.Name }} +{{- if hasKey $ctx $seenKey }}{{ continue }}{{ end }} +{{- $_ := set $ctx $seenKey true }} {{ template "type" (dict "Type" $field.Type "Recurse" true "Prefix" (min (add $prefix 1) 6)) }} {{- end }} {{- end }} diff --git a/docs/09-Configuration reference/02-Custom Resource Definitions.md b/docs/09-Configuration reference/02-Custom Resource Definitions.md index 3d03b7035..ee1b38b0f 100644 --- a/docs/09-Configuration reference/02-Custom Resource Definitions.md +++ b/docs/09-Configuration reference/02-Custom Resource Definitions.md @@ -40,6 +40,7 @@ Other resources : - [BrokerTopic](#brokertopic) - [Database](#database) - [GatewayHTTPAPI](#gatewayhttpapi) +- [OtelExporterEndpoint](#otelexporterendpoint) - [ResourceReference](#resourcereference) - [Versions](#versions) @@ -443,6 +444,33 @@ The auth service is basically a proxy to another OIDC compliant server. | `signingKeyFromSecret` _[SecretKeySelector](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.27/#secretkeyselector-v1-core)_ | Allow to override the default signing key used to sign JWT tokens using a k8s secret | | | | `enableScopes` _boolean_ | Allow to enable scopes usage on authentication.
If not enabled, each service will check the authentication but will not restrict access following scopes.
in this case, if authenticated, it is ok. | false | | +###### DelegatedOIDCServerConfiguration + + + + + + + + + + + + + + + + + + + +| Field | Description | Default | Validation | +| --- | --- | --- | --- | +| `issuer` _string_ | Issuer is the url of the delegated oidc server | | | +| `clientID` _string_ | ClientID is the client id to use for authentication | | | +| `clientSecret` _string_ | ClientSecret is the client secret to use for authentication | | | +| `clientSecretFromSecret` _[SecretKeySelector](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.27/#secretkeyselector-v1-core)_ | ClientSecretFromSecret is the client secret to use for authentication | | | + @@ -532,6 +560,59 @@ Gateway is the Schema for the gateways API | `version` _string_ | Version allow to override global version defined at stack level for a specific module | | | | `ingress` _[GatewayIngress](#gatewayingress)_ | Allow to customize the generated ingress | | | +###### GatewayIngress + + + +GatewayIngress represents the ingress configuration for the gateway. + + + + + + + + + + + + + + + +| Field | Description | Default | Validation | +| --- | --- | --- | --- | +| `host` _string_ | Indicates the hostname on which the stack will be served.
Example : `formance.example.com` | | | +| `hosts` _string array_ | Additional hosts for the ingress. Combined with Host. | | | +| `scheme` _string_ | Indicate the scheme.
Actually, It should be `https` unless you know what you are doing. | https | | +| `ingressClassName` _string_ | Ingress class to use | | | +| `annotations` _object (keys:string, values:string)_ | Custom annotations to add on the ingress | | | +| `tls` _[GatewayIngressTLS](#gatewayingresstls)_ | Allow to customize the tls part of the ingress | | | + +###### GatewayIngressTLS + + + + + + + + + + + + + + + + + + + +| Field | Description | Default | Validation | +| --- | --- | --- | --- | +| `secretName` _string_ | Specify the secret name used for the tls configuration on the ingress | | | + @@ -1156,6 +1237,56 @@ Stargate is the Schema for the stargates API | `auth` _[StargateAuthSpec](#stargateauthspec)_ | | | | | `tls` _[StargateTLSConfig](#stargatetlsconfig)_ | | | | +###### StargateAuthSpec + + + + + + + + + + + + + + + + + + + +| Field | Description | Default | Validation | +| --- | --- | --- | --- | +| `clientID` _string_ | | | | +| `clientSecret` _string_ | | | | +| `issuer` _string_ | | | | + +###### StargateTLSConfig + + + + + + + + + + + + + + + + + + + +| Field | Description | Default | Validation | +| --- | --- | --- | --- | +| `disable` _boolean_ | Disable TLS protocol -- use at your own risks, the transmission will be in clear. | | | + @@ -1808,6 +1939,27 @@ Broker is the Schema for the brokers API | `mode` _[Mode](#mode)_ | Mode indicating the configuration of the nats streams
Two modes are defined :
* ModeOneStreamByService: In this case, each service will have a dedicated stream created
* ModeOneStreamByStack: In this case, a stream will be created for the stack and each service will use a specific subject inside this stream | | Enum: [OneStreamByService OneStreamByStack]
| | `streams` _string array_ | Streams list streams created when Mode == ModeOneStreamByService | | | +###### Mode + +_Underlying type:_ _string_ + +Mode defined how streams are created on the broker (mainly nats) + + + + + + + + + + + + + + + + #### BrokerConsumer @@ -2146,6 +2298,32 @@ GatewayHTTPAPI is the Schema for the HTTPAPIs API | `rules` _[GatewayHTTPAPIRule](#gatewayhttpapirule) array_ | Rules | | | | `healthCheckEndpoint` _string_ | Health check endpoint | | | +###### GatewayHTTPAPIRule + + + + + + + + + + + + + + + + + + + +| Field | Description | Default | Validation | +| --- | --- | --- | --- | +| `path` _string_ | | | | +| `methods` _string array_ | | | | +| `secured` _boolean_ | | false | | + @@ -2177,6 +2355,150 @@ GatewayHTTPAPI is the Schema for the HTTPAPIs API | `ready` _boolean_ | | | | +#### OtelExporterEndpoint + + + +OtelExporterEndpoint configures an OpenTelemetry collector proxy for exporting traces and metrics. +Multiple OtelExporterEndpoints can target the same stacks — the collector fans out to all matching destinations. + + + + + + + + + + + + + + + +| Field | Description | Default | Validation | +| --- | --- | --- | --- | +| `apiVersion` _string_ | `formance.com/v1beta1` | | | +| `kind` _string_ | `OtelExporterEndpoint` | | | +| `metadata` _[ObjectMeta](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.27/#objectmeta-v1-meta)_ | Refer to Kubernetes API documentation for fields of `metadata`. | | | +| `spec` _[OtelExporterEndpointSpec](#otelexporterendpointspec)_ | | | | +| `status` _[OtelExporterEndpointStatus](#otelexporterendpointstatus)_ | | | | + + + +##### OtelExporterEndpointSpec + + + + + + + + + + + + + + + + + + + +| Field | Description | Default | Validation | +| --- | --- | --- | --- | +| `stackSelector` _[LabelSelector](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.27/#labelselector-v1-meta)_ | StackSelector is a standard Kubernetes LabelSelector (matchLabels/matchExpressions).
One CRD can target all current and future stacks with a single selector.
Matches the pattern established by Settings. | | | +| `traces` _[OtelSignalConfig](#otelsignalconfig)_ | Traces configures the traces signal. At least one of traces or metrics must be set.
Logs are intentionally out of scope. | | | +| `metrics` _[OtelSignalConfig](#otelsignalconfig)_ | Metrics configures the metrics signal. At least one of traces or metrics must be set.
Logs are intentionally out of scope. | | | +| `resourceAttributes` _object (keys:string, values:string)_ | ResourceAttributes are injected into outgoing telemetry via a collector processor. | | | + +###### OtelSignalConfig + + + +OtelSignalConfig configures a single signal type (traces or metrics). +Each signal type has its own endpoint and authentication block, allowing +different destinations or credentials per signal. +Protocol is inferred from the URL scheme: grpc:// for gRPC, http:// or https:// for HTTP/protobuf (default). + + + + + + + + + + + + + + + +| Field | Description | Default | Validation | +| --- | --- | --- | --- | +| `endpoint` _string_ | Endpoint URL for the signal (e.g., "http://my-collector:4318", "grpc://my-collector:4317").
Supported schemes: http, https, grpc. Bare host:port is treated as HTTP.
Protocol is inferred from the URL scheme. HTTP/protobuf is the default for firewall compatibility. | | MinLength: 1
Pattern: `^(https?://|grpc://|[a-zA-Z0-9])`
| +| `auth` _[OtelExporterAuth](#otelexporterauth)_ | Auth is the optional per-signal authentication configuration. | | | + +###### OtelExporterAuth + + + +OtelExporterAuth configures per-signal authentication. +Auth is per-signal so traces and metrics can use different credentials if needed. + + + + + + + + + + + + + + + +| Field | Description | Default | Validation | +| --- | --- | --- | --- | +| `type` _string_ | Type is the authentication type. | | Enum: [bearer]
| +| `fromSecret` _string_ | FromSecret references a Secret name.
The controller creates a ResourceReference to replicate the secret into each target stack namespace.
The source secret must have a "formance.com/stack" label set to "any" or a specific stack name. | | MinLength: 1
| +| `fromSecretKey` _string_ | FromSecretKey is the key within the Secret that contains the token. Defaults to "token". | token | | + + + + + +##### OtelExporterEndpointStatus + + + +OtelExporterEndpointStatus represents the observed state of an OtelExporterEndpoint. + + + + + + + + + + + + + + + +| Field | Description | Default | Validation | +| --- | --- | --- | --- | +| `ready` _boolean_ | Ready indicates if the resource is seen as completely reconciled | | | +| `info` _string_ | Info can contain any additional like reconciliation errors | | | +| `stacks` _string array_ | Stacks is a sorted list of stack names currently targeted by this endpoint.
Includes stacks with successful reconciliation and stacks with transient errors or pending cleanup.
Used by the finalizer to find previously matched stacks during deletion. | | | + + #### ResourceReference diff --git a/docs/09-Configuration reference/settings.catalog.json b/docs/09-Configuration reference/settings.catalog.json index a4316ca86..f8ccbb2a4 100644 --- a/docs/09-Configuration reference/settings.catalog.json +++ b/docs/09-Configuration reference/settings.catalog.json @@ -540,21 +540,36 @@ "key": "opentelemetry.\u003cmonitoring-type\u003e.dsn", "valueType": "uri", "sources": [ - "internal/resources/settings/opentelemetry.go:52" + "internal/resources/settings/opentelemetry.go:156" ] }, { "key": "opentelemetry.\u003cmonitoring-type\u003e.resource-attributes", "valueType": "map[string]string", "sources": [ - "internal/resources/settings/opentelemetry.go:96" + "internal/resources/settings/opentelemetry.go:200" + ] + }, + { + "key": "opentelemetry.\u003csignal\u003e.resource-attributes", + "valueType": "map[string]string", + "sources": [ + "internal/resources/settings/opentelemetry.go:81" + ] + }, + { + "key": "opentelemetry.metrics.dsn", + "valueType": "uri", + "sources": [ + "internal/resources/otelexporterendpoints/init.go:647" ] }, { "key": "opentelemetry.traces.dsn", "valueType": "uri", "sources": [ - "internal/resources/settings/opentelemetry.go:38" + "internal/resources/otelexporterendpoints/init.go:643", + "internal/resources/settings/opentelemetry.go:146" ] }, { diff --git a/helm/crds/templates/crds/apiextensions.k8s.io_v1_customresourcedefinition_otelexporterendpoints.formance.com.yaml b/helm/crds/templates/crds/apiextensions.k8s.io_v1_customresourcedefinition_otelexporterendpoints.formance.com.yaml new file mode 100644 index 000000000..8f333e743 --- /dev/null +++ b/helm/crds/templates/crds/apiextensions.k8s.io_v1_customresourcedefinition_otelexporterendpoints.formance.com.yaml @@ -0,0 +1,274 @@ +apiVersion: apiextensions.k8s.io/v1 +kind: CustomResourceDefinition +metadata: + annotations: + controller-gen.kubebuilder.io/version: v0.18.0 + helm.sh/resource-policy: keep + {{- with .Values.annotations }} + {{- toYaml . | nindent 4 }} + {{- end }} + name: otelexporterendpoints.formance.com +spec: + group: formance.com + names: + kind: OtelExporterEndpoint + listKind: OtelExporterEndpointList + plural: otelexporterendpoints + singular: otelexporterendpoint + scope: Cluster + versions: + - additionalPrinterColumns: + - description: Is ready + jsonPath: .status.ready + name: Ready + type: string + - description: Info + jsonPath: .status.info + name: Info + type: string + name: v1beta1 + schema: + openAPIV3Schema: + description: |- + OtelExporterEndpoint configures an OpenTelemetry collector proxy for exporting traces and metrics. + Multiple OtelExporterEndpoints can target the same stacks — the collector fans out to all matching destinations. + properties: + apiVersion: + description: |- + APIVersion defines the versioned schema of this representation of an object. + Servers should convert recognized schemas to the latest internal value, and + may reject unrecognized values. + More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources + type: string + kind: + description: |- + Kind is a string value representing the REST resource this object represents. + Servers may infer this from the endpoint the client submits requests to. + Cannot be updated. + In CamelCase. + More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds + type: string + metadata: + type: object + spec: + properties: + metrics: + description: |- + Metrics configures the metrics signal. At least one of traces or metrics must be set. + Logs are intentionally out of scope. + properties: + auth: + description: Auth is the optional per-signal authentication configuration. + properties: + fromSecret: + description: |- + FromSecret references a Secret name. + The controller creates a ResourceReference to replicate the secret into each target stack namespace. + The source secret must have a "formance.com/stack" label set to "any" or a specific stack name. + minLength: 1 + type: string + fromSecretKey: + default: token + description: FromSecretKey is the key within the Secret that + contains the token. Defaults to "token". + type: string + type: + description: Type is the authentication type. + enum: + - bearer + type: string + required: + - fromSecret + - type + type: object + endpoint: + description: |- + Endpoint URL for the signal (e.g., "http://my-collector:4318", "grpc://my-collector:4317"). + Supported schemes: http, https, grpc. Bare host:port is treated as HTTP. + Protocol is inferred from the URL scheme. HTTP/protobuf is the default for firewall compatibility. + minLength: 1 + pattern: ^(https?://|grpc://|[a-zA-Z0-9]) + type: string + required: + - endpoint + type: object + resourceAttributes: + additionalProperties: + type: string + description: ResourceAttributes are injected into outgoing telemetry + via a collector processor. + type: object + stackSelector: + description: |- + StackSelector is a standard Kubernetes LabelSelector (matchLabels/matchExpressions). + One CRD can target all current and future stacks with a single selector. + Matches the pattern established by Settings. + properties: + matchExpressions: + description: matchExpressions is a list of label selector requirements. + The requirements are ANDed. + items: + description: |- + A label selector requirement is a selector that contains values, a key, and an operator that + relates the key and values. + properties: + key: + description: key is the label key that the selector applies + to. + type: string + operator: + description: |- + operator represents a key's relationship to a set of values. + Valid operators are In, NotIn, Exists and DoesNotExist. + type: string + values: + description: |- + values is an array of string values. If the operator is In or NotIn, + the values array must be non-empty. If the operator is Exists or DoesNotExist, + the values array must be empty. This array is replaced during a strategic + merge patch. + items: + type: string + type: array + x-kubernetes-list-type: atomic + required: + - key + - operator + type: object + type: array + x-kubernetes-list-type: atomic + matchLabels: + additionalProperties: + type: string + description: |- + matchLabels is a map of {key,value} pairs. A single {key,value} in the matchLabels + map is equivalent to an element of matchExpressions, whose key field is "key", the + operator is "In", and the values array contains only "value". The requirements are ANDed. + type: object + type: object + x-kubernetes-map-type: atomic + traces: + description: |- + Traces configures the traces signal. At least one of traces or metrics must be set. + Logs are intentionally out of scope. + properties: + auth: + description: Auth is the optional per-signal authentication configuration. + properties: + fromSecret: + description: |- + FromSecret references a Secret name. + The controller creates a ResourceReference to replicate the secret into each target stack namespace. + The source secret must have a "formance.com/stack" label set to "any" or a specific stack name. + minLength: 1 + type: string + fromSecretKey: + default: token + description: FromSecretKey is the key within the Secret that + contains the token. Defaults to "token". + type: string + type: + description: Type is the authentication type. + enum: + - bearer + type: string + required: + - fromSecret + - type + type: object + endpoint: + description: |- + Endpoint URL for the signal (e.g., "http://my-collector:4318", "grpc://my-collector:4317"). + Supported schemes: http, https, grpc. Bare host:port is treated as HTTP. + Protocol is inferred from the URL scheme. HTTP/protobuf is the default for firewall compatibility. + minLength: 1 + pattern: ^(https?://|grpc://|[a-zA-Z0-9]) + type: string + required: + - endpoint + type: object + required: + - stackSelector + type: object + x-kubernetes-validations: + - message: at least one signal (traces or metrics) must be configured + rule: has(self.traces) || has(self.metrics) + status: + description: OtelExporterEndpointStatus represents the observed state + of an OtelExporterEndpoint. + properties: + conditions: + items: + description: Condition contains details for one aspect of the current + state of this API Resource. + properties: + lastTransitionTime: + description: |- + lastTransitionTime is the last time the condition transitioned from one status to another. + This should be when the underlying condition changed. If that is not known, then using the time when the API field changed is acceptable. + format: date-time + type: string + message: + description: |- + message is a human readable message indicating details about the transition. + This may be an empty string. + maxLength: 32768 + type: string + observedGeneration: + description: |- + observedGeneration represents the .metadata.generation that the condition was set based upon. + For instance, if .metadata.generation is currently 12, but the .status.conditions[x].observedGeneration is 9, the condition is out of date + with respect to the current state of the instance. + format: int64 + minimum: 0 + type: integer + reason: + description: |- + reason contains a programmatic identifier indicating the reason for the condition's last transition. + Producers of specific condition types may define expected values and meanings for this field, + and whether the values are considered a guaranteed API. + The value should be a CamelCase string. + This field may not be empty. + maxLength: 1024 + pattern: ^([A-Za-z]([A-Za-z0-9_,:]*[A-Za-z0-9_])?)?$ + type: string + status: + description: status of the condition, one of True, False, Unknown. + enum: + - "True" + - "False" + - Unknown + type: string + type: + description: type of condition in CamelCase or in foo.example.com/CamelCase. + maxLength: 316 + pattern: ^([a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*/)?(([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9])$ + type: string + required: + - lastTransitionTime + - message + - status + - type + type: object + type: array + info: + description: Info can contain any additional like reconciliation errors + type: string + ready: + description: Ready indicates if the resource is seen as completely + reconciled + type: boolean + stacks: + description: |- + Stacks is a sorted list of stack names currently targeted by this endpoint. + Includes stacks with successful reconciliation and stacks with transient errors or pending cleanup. + Used by the finalizer to find previously matched stacks during deletion. + items: + type: string + type: array + type: object + type: object + served: true + storage: true + subresources: + status: {} diff --git a/helm/operator/templates/deployment.yaml b/helm/operator/templates/deployment.yaml index 57c3db1a0..47a93de79 100644 --- a/helm/operator/templates/deployment.yaml +++ b/helm/operator/templates/deployment.yaml @@ -63,6 +63,9 @@ spec: - --disable-webhooks {{- end }} - --utils-version={{ .Values.operator.utils.tag | default .Chart.AppVersion }} + {{- with .Values.operator.collectorImage }} + - --collector-image={{ . }} + {{- end }} {{- if .Values.operator.dev }} - --zap-devel - Development diff --git a/helm/operator/templates/gen/rbac.authorization.k8s.io_v1_clusterrole_formance-manager-role.yaml b/helm/operator/templates/gen/rbac.authorization.k8s.io_v1_clusterrole_formance-manager-role.yaml index 24e51fa55..16f6c6e34 100644 --- a/helm/operator/templates/gen/rbac.authorization.k8s.io_v1_clusterrole_formance-manager-role.yaml +++ b/helm/operator/templates/gen/rbac.authorization.k8s.io_v1_clusterrole_formance-manager-role.yaml @@ -93,6 +93,7 @@ rules: - ledgers - mcps - orchestrations + - otelexporterendpoints - payments - reconciliations - resourcereferences @@ -128,6 +129,7 @@ rules: - ledgers/finalizers - mcps/finalizers - orchestrations/finalizers + - otelexporterendpoints/finalizers - payments/finalizers - reconciliations/finalizers - resourcereferences/finalizers @@ -157,6 +159,7 @@ rules: - ledgers/status - mcps/status - orchestrations/status + - otelexporterendpoints/status - payments/status - reconciliations/status - resourcereferences/status diff --git a/helm/operator/templates/otelexporterendpoint-support.yaml b/helm/operator/templates/otelexporterendpoint-support.yaml new file mode 100644 index 000000000..7e7a5755a --- /dev/null +++ b/helm/operator/templates/otelexporterendpoint-support.yaml @@ -0,0 +1,43 @@ +{{- if .Values.global.monitoring.support.enabled }} +{{- $authSecret := .Values.global.monitoring.support.authSecret }} +{{- if not $authSecret }} + {{- if .Values.global.licence.existingSecret }} + {{- $authSecret = .Values.global.licence.existingSecret }} + {{- else if and .Values.global.licence.createSecret .Values.global.licence.token }} + {{- $authSecret = (printf "%s-licence" (include "operator.fullname" .)) }} + {{- end }} +{{- end }} +apiVersion: formance.com/v1beta1 +kind: OtelExporterEndpoint +metadata: + name: formance-support + labels: + {{- include "operator.labels" . | nindent 4 }} +spec: + stackSelector: + {{- toYaml .Values.global.monitoring.support.stackSelector | nindent 4 }} + traces: + endpoint: {{ .Values.global.monitoring.support.endpoint | quote }} + {{- if $authSecret }} + auth: + type: bearer + fromSecret: {{ $authSecret | quote }} + {{- with .Values.global.monitoring.support.authSecretKey }} + fromSecretKey: {{ . | quote }} + {{- end }} + {{- end }} + metrics: + endpoint: {{ .Values.global.monitoring.support.endpoint | quote }} + {{- if $authSecret }} + auth: + type: bearer + fromSecret: {{ $authSecret | quote }} + {{- with .Values.global.monitoring.support.authSecretKey }} + fromSecretKey: {{ . | quote }} + {{- end }} + {{- end }} + {{- with .Values.global.monitoring.support.resourceAttributes }} + resourceAttributes: + {{- toYaml . | nindent 4 }} + {{- end }} +{{- end }} diff --git a/helm/operator/values.yaml b/helm/operator/values.yaml index 8e34b8cc9..07211b963 100644 --- a/helm/operator/values.yaml +++ b/helm/operator/values.yaml @@ -3,7 +3,26 @@ # Declare variables to be passed into your templates. global: - # Add your license information + monitoring: + support: + # Enable the Formance support OtelExporterEndpoint + enabled: false + # The support collector endpoint + endpoint: "https://otel-support.internal.frmnc.net" + # Secret name containing the bearer token for auth (key: "token"). + # Defaults to the licence secret if set. + authSecret: "" + # Key within the auth secret that contains the token. Defaults to "token" in the CRD. + authSecretKey: "" + # Resource attributes added to all signals sent to this endpoint + resourceAttributes: {} + # Stack selector for the support OtelExporterEndpoint + stackSelector: + matchExpressions: + - key: formance.com/stack + operator: Exists + + # Add your license information licence: # -- Is disabled automatically if an existingSecret is provided # @section -- Licence @@ -55,6 +74,9 @@ operator: utils: tag: "" + # Override the OTel Collector image used by OtelExporterEndpoint resources + collectorImage: "" + podAnnotations: {} podSecurityContext: {} # fsGroup: 2000 diff --git a/internal/core/platform.go b/internal/core/platform.go index 31152589a..f66dc4429 100644 --- a/internal/core/platform.go +++ b/internal/core/platform.go @@ -1,5 +1,15 @@ package core +const ( + DefaultCollectorImage = "otel/opentelemetry-collector-contrib:0.151.0" + + SignalTracesAnnotation = "formance.com/otel-traces-enabled" + SignalMetricsAnnotation = "formance.com/otel-metrics-enabled" + + CollectorManagedByLabel = "formance.com/managed-by" + CollectorManagedByValue = "otelexporterendpoint" +) + type Platform struct { // Cloud region where the stack is deployed Region string @@ -16,4 +26,6 @@ type Platform struct { LicenceState LicenceState // Human-readable message about the licence state LicenceMessage string + // The OTel Collector image used by OtelExporterEndpoint resources + CollectorImage string } diff --git a/internal/resources/all.go b/internal/resources/all.go index 619fde7af..a1dc3e5e9 100644 --- a/internal/resources/all.go +++ b/internal/resources/all.go @@ -13,6 +13,7 @@ import ( _ "github.com/formancehq/operator/v3/internal/resources/ledgers" _ "github.com/formancehq/operator/v3/internal/resources/mcps" _ "github.com/formancehq/operator/v3/internal/resources/orchestrations" + _ "github.com/formancehq/operator/v3/internal/resources/otelexporterendpoints" _ "github.com/formancehq/operator/v3/internal/resources/payments" _ "github.com/formancehq/operator/v3/internal/resources/reconciliations" _ "github.com/formancehq/operator/v3/internal/resources/resourcereferences" diff --git a/internal/resources/otelexporterendpoints/collector_config.go b/internal/resources/otelexporterendpoints/collector_config.go new file mode 100644 index 000000000..dd8f6963c --- /dev/null +++ b/internal/resources/otelexporterendpoints/collector_config.go @@ -0,0 +1,347 @@ +package otelexporterendpoints + +import ( + "crypto/sha256" + "encoding/hex" + "fmt" + "net/url" + "slices" + "sort" + "strings" + + "gopkg.in/yaml.v3" + + "github.com/formancehq/operator/v3/api/formance.com/v1beta1" +) + +type collectorConfig struct { + Extensions collectorExtensions `yaml:"extensions"` + Receivers collectorReceivers `yaml:"receivers"` + Processors map[string]any `yaml:"processors,omitempty"` + Exporters map[string]any `yaml:"exporters"` + Service collectorService `yaml:"service"` +} + +type collectorExtensions struct { + HealthCheck healthCheckExtension `yaml:"health_check"` +} + +type healthCheckExtension struct { + Endpoint string `yaml:"endpoint"` +} + +type collectorReceivers struct { + OTLP otlpReceiver `yaml:"otlp"` +} + +type otlpReceiver struct { + Protocols otlpProtocols `yaml:"protocols"` +} + +type otlpProtocols struct { + HTTP otlpHTTP `yaml:"http"` +} + +type otlpHTTP struct { + Endpoint string `yaml:"endpoint"` +} + +type collectorService struct { + Extensions []string `yaml:"extensions"` + Pipelines map[string]collectorPipeline `yaml:"pipelines"` +} + +type collectorPipeline struct { + Receivers []string `yaml:"receivers"` + Processors []string `yaml:"processors,omitempty"` + Exporters []string `yaml:"exporters"` +} + +type otlpExporter struct { + Endpoint string `yaml:"endpoint"` + Headers map[string]string `yaml:"headers,omitempty"` + TLS *otlpTLSConfig `yaml:"tls,omitempty"` +} + +type otlpTLSConfig struct { + Insecure bool `yaml:"insecure"` +} + +type resourceProcessorAttribute struct { + Key string `yaml:"key"` + Value string `yaml:"value"` + Action string `yaml:"action"` +} + +type resourceProcessor struct { + Attributes []resourceProcessorAttribute `yaml:"attributes"` +} + +type exporterInput struct { + name string + signal *v1beta1.OtelSignalConfig + envAlias string +} + +func inferProtocol(endpoint string) string { + u, err := url.Parse(endpoint) + if err == nil && u.Scheme == "grpc" { + return "grpc" + } + return "http" +} + +func stripScheme(endpoint string) string { + u, err := url.Parse(endpoint) + if err != nil { + return endpoint + } + if u.Scheme == "grpc" { + return u.Host + } + return endpoint +} + +func isInsecure(endpoint string) bool { + u, err := url.Parse(endpoint) + if err != nil { + return false + } + return u.Query().Get("insecure") == "true" +} + +func buildExporter(input exporterInput) (string, any) { + protocol := inferProtocol(input.signal.Endpoint) + endpoint := stripScheme(input.signal.Endpoint) + + var headers map[string]string + if input.signal.Auth != nil && input.signal.Auth.Type == "bearer" { + headers = map[string]string{ + "authorization": fmt.Sprintf("Bearer ${env:%s}", input.envAlias), + } + } + + var tls *otlpTLSConfig + if protocol == "grpc" && isInsecure(input.signal.Endpoint) { + tls = &otlpTLSConfig{Insecure: true} + } + + prefix := "otlphttp/" + if protocol == "grpc" { + prefix = "otlp/" + } + return prefix + input.name, otlpExporter{ + Endpoint: endpoint, + Headers: headers, + TLS: tls, + } +} + +type collectorInput struct { + Endpoint *v1beta1.OtelExporterEndpoint + TracesEnvAlias string + MetricsEnvAlias string +} + +type otelSettingsInput struct { + TracesEndpoint string + MetricsEndpoint string +} + +func generateMergedCollectorConfig(endpoints []collectorInput, otelSettings *otelSettingsInput) (string, error) { + exporters := map[string]any{} + processors := map[string]any{} + var tracesPipelines []pipelineContribution + var metricsPipelines []pipelineContribution + + sortedEndpoints := make([]collectorInput, len(endpoints)) + copy(sortedEndpoints, endpoints) + sort.Slice(sortedEndpoints, func(i, j int) bool { + return sortedEndpoints[i].Endpoint.Name < sortedEndpoints[j].Endpoint.Name + }) + + for _, ci := range sortedEndpoints { + ep := ci.Endpoint + crdName := sanitizeName(ep.Name) + + var resourceProc string + if len(ep.Spec.ResourceAttributes) > 0 { + procName := "resource/" + crdName + attrs := make([]resourceProcessorAttribute, 0, len(ep.Spec.ResourceAttributes)) + keys := make([]string, 0, len(ep.Spec.ResourceAttributes)) + for k := range ep.Spec.ResourceAttributes { + keys = append(keys, k) + } + slices.Sort(keys) + for _, k := range keys { + attrs = append(attrs, resourceProcessorAttribute{ + Key: k, + Value: ep.Spec.ResourceAttributes[k], + Action: "upsert", + }) + } + processors[procName] = resourceProcessor{Attributes: attrs} + resourceProc = procName + } + + if ep.Spec.Traces != nil && ep.Spec.Traces.Endpoint != "" { + name, exp := buildExporter(exporterInput{ + name: crdName + "-traces", + signal: ep.Spec.Traces, + envAlias: ci.TracesEnvAlias, + }) + exporters[name] = exp + tracesPipelines = append(tracesPipelines, pipelineContribution{ + exporter: name, + processor: resourceProc, + }) + } + + if ep.Spec.Metrics != nil && ep.Spec.Metrics.Endpoint != "" { + name, exp := buildExporter(exporterInput{ + name: crdName + "-metrics", + signal: ep.Spec.Metrics, + envAlias: ci.MetricsEnvAlias, + }) + exporters[name] = exp + metricsPipelines = append(metricsPipelines, pipelineContribution{ + exporter: name, + processor: resourceProc, + }) + } + } + + if otelSettings != nil { + if otelSettings.TracesEndpoint != "" { + name, exp := buildExporter(exporterInput{ + name: "settings-traces", + signal: &v1beta1.OtelSignalConfig{Endpoint: otelSettings.TracesEndpoint}, + }) + exporters[name] = exp + tracesPipelines = append(tracesPipelines, pipelineContribution{exporter: name}) + } + if otelSettings.MetricsEndpoint != "" { + name, exp := buildExporter(exporterInput{ + name: "settings-metrics", + signal: &v1beta1.OtelSignalConfig{Endpoint: otelSettings.MetricsEndpoint}, + }) + exporters[name] = exp + metricsPipelines = append(metricsPipelines, pipelineContribution{exporter: name}) + } + } + + if len(tracesPipelines) == 0 { + exporters["nop"] = struct{}{} + tracesPipelines = []pipelineContribution{{exporter: "nop"}} + } + if len(metricsPipelines) == 0 { + exporters["nop"] = struct{}{} + metricsPipelines = []pipelineContribution{{exporter: "nop"}} + } + + pipelines := buildPipelines(tracesPipelines, metricsPipelines) + + cfg := collectorConfig{ + Extensions: collectorExtensions{ + HealthCheck: healthCheckExtension{ + Endpoint: fmt.Sprintf("0.0.0.0:%d", healthCheckPort), + }, + }, + Receivers: collectorReceivers{ + OTLP: otlpReceiver{ + Protocols: otlpProtocols{ + HTTP: otlpHTTP{ + Endpoint: "0.0.0.0:4318", + }, + }, + }, + }, + Exporters: exporters, + Processors: processors, + Service: collectorService{ + Extensions: []string{"health_check"}, + Pipelines: pipelines, + }, + } + + if len(processors) == 0 { + cfg.Processors = nil + } + + data, err := yaml.Marshal(cfg) + if err != nil { + return "", err + } + + return string(data), nil +} + +type pipelineContribution struct { + exporter string + processor string +} + +func buildPipelines(traces, metrics []pipelineContribution) map[string]collectorPipeline { + pipelines := map[string]collectorPipeline{} + addSignalPipelines(pipelines, "traces", traces) + addSignalPipelines(pipelines, "metrics", metrics) + return pipelines +} + +func addSignalPipelines(pipelines map[string]collectorPipeline, signal string, contributions []pipelineContribution) { + grouped := groupByProcessor(contributions) + if len(grouped) == 1 { + for proc, exporters := range grouped { + p := collectorPipeline{ + Receivers: []string{"otlp"}, + Exporters: exporters, + } + if proc != "" { + p.Processors = []string{proc} + } + pipelines[signal] = p + } + return + } + procs := make([]string, 0, len(grouped)) + for proc := range grouped { + procs = append(procs, proc) + } + slices.Sort(procs) + + for _, proc := range procs { + exporterList := grouped[proc] + suffix := "default" + if proc != "" { + parts := strings.SplitN(proc, "/", 2) + if len(parts) == 2 { + suffix = parts[1] + } + } + p := collectorPipeline{ + Receivers: []string{"otlp"}, + Exporters: exporterList, + } + if proc != "" { + p.Processors = []string{proc} + } + pipelines[signal+"/"+suffix] = p + } +} + +func groupByProcessor(contributions []pipelineContribution) map[string][]string { + grouped := map[string][]string{} + for _, c := range contributions { + grouped[c.processor] = append(grouped[c.processor], c.exporter) + } + return grouped +} + +func sanitizeName(name string) string { + sanitized := strings.ReplaceAll(name, ".", "-") + if sanitized != name { + h := sha256.Sum256([]byte(name)) + sanitized = sanitized + "-" + hex.EncodeToString(h[:3]) + } + return sanitized +} diff --git a/internal/resources/otelexporterendpoints/collector_config_test.go b/internal/resources/otelexporterendpoints/collector_config_test.go new file mode 100644 index 000000000..f6060579d --- /dev/null +++ b/internal/resources/otelexporterendpoints/collector_config_test.go @@ -0,0 +1,363 @@ +package otelexporterendpoints + +import ( + "strings" + "testing" + + "github.com/stretchr/testify/require" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + "github.com/formancehq/operator/v3/api/formance.com/v1beta1" +) + +func endpoint(name string, spec v1beta1.OtelExporterEndpointSpec) *v1beta1.OtelExporterEndpoint { + return &v1beta1.OtelExporterEndpoint{ + ObjectMeta: metav1.ObjectMeta{Name: name}, + Spec: spec, + } +} + +func TestGenerateMergedCollectorConfig(t *testing.T) { + t.Parallel() + + type testCase struct { + name string + inputs []collectorInput + otelSettings *otelSettingsInput + expectedContains []string + expectedNotContains []string + } + + testCases := []testCase{ + { + name: "single CRD with traces endpoint", + inputs: []collectorInput{ + { + Endpoint: endpoint("monitoring", v1beta1.OtelExporterEndpointSpec{ + Traces: &v1beta1.OtelSignalConfig{ + Endpoint: "http://my-collector:4318", + }, + }), + }, + }, + expectedContains: []string{ + "otlphttp/monitoring-traces", + "http://my-collector:4318", + "nop", + "health_check", + "13133", + }, + }, + { + name: "single CRD with grpc endpoint", + inputs: []collectorInput{ + { + Endpoint: endpoint("monitoring", v1beta1.OtelExporterEndpointSpec{ + Traces: &v1beta1.OtelSignalConfig{ + Endpoint: "grpc://my-collector:4317", + }, + }), + }, + }, + expectedContains: []string{ + "otlp/monitoring-traces", + "my-collector:4317", + }, + expectedNotContains: []string{"otlphttp/monitoring-traces"}, + }, + { + name: "single CRD with auth", + inputs: []collectorInput{ + { + Endpoint: endpoint("support", v1beta1.OtelExporterEndpointSpec{ + Traces: &v1beta1.OtelSignalConfig{ + Endpoint: "https://support.frmnc.net", + Auth: &v1beta1.OtelExporterAuth{ + Type: "bearer", + FromSecret: "formance-license", + }, + }, + }), + TracesEnvAlias: "AUTH_SUPPORT_TRACES", + }, + }, + expectedContains: []string{ + "otlphttp/support-traces", + "https://support.frmnc.net", + "authorization: Bearer ${env:AUTH_SUPPORT_TRACES}", + }, + }, + { + name: "multiple CRDs fan out", + inputs: []collectorInput{ + { + Endpoint: endpoint("monitoring", v1beta1.OtelExporterEndpointSpec{ + Traces: &v1beta1.OtelSignalConfig{ + Endpoint: "http://my-collector:4318", + }, + }), + }, + { + Endpoint: endpoint("support", v1beta1.OtelExporterEndpointSpec{ + Traces: &v1beta1.OtelSignalConfig{ + Endpoint: "https://support.frmnc.net", + Auth: &v1beta1.OtelExporterAuth{ + Type: "bearer", + FromSecret: "formance-license", + }, + }, + }), + TracesEnvAlias: "AUTH_SUPPORT_TRACES", + }, + }, + expectedContains: []string{ + "otlphttp/monitoring-traces", + "http://my-collector:4318", + "otlphttp/support-traces", + "https://support.frmnc.net", + "authorization: Bearer ${env:AUTH_SUPPORT_TRACES}", + }, + }, + { + name: "grpc endpoint with insecure query param", + inputs: []collectorInput{ + { + Endpoint: endpoint("monitoring", v1beta1.OtelExporterEndpointSpec{ + Traces: &v1beta1.OtelSignalConfig{ + Endpoint: "grpc://my-collector:4317?insecure=true", + }, + }), + }, + }, + expectedContains: []string{ + "otlp/monitoring-traces", + "my-collector:4317", + "tls:", + "insecure: true", + }, + expectedNotContains: []string{"otlphttp/monitoring-traces"}, + }, + { + name: "grpc endpoint without insecure has no tls config", + inputs: []collectorInput{ + { + Endpoint: endpoint("monitoring", v1beta1.OtelExporterEndpointSpec{ + Traces: &v1beta1.OtelSignalConfig{ + Endpoint: "grpc://my-collector:4317", + }, + }), + }, + }, + expectedContains: []string{"otlp/monitoring-traces", "my-collector:4317"}, + expectedNotContains: []string{"tls:"}, + }, + { + name: "otel settings grpc with insecure", + inputs: []collectorInput{}, + otelSettings: &otelSettingsInput{ + TracesEndpoint: "grpc://settings-collector:4317?insecure=true", + MetricsEndpoint: "grpc://settings-collector:4317?insecure=true", + }, + expectedContains: []string{ + "otlp/settings-traces", + "otlp/settings-metrics", + "settings-collector:4317", + "tls:", + "insecure: true", + }, + }, + { + name: "no endpoints produces nop", + inputs: []collectorInput{}, + expectedContains: []string{"nop"}, + expectedNotContains: []string{"otlphttp", "otlp/"}, + }, + { + name: "otel settings traces", + inputs: []collectorInput{}, + otelSettings: &otelSettingsInput{ + TracesEndpoint: "http://settings-collector:4318", + }, + expectedContains: []string{ + "otlphttp/settings-traces", + "http://settings-collector:4318", + }, + }, + { + name: "CRD plus otel settings both appear", + inputs: []collectorInput{ + { + Endpoint: endpoint("monitoring", v1beta1.OtelExporterEndpointSpec{ + Traces: &v1beta1.OtelSignalConfig{ + Endpoint: "http://my-collector:4318", + }, + }), + }, + }, + otelSettings: &otelSettingsInput{ + TracesEndpoint: "http://settings-collector:4318", + }, + expectedContains: []string{ + "otlphttp/monitoring-traces", + "otlphttp/settings-traces", + }, + }, + { + name: "resource attributes produce processor", + inputs: []collectorInput{ + { + Endpoint: endpoint("support", v1beta1.OtelExporterEndpointSpec{ + Traces: &v1beta1.OtelSignalConfig{ + Endpoint: "https://support.frmnc.net", + }, + ResourceAttributes: map[string]string{ + "cluster.id": "abc-123", + }, + }), + }, + }, + expectedContains: []string{ + "resource/support", + "cluster.id", + "abc-123", + "upsert", + }, + }, + { + name: "traces and metrics with separate endpoints", + inputs: []collectorInput{ + { + Endpoint: endpoint("monitoring", v1beta1.OtelExporterEndpointSpec{ + Traces: &v1beta1.OtelSignalConfig{ + Endpoint: "http://traces-collector:4318", + }, + Metrics: &v1beta1.OtelSignalConfig{ + Endpoint: "http://metrics-collector:4318", + }, + }), + }, + }, + expectedContains: []string{ + "otlphttp/monitoring-traces", + "http://traces-collector:4318", + "otlphttp/monitoring-metrics", + "http://metrics-collector:4318", + }, + expectedNotContains: []string{"nop"}, + }, + { + name: "metrics-only uses nop for traces", + inputs: []collectorInput{ + { + Endpoint: endpoint("monitoring", v1beta1.OtelExporterEndpointSpec{ + Metrics: &v1beta1.OtelSignalConfig{ + Endpoint: "http://metrics-collector:4318", + }, + }), + }, + }, + expectedContains: []string{ + "otlphttp/monitoring-metrics", + "http://metrics-collector:4318", + "nop", + }, + expectedNotContains: []string{"otlphttp/monitoring-traces"}, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + config, err := generateMergedCollectorConfig(tc.inputs, tc.otelSettings) + require.NoError(t, err) + + for _, s := range tc.expectedContains { + require.True(t, strings.Contains(config, s), + "expected config to contain %q, got:\n%s", s, config) + } + for _, s := range tc.expectedNotContains { + require.False(t, strings.Contains(config, s), + "expected config NOT to contain %q, got:\n%s", s, config) + } + }) + } +} + +func TestInferProtocol(t *testing.T) { + t.Parallel() + + require.Equal(t, "grpc", inferProtocol("grpc://my-collector:4317")) + require.Equal(t, "http", inferProtocol("http://my-collector:4318")) + require.Equal(t, "http", inferProtocol("https://support.frmnc.net")) + require.Equal(t, "http", inferProtocol("my-collector:4318")) +} + +func TestIsInsecure(t *testing.T) { + t.Parallel() + + require.True(t, isInsecure("grpc://my-collector:4317?insecure=true")) + require.False(t, isInsecure("grpc://my-collector:4317")) + require.False(t, isInsecure("grpc://my-collector:4317?insecure=false")) + require.False(t, isInsecure("http://my-collector:4318")) + require.False(t, isInsecure("https://support.frmnc.net")) +} + +func TestStripScheme(t *testing.T) { + t.Parallel() + + require.Equal(t, "my-collector:4317", stripScheme("grpc://my-collector:4317")) + require.Equal(t, "http://my-collector:4318", stripScheme("http://my-collector:4318")) + require.Equal(t, "https://support.frmnc.net", stripScheme("https://support.frmnc.net")) +} + +func TestSanitizeName(t *testing.T) { + t.Parallel() + + require.Equal(t, "my-endpoint", sanitizeName("my-endpoint")) + require.Contains(t, sanitizeName("my.endpoint"), "my-endpoint-") + require.NotEqual(t, sanitizeName("my-endpoint"), sanitizeName("my.endpoint")) + + require.NotEqual(t, sanitizeName("a.b-c"), sanitizeName("a-b.c")) +} + +func TestEnvSafe(t *testing.T) { + t.Parallel() + + require.Equal(t, "FORMANCE_SUPPORT", envSafe("formance-support")) + require.Equal(t, "MY_MONITORING", envSafe("my-monitoring")) + require.Equal(t, "TEST_123", envSafe("test.123")) +} + +func TestBuildCollectorInputs(t *testing.T) { + t.Parallel() + + endpoints := []v1beta1.OtelExporterEndpoint{ + { + ObjectMeta: metav1.ObjectMeta{Name: "support"}, + Spec: v1beta1.OtelExporterEndpointSpec{ + Traces: &v1beta1.OtelSignalConfig{ + Endpoint: "https://support.frmnc.net", + Auth: &v1beta1.OtelExporterAuth{ + Type: "bearer", + FromSecret: "formance-license", + }, + }, + }, + }, + { + ObjectMeta: metav1.ObjectMeta{Name: "monitoring"}, + Spec: v1beta1.OtelExporterEndpointSpec{ + Traces: &v1beta1.OtelSignalConfig{ + Endpoint: "http://my-collector:4318", + }, + }, + }, + } + + inputs, envVars := buildCollectorInputs(endpoints) + require.Len(t, inputs, 2) + require.Len(t, envVars, 1) + require.Equal(t, "AUTH_SUPPORT_TRACES", envVars[0].Name) + require.Equal(t, "formance-license", envVars[0].ValueFrom.SecretKeyRef.LocalObjectReference.Name) +} diff --git a/internal/resources/otelexporterendpoints/init.go b/internal/resources/otelexporterendpoints/init.go new file mode 100644 index 000000000..41dd2ef1b --- /dev/null +++ b/internal/resources/otelexporterendpoints/init.go @@ -0,0 +1,904 @@ +/* +Copyright 2023. + +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 otelexporterendpoints + +import ( + "context" + "crypto/sha256" + "encoding/base64" + "encoding/hex" + "errors" + "fmt" + "reflect" + "slices" + "sort" + "strings" + + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/resource" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/types" + "k8s.io/apimachinery/pkg/util/intstr" + "sigs.k8s.io/controller-runtime/pkg/builder" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/handler" + "sigs.k8s.io/controller-runtime/pkg/log" + "sigs.k8s.io/controller-runtime/pkg/predicate" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + + collectionutils "github.com/formancehq/go-libs/v5/pkg/types/collections" + "github.com/formancehq/go-libs/v5/pkg/types/pointer" + + v1beta1 "github.com/formancehq/operator/v3/api/formance.com/v1beta1" + . "github.com/formancehq/operator/v3/internal/core" + "github.com/formancehq/operator/v3/internal/resources/settings" +) + +//+kubebuilder:rbac:groups=formance.com,resources=otelexporterendpoints,verbs=get;list;watch;create;update;patch;delete +//+kubebuilder:rbac:groups=formance.com,resources=otelexporterendpoints/status,verbs=get;update;patch +//+kubebuilder:rbac:groups=formance.com,resources=otelexporterendpoints/finalizers,verbs=update + +const ( + deploymentName = "otel-collector" + serviceName = "otel-collector" + collectorPort = 4318 + healthCheckPort = 13133 + + collectorFinalizer = "otelexporterendpoint.formance.com/finalizer" + + managedByLabel = CollectorManagedByLabel + managedByValue = CollectorManagedByValue + + collectorSignalAnnotation = "formance.com/otel-collector-signals" +) + +func collectorLabels() map[string]string { + return map[string]string{ + "app.kubernetes.io/name": deploymentName, + managedByLabel: managedByValue, + } +} + +func Reconcile(ctx Context, endpoint *v1beta1.OtelExporterEndpoint) error { + selector, err := selectorFromSpec(endpoint.Spec.StackSelector) + if err != nil { + return err + } + + var stacks v1beta1.StackList + if err := ctx.GetClient().List(ctx, &stacks, client.MatchingLabelsSelector{Selector: selector}); err != nil { + return err + } + + logger := log.FromContext(ctx) + targetedStacks := map[string]bool{} + stackNames := make([]string, 0, len(stacks.Items)) + var stackErrors []string + var pendingStacks []string + for i := range stacks.Items { + stack := &stacks.Items[i] + if !stack.GetDeletionTimestamp().IsZero() { + continue + } + targetedStacks[stack.Name] = true + stackNames = append(stackNames, stack.Name) + if err := reconcileStackCollector(ctx, stack); err != nil { + if IsApplicationError(err) { + pendingStacks = append(pendingStacks, stack.Name) + } else { + logger.Error(err, "skipping stack due to reconciliation error", "stack", stack.Name) + stackErrors = append(stackErrors, fmt.Sprintf("%s: %s", stack.Name, err.Error())) + } + continue + } + + if reason := checkCollectorReady(ctx, stack.Name); reason != "" { + pendingStacks = append(pendingStacks, fmt.Sprintf("%s (%s)", stack.Name, reason)) + } + } + + for _, prev := range endpoint.Status.Stacks { + if !targetedStacks[prev] { + if err := cleanupStackCollector(ctx, prev); err != nil { + logger.Error(err, "failed to clean up orphaned collector", "stack", prev) + stackNames = append(stackNames, prev) + stackErrors = append(stackErrors, fmt.Sprintf("%s cleanup: %s", prev, err.Error())) + } + } + } + + slices.Sort(stackNames) + endpoint.Status.Stacks = slices.Compact(stackNames) + + if len(stackErrors) > 0 { + endpoint.SetError(fmt.Sprintf("errors in stacks: %s", strings.Join(stackErrors, "; "))) + } else { + endpoint.SetError("") + } + + condition := v1beta1.NewCondition("CollectorsReady", endpoint.GetGeneration()) + if len(pendingStacks) > 0 { + condition.Fail(fmt.Sprintf("waiting for collectors in: %s", strings.Join(pendingStacks, ", "))) + } + endpoint.GetConditions().AppendOrReplace(*condition, func(c v1beta1.Condition) bool { + return c.Type == "CollectorsReady" + }) + + if len(stackErrors) > 0 { + return fmt.Errorf("partial failure: %s", strings.Join(stackErrors, "; ")) + } + return nil +} + +func checkCollectorReady(ctx Context, stackName string) string { + deployment := &appsv1.Deployment{} + err := ctx.GetClient().Get(ctx, types.NamespacedName{ + Namespace: stackName, + Name: deploymentName, + }, deployment) + if err != nil { + log.FromContext(ctx).V(1).Info("collector deployment not found", "stack", stackName, "error", err) + return "deployment not found" + } + if deployment.Status.ObservedGeneration != deployment.Generation { + return "waiting for rollout" + } + if deployment.Spec.Replicas != nil && deployment.Status.UpdatedReplicas < *deployment.Spec.Replicas { + return "waiting for updated replicas" + } + if deployment.Status.AvailableReplicas < deployment.Status.UpdatedReplicas { + for _, cond := range deployment.Status.Conditions { + if cond.Type == appsv1.DeploymentAvailable && cond.Status != corev1.ConditionTrue { + return fmt.Sprintf("not available: %s", cond.Message) + } + if cond.Type == appsv1.DeploymentProgressing && cond.Status == corev1.ConditionFalse { + return fmt.Sprintf("progress stalled: %s", cond.Message) + } + } + return "waiting for availability" + } + return "" +} + +func Cleanup(ctx Context, endpoint *v1beta1.OtelExporterEndpoint) error { + selector, err := selectorFromSpec(endpoint.Spec.StackSelector) + if err != nil { + return err + } + + var stacks v1beta1.StackList + if err := ctx.GetClient().List(ctx, &stacks, client.MatchingLabelsSelector{Selector: selector}); err != nil { + return err + } + + var errs []error + reconciledStacks := map[string]bool{} + for i := range stacks.Items { + if !stacks.Items[i].GetDeletionTimestamp().IsZero() { + continue + } + reconciledStacks[stacks.Items[i].Name] = true + if err := reconcileStackCollector(ctx, &stacks.Items[i]); err != nil { + errs = append(errs, err) + } + } + + for _, stackName := range endpoint.Status.Stacks { + if reconciledStacks[stackName] { + continue + } + stack := &v1beta1.Stack{} + if err := ctx.GetClient().Get(ctx, types.NamespacedName{Name: stackName}, stack); err != nil { + if client.IgnoreNotFound(err) == nil { + if err := cleanupStackCollector(ctx, stackName); err != nil { + errs = append(errs, err) + } + continue + } + errs = append(errs, err) + continue + } + if err := reconcileStackCollector(ctx, stack); err != nil { + errs = append(errs, err) + } + } + return errors.Join(errs...) +} + +func selectorFromSpec(ls *metav1.LabelSelector) (labels.Selector, error) { + if ls == nil { + return nil, fmt.Errorf("stackSelector is required") + } + return metav1.LabelSelectorAsSelector(ls) +} + +func hasTraces(ep *v1beta1.OtelExporterEndpoint) bool { + return ep.Spec.Traces != nil && ep.Spec.Traces.Endpoint != "" +} + +func hasMetrics(ep *v1beta1.OtelExporterEndpoint) bool { + return ep.Spec.Metrics != nil && ep.Spec.Metrics.Endpoint != "" +} + +func hasRealSignal(ep *v1beta1.OtelExporterEndpoint) bool { + return hasTraces(ep) || hasMetrics(ep) +} + +func reconcileStackCollector(ctx Context, stack *v1beta1.Stack) error { + endpoints, err := findMatchingEndpoints(ctx, stack) + if err != nil { + return err + } + + activeEndpoints := make([]v1beta1.OtelExporterEndpoint, 0, len(endpoints)) + for i := range endpoints { + if hasRealSignal(&endpoints[i]) { + activeEndpoints = append(activeEndpoints, endpoints[i]) + } + } + + otelSettings, err := readOtelSettings(ctx, stack.Name) + if err != nil { + return err + } + + hasSettingsSignal := otelSettings != nil && (otelSettings.TracesEndpoint != "" || otelSettings.MetricsEndpoint != "") + if len(activeEndpoints) == 0 && !hasSettingsSignal { + return cleanupStackCollector(ctx, stack.Name) + } + + if err := ensureNoConflict(ctx, stack.Name); err != nil { + return err + } + + if err := ensureAuthSecretReferences(ctx, stack, activeEndpoints); err != nil { + return err + } + + inputs, envVars := buildCollectorInputs(activeEndpoints) + + hasTraces, hasMetrics := computeActiveSignals(activeEndpoints, otelSettings) + + collectorConfigYAML, err := generateMergedCollectorConfig(inputs, otelSettings) + if err != nil { + return fmt.Errorf("generating collector config: %w", err) + } + + configMap, _, err := CreateOrUpdate(ctx, types.NamespacedName{ + Namespace: stack.Name, + Name: "otel-collector-config", + }, + func(cm *corev1.ConfigMap) error { + cm.Data = map[string]string{ + "otel-collector-config.yaml": collectorConfigYAML, + } + return nil + }, + WithOwner[*corev1.ConfigMap](ctx.GetScheme(), stack), + WithLabels[*corev1.ConfigMap](collectorLabels()), + ) + if err != nil { + return fmt.Errorf("creating collector configmap: %w", err) + } + + secretHashes, err := hashAuthSecrets(ctx, stack.Name, activeEndpoints) + if err != nil { + return err + } + + podAnnotations := map[string]string{ + "config-hash": HashFromConfigMaps(configMap), + } + if secretHashes != "" { + podAnnotations["secret-hash"] = secretHashes + } + + // Phase 1 (RFC-0008): replicas, resource limits, probes, security context, and scheduling + // are hardcoded. They will become configurable via Settings keys in a future phase. + replicas := int32(1) + _, _, err = CreateOrUpdate(ctx, types.NamespacedName{ + Namespace: stack.Name, + Name: deploymentName, + }, + func(deployment *appsv1.Deployment) error { + deployment.Spec = appsv1.DeploymentSpec{ + Replicas: &replicas, + Selector: &metav1.LabelSelector{ + MatchLabels: map[string]string{ + "app.kubernetes.io/name": deploymentName, + }, + }, + Template: corev1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{ + Labels: collectorLabels(), + Annotations: podAnnotations, + }, + Spec: corev1.PodSpec{ + Containers: []corev1.Container{{ + Name: "otel-collector", + Image: collectorImageForPlatform(ctx), + Args: []string{"--config=/etc/otel/otel-collector-config.yaml"}, + Env: envVars, + Ports: []corev1.ContainerPort{ + { + Name: "otlp-http", + ContainerPort: collectorPort, + Protocol: corev1.ProtocolTCP, + }, + { + Name: "health", + ContainerPort: healthCheckPort, + Protocol: corev1.ProtocolTCP, + }, + }, + Resources: corev1.ResourceRequirements{ + Requests: corev1.ResourceList{ + corev1.ResourceCPU: resource.MustParse("50m"), + corev1.ResourceMemory: resource.MustParse("64Mi"), + }, + Limits: corev1.ResourceList{ + corev1.ResourceCPU: resource.MustParse("200m"), + corev1.ResourceMemory: resource.MustParse("256Mi"), + }, + }, + LivenessProbe: &corev1.Probe{ + ProbeHandler: corev1.ProbeHandler{ + HTTPGet: &corev1.HTTPGetAction{ + Path: "/", + Port: intstr.FromInt32(healthCheckPort), + }, + }, + InitialDelaySeconds: 5, + PeriodSeconds: 10, + }, + ReadinessProbe: &corev1.Probe{ + ProbeHandler: corev1.ProbeHandler{ + HTTPGet: &corev1.HTTPGetAction{ + Path: "/", + Port: intstr.FromInt32(healthCheckPort), + }, + }, + InitialDelaySeconds: 3, + PeriodSeconds: 5, + }, + SecurityContext: &corev1.SecurityContext{ + Capabilities: &corev1.Capabilities{ + Drop: []corev1.Capability{"ALL"}, + }, + Privileged: pointer.For(false), + ReadOnlyRootFilesystem: pointer.For(true), + AllowPrivilegeEscalation: pointer.For(false), + RunAsNonRoot: pointer.For(true), + RunAsUser: pointer.For(int64(65534)), + }, + VolumeMounts: []corev1.VolumeMount{ + NewVolumeMount("config", "/etc/otel", true), + NewVolumeMount("tmp", "/tmp", false), + }, + }}, + Volumes: []corev1.Volume{ + { + Name: "config", + VolumeSource: corev1.VolumeSource{ + ConfigMap: &corev1.ConfigMapVolumeSource{ + LocalObjectReference: corev1.LocalObjectReference{ + Name: configMap.Name, + }, + }, + }, + }, + { + Name: "tmp", + VolumeSource: corev1.VolumeSource{ + EmptyDir: &corev1.EmptyDirVolumeSource{}, + }, + }, + }, + }, + }, + } + return nil + }, + WithOwner[*appsv1.Deployment](ctx.GetScheme(), stack), + WithLabels[*appsv1.Deployment](collectorLabels()), + ) + if err != nil { + return fmt.Errorf("creating collector deployment: %w", err) + } + + signalAnnotations := map[string]string{ + SignalTracesAnnotation: fmt.Sprintf("%t", hasTraces), + SignalMetricsAnnotation: fmt.Sprintf("%t", hasMetrics), + } + _, _, err = CreateOrUpdate(ctx, types.NamespacedName{ + Namespace: stack.Name, + Name: serviceName, + }, + func(svc *corev1.Service) error { + svc.Spec = corev1.ServiceSpec{ + Selector: map[string]string{ + "app.kubernetes.io/name": deploymentName, + }, + Ports: []corev1.ServicePort{{ + Name: "otlp-http", + Port: collectorPort, + Protocol: corev1.ProtocolTCP, + }}, + } + if svc.Annotations == nil { + svc.Annotations = map[string]string{} + } + for k, v := range signalAnnotations { + svc.Annotations[k] = v + } + return nil + }, + WithOwner[*corev1.Service](ctx.GetScheme(), stack), + WithLabels[*corev1.Service](collectorLabels()), + ) + if err != nil { + return fmt.Errorf("creating collector service: %w", err) + } + + return annotateStackCollectorSignals(ctx, stack, hasTraces, hasMetrics) +} + +func annotateStackCollectorSignals(ctx Context, stack *v1beta1.Stack, hasTraces, hasMetrics bool) error { + value := fmt.Sprintf("traces=%t,metrics=%t", hasTraces, hasMetrics) + annotations := stack.GetAnnotations() + if annotations != nil && annotations[collectorSignalAnnotation] == value { + return nil + } + patch := client.MergeFrom(stack.DeepCopy()) + if annotations == nil { + annotations = map[string]string{} + } + annotations[collectorSignalAnnotation] = value + stack.SetAnnotations(annotations) + return ctx.GetClient().Patch(ctx, stack, patch) +} + +func removeStackCollectorAnnotation(ctx Context, stackName string) error { + stack := &v1beta1.Stack{} + if err := ctx.GetClient().Get(ctx, types.NamespacedName{Name: stackName}, stack); err != nil { + if client.IgnoreNotFound(err) == nil { + return nil + } + return err + } + annotations := stack.GetAnnotations() + if annotations == nil { + return nil + } + if _, ok := annotations[collectorSignalAnnotation]; !ok { + return nil + } + patch := client.MergeFrom(stack.DeepCopy()) + delete(annotations, collectorSignalAnnotation) + stack.SetAnnotations(annotations) + return ctx.GetClient().Patch(ctx, stack, patch) +} + +func computeActiveSignals(endpoints []v1beta1.OtelExporterEndpoint, otelSettings *otelSettingsInput) (activeTraces, activeMetrics bool) { + for i := range endpoints { + if hasTraces(&endpoints[i]) { + activeTraces = true + } + if hasMetrics(&endpoints[i]) { + activeMetrics = true + } + } + if otelSettings != nil { + if otelSettings.TracesEndpoint != "" { + activeTraces = true + } + if otelSettings.MetricsEndpoint != "" { + activeMetrics = true + } + } + return +} + +func findMatchingEndpoints(ctx Context, stack *v1beta1.Stack) ([]v1beta1.OtelExporterEndpoint, error) { + var allEndpoints v1beta1.OtelExporterEndpointList + if err := ctx.GetClient().List(ctx, &allEndpoints); err != nil { + return nil, err + } + + stackLabels := labels.Set(stack.GetLabels()) + var matching []v1beta1.OtelExporterEndpoint + + for _, ep := range allEndpoints.Items { + if !ep.GetDeletionTimestamp().IsZero() { + continue + } + selector, err := selectorFromSpec(ep.Spec.StackSelector) + if err != nil { + log.FromContext(ctx).Error(err, "invalid stackSelector on OtelExporterEndpoint, skipping", "endpoint", ep.Name) + continue + } + if selector.Matches(stackLabels) { + matching = append(matching, ep) + } + } + + sort.Slice(matching, func(i, j int) bool { + return matching[i].Name < matching[j].Name + }) + return matching, nil +} + +type authSecretRef struct { + SecretName string + SecretKey string + Signal string + CRDName string +} + +func referencedAuthSecrets(endpoints []v1beta1.OtelExporterEndpoint) []authSecretRef { + var refs []authSecretRef + for _, ep := range endpoints { + crdName := sanitizeName(ep.Name) + for _, entry := range []struct { + signal string + config *v1beta1.OtelSignalConfig + }{ + {"TRACES", ep.Spec.Traces}, + {"METRICS", ep.Spec.Metrics}, + } { + if entry.config == nil || entry.config.Auth == nil || entry.config.Auth.Type != "bearer" { + continue + } + secretKey := entry.config.Auth.FromSecretKey + if secretKey == "" { + secretKey = "token" + } + refs = append(refs, authSecretRef{ + SecretName: entry.config.Auth.FromSecret, + SecretKey: secretKey, + Signal: entry.signal, + CRDName: crdName, + }) + } + } + return refs +} + +func buildCollectorInputs(endpoints []v1beta1.OtelExporterEndpoint) ([]collectorInput, []corev1.EnvVar) { + var inputs []collectorInput + var envVars []corev1.EnvVar + + refs := referencedAuthSecrets(endpoints) + refsByKey := map[string]string{} + for _, ref := range refs { + envName := fmt.Sprintf("AUTH_%s_%s", envSafe(ref.CRDName), ref.Signal) + refsByKey[ref.CRDName+"/"+ref.Signal] = envName + envVars = append(envVars, EnvFromSecret(envName, ref.SecretName, ref.SecretKey)) + } + + for _, ep := range endpoints { + crdName := sanitizeName(ep.Name) + ci := collectorInput{Endpoint: &ep} + ci.TracesEnvAlias = refsByKey[crdName+"/TRACES"] + ci.MetricsEnvAlias = refsByKey[crdName+"/METRICS"] + inputs = append(inputs, ci) + } + + return inputs, envVars +} + +func hashAuthSecrets(ctx Context, stackNamespace string, endpoints []v1beta1.OtelExporterEndpoint) (string, error) { + refs := referencedAuthSecrets(endpoints) + if len(refs) == 0 { + return "", nil + } + + seen := map[string]bool{} + digest := sha256.New() + + for _, ref := range refs { + dedupKey := ref.SecretName + "/" + ref.SecretKey + if seen[dedupKey] { + continue + } + seen[dedupKey] = true + + secret := &corev1.Secret{} + err := ctx.GetClient().Get(ctx, types.NamespacedName{ + Name: ref.SecretName, + Namespace: stackNamespace, + }, secret) + if err != nil { + return "", fmt.Errorf("auth secret %q in namespace %q: %w", ref.SecretName, stackNamespace, err) + } + tokenData, ok := secret.Data[ref.SecretKey] + if !ok { + return "", fmt.Errorf("auth secret %q in namespace %q is missing required key %q", ref.SecretName, stackNamespace, ref.SecretKey) + } + if _, err := digest.Write(tokenData); err != nil { + return "", err + } + } + + return base64.StdEncoding.EncodeToString(digest.Sum(nil)), nil +} + +func readOtelSettings(ctx Context, stackName string) (*otelSettingsInput, error) { + tracesURL, err := settings.GetURL(ctx, stackName, "opentelemetry", "traces", "dsn") + if err != nil { + return nil, err + } + metricsURL, err := settings.GetURL(ctx, stackName, "opentelemetry", "metrics", "dsn") + if err != nil { + return nil, err + } + + if tracesURL == nil && metricsURL == nil { + return nil, nil + } + + input := &otelSettingsInput{} + if tracesURL != nil { + input.TracesEndpoint = tracesURL.String() + } + if metricsURL != nil { + input.MetricsEndpoint = metricsURL.String() + } + return input, nil +} + +func deleteIfManaged[T client.Object](ctx Context, name types.NamespacedName) error { + var t T + t = reflect.New(reflect.TypeOf(t).Elem()).Interface().(T) + if err := ctx.GetClient().Get(ctx, name, t); err != nil { + if client.IgnoreNotFound(err) == nil { + return nil + } + return err + } + if t.GetLabels()[managedByLabel] != managedByValue { + return nil + } + LogDeletion(ctx, t, "deleteIfManaged") + return ctx.GetClient().Delete(ctx, t) +} + +func checkNotUnmanaged[T client.Object](ctx Context, name types.NamespacedName) error { + var t T + t = reflect.New(reflect.TypeOf(t).Elem()).Interface().(T) + if err := ctx.GetClient().Get(ctx, name, t); err != nil { + if client.IgnoreNotFound(err) == nil { + return nil + } + return err + } + if t.GetLabels()[managedByLabel] != managedByValue { + return fmt.Errorf("resource %s/%s already exists and is not managed by otelexporterendpoint", name.Namespace, name.Name) + } + return nil +} + +func otelRefName(stackName, secretName string) string { + name := fmt.Sprintf("%s--otel--%s", stackName, secretName) + if len(name) > 253 { + h := sha256.Sum256([]byte(name)) + name = name[:240] + "-" + hex.EncodeToString(h[:6]) + } + return name +} + +func ensureAuthSecretReferences(ctx Context, stack *v1beta1.Stack, endpoints []v1beta1.OtelExporterEndpoint) error { + wanted := map[string]bool{} + for _, ref := range referencedAuthSecrets(endpoints) { + if wanted[ref.SecretName] { + continue + } + wanted[ref.SecretName] = true + + existing := &corev1.Secret{} + if err := ctx.GetClient().Get(ctx, types.NamespacedName{ + Name: ref.SecretName, + Namespace: stack.Name, + }, existing); err == nil { + continue + } else if client.IgnoreNotFound(err) != nil { + return err + } + + refName := otelRefName(stack.Name, ref.SecretName) + rr, _, err := CreateOrUpdate(ctx, types.NamespacedName{ + Name: refName, + }, func(rr *v1beta1.ResourceReference) error { + rr.Spec.Stack = stack.Name + rr.Spec.Name = ref.SecretName + rr.Spec.GroupVersionKind = &metav1.GroupVersionKind{ + Group: "", + Version: "v1", + Kind: "Secret", + } + return nil + }, + WithOwner[*v1beta1.ResourceReference](ctx.GetScheme(), stack), + ) + if err != nil { + return fmt.Errorf("creating ResourceReference for secret %q in stack %q: %w", ref.SecretName, stack.Name, err) + } + if !rr.Status.Ready { + return NewPendingError() + } + } + + wantedRefNames := map[string]bool{} + for secretName := range wanted { + wantedRefNames[otelRefName(stack.Name, secretName)] = true + } + + return reconcileOtelResourceReferences(ctx, stack.Name, wantedRefNames) +} + +func ensureNoConflict(ctx Context, namespace string) error { + if err := checkNotUnmanaged[*corev1.ConfigMap](ctx, types.NamespacedName{ + Name: "otel-collector-config", Namespace: namespace, + }); err != nil { + return err + } + if err := checkNotUnmanaged[*appsv1.Deployment](ctx, types.NamespacedName{ + Name: deploymentName, Namespace: namespace, + }); err != nil { + return err + } + return checkNotUnmanaged[*corev1.Service](ctx, types.NamespacedName{ + Name: serviceName, Namespace: namespace, + }) +} + +func cleanupStackCollector(ctx Context, namespace string) error { + if err := deleteIfManaged[*corev1.Service](ctx, types.NamespacedName{ + Name: serviceName, Namespace: namespace, + }); err != nil { + return err + } + if err := deleteIfManaged[*appsv1.Deployment](ctx, types.NamespacedName{ + Name: deploymentName, Namespace: namespace, + }); err != nil { + return err + } + if err := deleteIfManaged[*corev1.ConfigMap](ctx, types.NamespacedName{ + Name: "otel-collector-config", Namespace: namespace, + }); err != nil { + return err + } + if err := reconcileOtelResourceReferences(ctx, namespace, nil); err != nil { + return err + } + return removeStackCollectorAnnotation(ctx, namespace) +} + +func reconcileOtelResourceReferences(ctx Context, stackName string, wanted map[string]bool) error { + var allRefs v1beta1.ResourceReferenceList + if err := ctx.GetClient().List(ctx, &allRefs, client.MatchingFields{ + "stack": stackName, + }); err != nil { + return err + } + prefix := stackName + "--otel--" + for i := range allRefs.Items { + rr := &allRefs.Items[i] + if !strings.HasPrefix(rr.Name, prefix) { + continue + } + if wanted[rr.Name] { + continue + } + if err := ctx.GetClient().Delete(ctx, rr); client.IgnoreNotFound(err) != nil { + return err + } + } + return nil +} + +func collectorImageForPlatform(ctx Context) string { + if img := ctx.GetPlatform().CollectorImage; img != "" { + return img + } + return DefaultCollectorImage +} + +func envSafe(s string) string { + return strings.Map(func(r rune) rune { + if (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9') || r == '_' { + return r + } + if r >= 'a' && r <= 'z' { + return r - 32 + } + return '_' + }, s) +} + +func isOtelSettingsKey(key string) bool { + parts := strings.Split(key, ".") + if len(parts) < 3 { + return false + } + if parts[0] != "opentelemetry" && parts[0] != "*" { + return false + } + signal := parts[1] + if signal != "traces" && signal != "metrics" && signal != "*" { + return false + } + return true +} + +func isCollectorResource(obj client.Object) bool { + return obj.GetLabels()[managedByLabel] == managedByValue +} + +func enqueueAllEndpoints(ctx Context) []reconcile.Request { + var endpoints v1beta1.OtelExporterEndpointList + if err := ctx.GetClient().List(ctx, &endpoints); err != nil { + log.FromContext(ctx).Error(err, "failed to list OtelExporterEndpoints for requeue") + return nil + } + return MapObjectToReconcileRequests( + collectionutils.Map(endpoints.Items, func(e v1beta1.OtelExporterEndpoint) *v1beta1.OtelExporterEndpoint { return &e })..., + ) +} + +func init() { + Init( + WithStdReconciler(Reconcile, + WithFinalizer[*v1beta1.OtelExporterEndpoint](collectorFinalizer, Cleanup), + WithWatch[*v1beta1.OtelExporterEndpoint, *v1beta1.Stack](func(ctx Context, _ *v1beta1.Stack) []reconcile.Request { + return enqueueAllEndpoints(ctx) + }), + WithWatch[*v1beta1.OtelExporterEndpoint, *v1beta1.ResourceReference](func(ctx Context, rr *v1beta1.ResourceReference) []reconcile.Request { + if !strings.HasPrefix(rr.Name, rr.Spec.Stack+"--otel--") { + return nil + } + return enqueueAllEndpoints(ctx) + }), + WithRaw[*v1beta1.OtelExporterEndpoint](func(ctx Context, b *builder.Builder) error { + b.Watches(&v1beta1.Settings{}, handler.EnqueueRequestsFromMapFunc( + func(_ context.Context, obj client.Object) []reconcile.Request { + s := obj.(*v1beta1.Settings) + if !isOtelSettingsKey(s.Spec.Key) { + return nil + } + return enqueueAllEndpoints(ctx) + }, + )) + return nil + }), + WithRaw[*v1beta1.OtelExporterEndpoint](func(ctx Context, b *builder.Builder) error { + collectorPredicate := predicate.NewPredicateFuncs(isCollectorResource) + enqueueHandler := handler.EnqueueRequestsFromMapFunc( + func(_ context.Context, _ client.Object) []reconcile.Request { + return enqueueAllEndpoints(ctx) + }, + ) + b.Watches(&corev1.ConfigMap{}, enqueueHandler, builder.WithPredicates(collectorPredicate)) + b.Watches(&appsv1.Deployment{}, enqueueHandler, builder.WithPredicates(collectorPredicate)) + b.Watches(&corev1.Service{}, enqueueHandler, builder.WithPredicates(collectorPredicate)) + return nil + }), + ), + ) +} diff --git a/internal/resources/settings/opentelemetry.go b/internal/resources/settings/opentelemetry.go index 3065c37e9..68986209e 100644 --- a/internal/resources/settings/opentelemetry.go +++ b/internal/resources/settings/opentelemetry.go @@ -5,7 +5,9 @@ import ( "slices" "strings" - v1 "k8s.io/api/core/v1" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/types" "github.com/formancehq/operator/v3/internal/core" ) @@ -15,9 +17,20 @@ type MonitoringType string const ( MonitoringTypeTraces MonitoringType = "TRACES" MonitoringTypeMetrics MonitoringType = "METRICS" + + collectorServiceName = "otel-collector" + collectorServicePort = 4318 ) -func GetOTELEnvVars(ctx core.Context, stack, serviceName string, sliceStringSeparator string) ([]v1.EnvVar, error) { +func GetOTELEnvVars(ctx core.Context, stack, serviceName string, sliceStringSeparator string) ([]corev1.EnvVar, error) { + info, err := getCollectorInfo(ctx, stack) + if err != nil { + return nil, err + } + if info != nil { + return collectorEnvVars(ctx, info, stack, serviceName, sliceStringSeparator) + } + traces, err := otelEnvVars(ctx, stack, MonitoringTypeTraces, serviceName, sliceStringSeparator) if err != nil { return nil, err @@ -34,20 +47,111 @@ func GetOTELEnvVars(ctx core.Context, stack, serviceName string, sliceStringSepa return append(traces, metrics...), nil } +type collectorInfo struct { + endpoint string + hasTraces bool + hasMetrics bool +} + +func getCollectorInfo(ctx core.Context, stack string) (*collectorInfo, error) { + svc := &corev1.Service{} + err := ctx.GetClient().Get(ctx, types.NamespacedName{ + Namespace: stack, + Name: collectorServiceName, + }, svc) + if err != nil { + if apierrors.IsNotFound(err) { + return nil, nil + } + return nil, err + } + if svc.Labels[core.CollectorManagedByLabel] != core.CollectorManagedByValue { + return nil, nil + } + return &collectorInfo{ + endpoint: fmt.Sprintf("%s.%s:%d", collectorServiceName, stack, collectorServicePort), + hasTraces: svc.Annotations[core.SignalTracesAnnotation] == "true", + hasMetrics: svc.Annotations[core.SignalMetricsAnnotation] == "true", + }, nil +} + +func collectorEnvVars(ctx core.Context, info *collectorInfo, stack, serviceName, sliceStringSeparator string) ([]corev1.EnvVar, error) { + resourceAttributes := map[string]string{} + for _, signal := range []string{"traces", "metrics"} { + attrs, err := GetMap(ctx, stack, "opentelemetry", signal, "resource-attributes") + if err != nil { + return nil, err + } + for k, v := range attrs { + resourceAttributes[k] = v + } + } + resourceAttributes["stack"] = stack + resourceAttributes["pod-name"] = "$(POD_NAME)" + + resourceAttributesArray := make([]string, 0, len(resourceAttributes)) + for k, v := range resourceAttributes { + resourceAttributesArray = append(resourceAttributesArray, fmt.Sprintf("%s=%s", k, v)) + } + slices.Sort(resourceAttributesArray) + + envVars := []corev1.EnvVar{ + core.Env("OTEL_SERVICE_NAME", serviceName), + { + Name: "POD_NAME", + ValueFrom: &corev1.EnvVarSource{ + FieldRef: &corev1.ObjectFieldSelector{ + FieldPath: "metadata.name", + }, + }, + }, + core.Env("OTEL_RESOURCE_ATTRIBUTES", strings.Join(resourceAttributesArray, sliceStringSeparator)), + } + + if info.hasTraces { + envVars = append(envVars, + core.Env("OTEL_TRACES", "true"), + core.Env("OTEL_TRACES_BATCH", "true"), + core.Env("OTEL_TRACES_EXPORTER", "otlp"), + core.Env("OTEL_TRACES_EXPORTER_OTLP_ENDPOINT", info.endpoint), + core.Env("OTEL_TRACES_EXPORTER_OTLP_MODE", "http"), + core.EnvFromBool("OTEL_TRACES_EXPORTER_OTLP_INSECURE", true), + ) + } + + if info.hasMetrics { + envVars = append(envVars, + core.Env("OTEL_METRICS", "true"), + core.Env("OTEL_METRICS_BATCH", "true"), + core.Env("OTEL_METRICS_EXPORTER", "otlp"), + core.Env("OTEL_METRICS_EXPORTER_OTLP_ENDPOINT", info.endpoint), + core.Env("OTEL_METRICS_EXPORTER_OTLP_MODE", "http"), + core.EnvFromBool("OTEL_METRICS_EXPORTER_OTLP_INSECURE", true), + core.Env("OTEL_METRICS_RUNTIME", "true"), + ) + } + + return envVars, nil +} + func HasOpenTelemetryTracesEnabled(ctx core.Context, stack string) (bool, error) { - v, err := GetURL(ctx, stack, "opentelemetry", "traces", "dsn") + info, err := getCollectorInfo(ctx, stack) if err != nil { return false, err } + if info != nil { + return info.hasTraces, nil + } - if v == nil { - return false, nil + v, err := GetURL(ctx, stack, "opentelemetry", "traces", "dsn") + if err != nil { + return false, err } - return true, nil + return v != nil, nil } -func otelEnvVars(ctx core.Context, stack string, monitoringType MonitoringType, serviceName, sliceStringSeparator string) ([]v1.EnvVar, error) { +func otelEnvVars(ctx core.Context, stack string, monitoringType MonitoringType, serviceName, sliceStringSeparator string) ([]corev1.EnvVar, error) { otlp, err := GetURL(ctx, stack, "opentelemetry", strings.ToLower(string(monitoringType)), "dsn") if err != nil { @@ -57,7 +161,7 @@ func otelEnvVars(ctx core.Context, stack string, monitoringType MonitoringType, return nil, nil } - ret := []v1.EnvVar{ + ret := []corev1.EnvVar{ core.Env(fmt.Sprintf("OTEL_%s", string(monitoringType)), "true"), core.Env(fmt.Sprintf("OTEL_%s_BATCH", string(monitoringType)), "true"), core.Env(fmt.Sprintf("OTEL_%s_EXPORTER", string(monitoringType)), "otlp"), @@ -66,8 +170,8 @@ func otelEnvVars(ctx core.Context, stack string, monitoringType MonitoringType, core.Env(fmt.Sprintf("OTEL_%s_EXPORTER_OTLP_MODE", string(monitoringType)), otlp.Scheme), { Name: "POD_NAME", - ValueFrom: &v1.EnvVarSource{ - FieldRef: &v1.ObjectFieldSelector{ + ValueFrom: &corev1.EnvVarSource{ + FieldRef: &corev1.ObjectFieldSelector{ FieldPath: "metadata.name", }, }, @@ -75,7 +179,7 @@ func otelEnvVars(ctx core.Context, stack string, monitoringType MonitoringType, } // If the path is not empty, we use the full URL as the endpoint. - var otlpEndpoint v1.EnvVar + var otlpEndpoint corev1.EnvVar otlpEndpointEnvName := fmt.Sprintf("OTEL_%s_EXPORTER_OTLP_ENDPOINT", string(monitoringType)) if otlp.Path != "" { otlpEndpoint = core.Env(otlpEndpointEnvName, otlp.String()) diff --git a/internal/tests/otelexporterendpoint_controller_test.go b/internal/tests/otelexporterendpoint_controller_test.go new file mode 100644 index 000000000..914b4b14f --- /dev/null +++ b/internal/tests/otelexporterendpoint_controller_test.go @@ -0,0 +1,318 @@ +package tests_test + +import ( + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + v1beta1 "github.com/formancehq/operator/v3/api/formance.com/v1beta1" + . "github.com/formancehq/operator/v3/internal/tests/internal" +) + +var _ = Describe("OtelExporterEndpointController", func() { + Context("When creating an OtelExporterEndpoint with stackSelector matching a stack", func() { + var ( + stack *v1beta1.Stack + endpoint *v1beta1.OtelExporterEndpoint + ) + BeforeEach(func() { + stack = &v1beta1.Stack{ + ObjectMeta: metav1.ObjectMeta{ + Name: RandObjectMeta().Name, + Labels: map[string]string{ + "formance.com/stack": "sdymzzszghxw-ryeg", + }, + }, + Spec: v1beta1.StackSpec{Version: "v99.0.0"}, + } + endpoint = &v1beta1.OtelExporterEndpoint{ + ObjectMeta: RandObjectMeta(), + Spec: v1beta1.OtelExporterEndpointSpec{ + StackSelector: &metav1.LabelSelector{ + MatchLabels: map[string]string{ + "formance.com/stack": "sdymzzszghxw-ryeg", + }, + }, + Traces: &v1beta1.OtelSignalConfig{ + Endpoint: "http://my-collector:4318", + }, + }, + } + }) + JustBeforeEach(func() { + Expect(Create(stack)).To(Succeed()) + Expect(Create(endpoint)).To(Succeed()) + }) + AfterEach(func() { + Expect(Delete(endpoint)).To(Succeed()) + Expect(Delete(stack)).To(Succeed()) + }) + It("Should create a ConfigMap, Deployment, and Service", func() { + By("Should set the status to ready", func() { + Eventually(func(g Gomega) bool { + g.Expect(LoadResource("", endpoint.Name, endpoint)).To(Succeed()) + return endpoint.Status.Ready + }).Should(BeTrue()) + }) + By("Should track the matching stack", func() { + Expect(endpoint.Status.Stacks).To(ContainElement(stack.Name)) + }) + By("Should create a ConfigMap with collector config", func() { + cm := &corev1.ConfigMap{} + Eventually(func() error { + return LoadResource(stack.Name, "otel-collector-config", cm) + }).Should(Succeed()) + Expect(cm.Data).To(HaveKey("otel-collector-config.yaml")) + Expect(cm.Data["otel-collector-config.yaml"]).To(ContainSubstring("http://my-collector:4318")) + }) + By("Should create a Deployment", func() { + deployment := &appsv1.Deployment{} + Eventually(func() error { + return LoadResource(stack.Name, "otel-collector", deployment) + }).Should(Succeed()) + }) + By("Should create a Service", func() { + svc := &corev1.Service{} + Eventually(func() error { + return LoadResource(stack.Name, "otel-collector", svc) + }).Should(Succeed()) + }) + }) + }) + + Context("When deleting an OtelExporterEndpoint", func() { + var ( + stack *v1beta1.Stack + endpoint *v1beta1.OtelExporterEndpoint + ) + BeforeEach(func() { + stack = &v1beta1.Stack{ + ObjectMeta: metav1.ObjectMeta{ + Name: RandObjectMeta().Name, + Labels: map[string]string{ + "formance.com/stack": "sdymzzszghxw-ryeg", + }, + }, + Spec: v1beta1.StackSpec{Version: "v99.0.0"}, + } + endpoint = &v1beta1.OtelExporterEndpoint{ + ObjectMeta: RandObjectMeta(), + Spec: v1beta1.OtelExporterEndpointSpec{ + StackSelector: &metav1.LabelSelector{ + MatchLabels: map[string]string{ + "formance.com/stack": "sdymzzszghxw-ryeg", + }, + }, + Traces: &v1beta1.OtelSignalConfig{ + Endpoint: "http://my-collector:4318", + }, + }, + } + }) + JustBeforeEach(func() { + Expect(Create(stack)).To(Succeed()) + Expect(Create(endpoint)).To(Succeed()) + }) + AfterEach(func() { + _ = Delete(endpoint) + Expect(Delete(stack)).To(Succeed()) + }) + It("Should clean up collector resources from the stack namespace", func() { + By("Waiting for the collector to be created", func() { + Eventually(func() error { + return LoadResource(stack.Name, "otel-collector", &appsv1.Deployment{}) + }).Should(Succeed()) + }) + By("Deleting the endpoint", func() { + Expect(Delete(endpoint)).To(Succeed()) + }) + By("Verifying the Deployment is removed", func() { + Eventually(func() error { + return LoadResource(stack.Name, "otel-collector", &appsv1.Deployment{}) + }).ShouldNot(Succeed()) + }) + By("Verifying the Service is removed", func() { + Eventually(func() error { + return LoadResource(stack.Name, "otel-collector", &corev1.Service{}) + }).ShouldNot(Succeed()) + }) + By("Verifying the ConfigMap is removed", func() { + Eventually(func() error { + return LoadResource(stack.Name, "otel-collector-config", &corev1.ConfigMap{}) + }).ShouldNot(Succeed()) + }) + }) + }) + + Context("When no OtelExporterEndpoints exist and no Settings", func() { + var ( + stack *v1beta1.Stack + ) + BeforeEach(func() { + stack = &v1beta1.Stack{ + ObjectMeta: metav1.ObjectMeta{ + Name: RandObjectMeta().Name, + Labels: map[string]string{ + "formance.com/stack": "sdymzzszghxw-ryeg", + }, + }, + Spec: v1beta1.StackSpec{Version: "v99.0.0"}, + } + }) + JustBeforeEach(func() { + Expect(Create(stack)).To(Succeed()) + }) + AfterEach(func() { + Expect(Delete(stack)).To(Succeed()) + }) + It("Should not create a collector service", func() { + svc := &corev1.Service{} + Consistently(func() error { + return LoadResource(stack.Name, "otel-collector", svc) + }).ShouldNot(Succeed()) + }) + }) + + Context("When creating an OtelExporterEndpoint targeting all stacks", func() { + var ( + stack *v1beta1.Stack + endpoint *v1beta1.OtelExporterEndpoint + authSecret *corev1.Secret + ) + BeforeEach(func() { + stack = &v1beta1.Stack{ + ObjectMeta: metav1.ObjectMeta{ + Name: RandObjectMeta().Name, + Labels: map[string]string{ + "formance.com/stack": "sdymzzszghxw-ryeg", + }, + }, + Spec: v1beta1.StackSpec{Version: "v99.0.0"}, + } + endpoint = &v1beta1.OtelExporterEndpoint{ + ObjectMeta: RandObjectMeta(), + Spec: v1beta1.OtelExporterEndpointSpec{ + StackSelector: &metav1.LabelSelector{ + MatchExpressions: []metav1.LabelSelectorRequirement{ + { + Key: "formance.com/stack", + Operator: metav1.LabelSelectorOpExists, + }, + }, + }, + Traces: &v1beta1.OtelSignalConfig{ + Endpoint: "https://support.frmnc.net", + Auth: &v1beta1.OtelExporterAuth{ + Type: "bearer", + FromSecret: "formance-license", + }, + }, + ResourceAttributes: map[string]string{ + "cluster.id": "test-cluster", + }, + }, + } + }) + JustBeforeEach(func() { + Expect(Create(stack)).To(Succeed()) + Eventually(func() error { + ns := &corev1.Namespace{} + return LoadResource("", stack.Name, ns) + }).Should(Succeed()) + authSecret = &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{ + Name: "formance-license", + Namespace: stack.Name, + }, + Data: map[string][]byte{ + "token": []byte("test-token"), + }, + } + Expect(Create(authSecret)).To(Succeed()) + Expect(Create(endpoint)).To(Succeed()) + }) + AfterEach(func() { + Expect(Delete(endpoint)).To(Succeed()) + Expect(Delete(authSecret)).To(Succeed()) + Expect(Delete(stack)).To(Succeed()) + }) + It("Should create a collector with auth headers", func() { + Eventually(func(g Gomega) bool { + g.Expect(LoadResource("", endpoint.Name, endpoint)).To(Succeed()) + return endpoint.Status.Ready + }).Should(BeTrue()) + + cm := &corev1.ConfigMap{} + Eventually(func() error { + return LoadResource(stack.Name, "otel-collector-config", cm) + }).Should(Succeed()) + Expect(cm.Data["otel-collector-config.yaml"]).To(ContainSubstring("https://support.frmnc.net")) + Expect(cm.Data["otel-collector-config.yaml"]).To(ContainSubstring("authorization")) + }) + }) + + Context("When creating an OtelExporterEndpoint with Settings fallback", func() { + var ( + stack *v1beta1.Stack + setting *v1beta1.Settings + endpoint *v1beta1.OtelExporterEndpoint + ) + BeforeEach(func() { + stack = &v1beta1.Stack{ + ObjectMeta: metav1.ObjectMeta{ + Name: RandObjectMeta().Name, + Labels: map[string]string{ + "formance.com/stack": "sdymzzszghxw-ryeg", + }, + }, + Spec: v1beta1.StackSpec{Version: "v99.0.0"}, + } + setting = &v1beta1.Settings{ + ObjectMeta: RandObjectMeta(), + Spec: v1beta1.SettingsSpec{ + Stacks: []string{"*"}, + Key: "opentelemetry.traces.dsn", + Value: "http://settings-collector:4318", + }, + } + endpoint = &v1beta1.OtelExporterEndpoint{ + ObjectMeta: RandObjectMeta(), + Spec: v1beta1.OtelExporterEndpointSpec{ + StackSelector: &metav1.LabelSelector{ + MatchLabels: map[string]string{ + "formance.com/stack": "sdymzzszghxw-ryeg", + }, + }, + Traces: &v1beta1.OtelSignalConfig{ + Endpoint: "http://my-collector:4318", + }, + }, + } + }) + JustBeforeEach(func() { + Expect(Create(stack)).To(Succeed()) + Expect(Create(setting)).To(Succeed()) + Expect(Create(endpoint)).To(Succeed()) + }) + AfterEach(func() { + Expect(Delete(setting)).To(Succeed()) + Expect(Delete(endpoint)).To(Succeed()) + Expect(Delete(stack)).To(Succeed()) + }) + It("Should include both CRD and Settings endpoints in collector config", func() { + Eventually(func(g Gomega) bool { + g.Expect(LoadResource("", endpoint.Name, endpoint)).To(Succeed()) + return endpoint.Status.Ready + }).Should(BeTrue()) + + cm := &corev1.ConfigMap{} + Eventually(func() error { + return LoadResource(stack.Name, "otel-collector-config", cm) + }).Should(Succeed()) + Expect(cm.Data["otel-collector-config.yaml"]).To(ContainSubstring("http://my-collector:4318")) + Expect(cm.Data["otel-collector-config.yaml"]).To(ContainSubstring("settings-collector:4318")) + }) + }) +})