diff --git a/api/v1/logcollector_types.go b/api/v1/logcollector_types.go index f501a1855f..dceb062f05 100644 --- a/api/v1/logcollector_types.go +++ b/api/v1/logcollector_types.go @@ -59,6 +59,15 @@ type LogCollectorSpec struct { // EKSLogForwarderDeployment configures the EKSLogForwarderDeployment Deployment. // +optional EKSLogForwarderDeployment *EKSLogForwarderDeployment `json:"eksLogForwarderDeployment,omitempty"` + + // OTelCollector configures the OpenTelemetry Collector for exporting logs + // and metrics via OTLP. Unlike AdditionalStores entries (S3, Syslog, + // Splunk), which point at external systems, the OTel Collector is + // operator-managed infrastructure (StatefulSet, ConfigMap, RBAC, certs) + // with its own lifecycle, so it lives at the top level rather than under + // AdditionalStores. + // +optional + OTelCollector *OTelCollectorSpec `json:"otelCollector,omitempty"` } type CollectProcessPathOption string @@ -260,6 +269,38 @@ type LogCollectorList struct { Items []LogCollector `json:"items"` } +// OTelCollectorSpec defines the desired state of the OpenTelemetry Collector. +type OTelCollectorSpec struct { + // Logs configures which log types are exported via OTLP. + // +optional + Logs *OTelLogs `json:"logs,omitempty"` + + // Metrics configures whether Calico component metrics are exported via OTLP. + // +optional + Metrics *OTelMetrics `json:"metrics,omitempty"` + + // Exporters configures the OTLP export endpoints. + // +optional + Exporters []OTelExporter `json:"exporters,omitempty"` + + // OTelCollectorStatefulSet configures the OTel Collector StatefulSet. + // +optional + OTelCollectorStatefulSet *OTelCollectorStatefulSet `json:"openTelemetryCollectorStatefulSet,omitempty"` +} + +func (s *OTelCollectorSpec) HasLogs() bool { + return s != nil && s.Logs != nil && len(s.Logs.Types) > 0 +} + +func (s *OTelCollectorSpec) MetricsEnabled() bool { + return s != nil && s.Metrics != nil && s.Metrics.Enabled != nil && + *s.Metrics.Enabled == OTelMetricsEnable +} + +func (s *OTelCollectorSpec) HasDataSources() bool { + return s.HasLogs() || s.MetricsEnabled() +} + func init() { SchemeBuilder.Register(&LogCollector{}, &LogCollectorList{}) } diff --git a/api/v1/otelcollector_types.go b/api/v1/otelcollector_types.go new file mode 100644 index 0000000000..2c6f0cfb87 --- /dev/null +++ b/api/v1/otelcollector_types.go @@ -0,0 +1,169 @@ +// Copyright (c) 2026 Tigera, Inc. All rights reserved. +// +// 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 v1 + +import ( + corev1 "k8s.io/api/core/v1" +) + +// OTelLogType represents the allowable log types for OTel export. +// +kubebuilder:validation:Enum=Audit;DNS;Flows +type OTelLogType string + +const ( + OTelAuditLog OTelLogType = "Audit" + OTelDNSLog OTelLogType = "DNS" + OTelFlowLog OTelLogType = "Flows" +) + +// OTelLogs configures log export. +type OTelLogs struct { + // Types specifies which log types to export. Supported values: Audit, DNS, Flows. + // +optional + Types []OTelLogType `json:"types,omitempty"` +} + +// OTelMetricsEnabled is the option to enable or disable metrics export. +// +kubebuilder:validation:Enum=Enabled;Disabled +type OTelMetricsEnabled string + +const ( + OTelMetricsEnable OTelMetricsEnabled = "Enabled" + OTelMetricsDisable OTelMetricsEnabled = "Disabled" +) + +// OTelMetrics configures metrics export. +type OTelMetrics struct { + // Enabled specifies whether to scrape and export Calico component metrics via OTLP. + // Default: Disabled + // +optional + Enabled *OTelMetricsEnabled `json:"enabled,omitempty"` +} + +// OTelExporterProtocol specifies the OTLP transport protocol. +// +kubebuilder:validation:Enum=grpc;http +type OTelExporterProtocol string + +const ( + OTelProtocolGRPC OTelExporterProtocol = "grpc" + OTelProtocolHTTP OTelExporterProtocol = "http" +) + +// OTelExporter defines an OTLP export endpoint. +type OTelExporter struct { + // Name is a unique identifier for this exporter. + Name string `json:"name"` + + // Endpoint is the OTLP endpoint URL. + Endpoint string `json:"endpoint"` + + // Protocol specifies the OTLP transport protocol. Default: grpc. + // +optional + // +kubebuilder:default=grpc + Protocol OTelExporterProtocol `json:"protocol,omitempty"` + + // TLSInsecure disables TLS verification for this exporter. Only use for trusted in-cluster targets. + // Default: false + // +optional + TLSInsecure *bool `json:"tlsInsecure,omitempty"` +} + +// OTelCollectorStatefulSet is the configuration for the OTel Collector StatefulSet. +type OTelCollectorStatefulSet struct { + // Metadata is a subset of a Kubernetes object's metadata that is added to the StatefulSet. + // +optional + Metadata *Metadata `json:"metadata,omitempty"` + // Spec is the specification of the OTel Collector StatefulSet. + // +optional + Spec *OTelCollectorStatefulSetSpec `json:"spec,omitempty"` +} + +// OTelCollectorStatefulSetSpec defines configuration for the OTel Collector StatefulSet. +type OTelCollectorStatefulSetSpec struct { + // MinReadySeconds is the minimum number of seconds for which a newly created StatefulSet pod should + // be ready without any of its container crashing, for it to be considered available. + // If specified, this overrides any minReadySeconds value that may be set on the OTel Collector StatefulSet. + // If omitted, the OTel Collector StatefulSet will use its default value for minReadySeconds. + // +optional + // +kubebuilder:validation:Minimum=0 + // +kubebuilder:validation:Maximum=2147483647 + MinReadySeconds *int32 `json:"minReadySeconds,omitempty"` + + // Template describes the OTel Collector StatefulSet pod that will be created. + // +optional + Template *OTelCollectorStatefulSetPodTemplateSpec `json:"template,omitempty"` +} + +// OTelCollectorStatefulSetPodTemplateSpec is the OTel Collector StatefulSet's PodTemplateSpec. +type OTelCollectorStatefulSetPodTemplateSpec struct { + // Metadata is a subset of a Kubernetes object's metadata that is added to the pod's metadata. + // +optional + Metadata *Metadata `json:"metadata,omitempty"` + // Spec is the OTel Collector StatefulSet's PodSpec. + // +optional + Spec *OTelCollectorStatefulSetPodSpec `json:"spec,omitempty"` +} + +// OTelCollectorStatefulSetPodSpec is the OTel Collector StatefulSet's PodSpec. +type OTelCollectorStatefulSetPodSpec struct { + // Affinity is a group of affinity scheduling rules for the OTel Collector pods. + // +optional + Affinity *corev1.Affinity `json:"affinity"` + // Containers is a list of OTel Collector containers. + // If specified, this overrides the specified OTel Collector StatefulSet containers. + // If omitted, the OTel Collector StatefulSet will use its default values for its containers. + // +optional + Containers []OTelCollectorStatefulSetContainer `json:"containers,omitempty"` + // NodeSelector gives more control over the nodes where the OTel Collector pods will run on. + // +optional + NodeSelector map[string]string `json:"nodeSelector,omitempty"` + // TopologySpreadConstraints describes how a group of pods ought to spread across topology + // domains. Scheduler will schedule pods in a way which abides by the constraints. + // All topologySpreadConstraints are ANDed. + // +optional + TopologySpreadConstraints []corev1.TopologySpreadConstraint `json:"topologySpreadConstraints,omitempty"` + // Tolerations is the OTel Collector pod's tolerations. + // If specified, this overrides any tolerations that may be set on the OTel Collector StatefulSet. + // If omitted, the OTel Collector StatefulSet will use its default value for tolerations. + // +optional + Tolerations []corev1.Toleration `json:"tolerations"` + // PriorityClassName allows to specify a PriorityClass resource to be used. + // +optional + PriorityClassName string `json:"priorityClassName,omitempty"` +} + +// OTelCollectorStatefulSetContainer is an OTel Collector StatefulSet container. +type OTelCollectorStatefulSetContainer struct { + // Name is an enum which identifies the OTel Collector StatefulSet container by name. + // Supported values are: otel-collector + // +kubebuilder:validation:Enum=otel-collector + Name string `json:"name"` + + // Resources allows customization of limits and requests for compute resources such as cpu and memory. + // If specified, this overrides the named OTel Collector StatefulSet container's resources. + // If omitted, the OTel Collector StatefulSet will use its default value for this container's resources. + // +optional + Resources *corev1.ResourceRequirements `json:"resources,omitempty"` + + // ReadinessProbe allows customization of the readiness probe timing parameters. + // The probe handler is set by the operator and cannot be overridden. + // +optional + ReadinessProbe *ProbeOverride `json:"readinessProbe,omitempty"` + + // LivenessProbe allows customization of the liveness probe timing parameters. + // The probe handler is set by the operator and cannot be overridden. + // +optional + LivenessProbe *ProbeOverride `json:"livenessProbe,omitempty"` +} diff --git a/api/v1/zz_generated.deepcopy.go b/api/v1/zz_generated.deepcopy.go index 30e6ba5f6c..152ac96598 100644 --- a/api/v1/zz_generated.deepcopy.go +++ b/api/v1/zz_generated.deepcopy.go @@ -7293,6 +7293,11 @@ func (in *LogCollectorSpec) DeepCopyInto(out *LogCollectorSpec) { *out = new(EKSLogForwarderDeployment) (*in).DeepCopyInto(*out) } + if in.OTelCollector != nil { + in, out := &in.OTelCollector, &out.OTelCollector + *out = new(OTelCollectorSpec) + (*in).DeepCopyInto(*out) + } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new LogCollectorSpec. @@ -8333,6 +8338,256 @@ func (in *NonClusterHostSpec) DeepCopy() *NonClusterHostSpec { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *OTelCollectorSpec) DeepCopyInto(out *OTelCollectorSpec) { + *out = *in + if in.Logs != nil { + in, out := &in.Logs, &out.Logs + *out = new(OTelLogs) + (*in).DeepCopyInto(*out) + } + if in.Metrics != nil { + in, out := &in.Metrics, &out.Metrics + *out = new(OTelMetrics) + (*in).DeepCopyInto(*out) + } + if in.Exporters != nil { + in, out := &in.Exporters, &out.Exporters + *out = make([]OTelExporter, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } + if in.OTelCollectorStatefulSet != nil { + in, out := &in.OTelCollectorStatefulSet, &out.OTelCollectorStatefulSet + *out = new(OTelCollectorStatefulSet) + (*in).DeepCopyInto(*out) + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new OTelCollectorSpec. +func (in *OTelCollectorSpec) DeepCopy() *OTelCollectorSpec { + if in == nil { + return nil + } + out := new(OTelCollectorSpec) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *OTelCollectorStatefulSet) DeepCopyInto(out *OTelCollectorStatefulSet) { + *out = *in + if in.Metadata != nil { + in, out := &in.Metadata, &out.Metadata + *out = new(Metadata) + (*in).DeepCopyInto(*out) + } + if in.Spec != nil { + in, out := &in.Spec, &out.Spec + *out = new(OTelCollectorStatefulSetSpec) + (*in).DeepCopyInto(*out) + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new OTelCollectorStatefulSet. +func (in *OTelCollectorStatefulSet) DeepCopy() *OTelCollectorStatefulSet { + if in == nil { + return nil + } + out := new(OTelCollectorStatefulSet) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *OTelCollectorStatefulSetContainer) DeepCopyInto(out *OTelCollectorStatefulSetContainer) { + *out = *in + if in.Resources != nil { + in, out := &in.Resources, &out.Resources + *out = new(corev1.ResourceRequirements) + (*in).DeepCopyInto(*out) + } + if in.ReadinessProbe != nil { + in, out := &in.ReadinessProbe, &out.ReadinessProbe + *out = new(ProbeOverride) + (*in).DeepCopyInto(*out) + } + if in.LivenessProbe != nil { + in, out := &in.LivenessProbe, &out.LivenessProbe + *out = new(ProbeOverride) + (*in).DeepCopyInto(*out) + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new OTelCollectorStatefulSetContainer. +func (in *OTelCollectorStatefulSetContainer) DeepCopy() *OTelCollectorStatefulSetContainer { + if in == nil { + return nil + } + out := new(OTelCollectorStatefulSetContainer) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *OTelCollectorStatefulSetPodSpec) DeepCopyInto(out *OTelCollectorStatefulSetPodSpec) { + *out = *in + if in.Affinity != nil { + in, out := &in.Affinity, &out.Affinity + *out = new(corev1.Affinity) + (*in).DeepCopyInto(*out) + } + if in.Containers != nil { + in, out := &in.Containers, &out.Containers + *out = make([]OTelCollectorStatefulSetContainer, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } + if in.NodeSelector != nil { + in, out := &in.NodeSelector, &out.NodeSelector + *out = make(map[string]string, len(*in)) + for key, val := range *in { + (*out)[key] = val + } + } + if in.TopologySpreadConstraints != nil { + in, out := &in.TopologySpreadConstraints, &out.TopologySpreadConstraints + *out = make([]corev1.TopologySpreadConstraint, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } + if in.Tolerations != nil { + in, out := &in.Tolerations, &out.Tolerations + *out = make([]corev1.Toleration, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new OTelCollectorStatefulSetPodSpec. +func (in *OTelCollectorStatefulSetPodSpec) DeepCopy() *OTelCollectorStatefulSetPodSpec { + if in == nil { + return nil + } + out := new(OTelCollectorStatefulSetPodSpec) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *OTelCollectorStatefulSetPodTemplateSpec) DeepCopyInto(out *OTelCollectorStatefulSetPodTemplateSpec) { + *out = *in + if in.Metadata != nil { + in, out := &in.Metadata, &out.Metadata + *out = new(Metadata) + (*in).DeepCopyInto(*out) + } + if in.Spec != nil { + in, out := &in.Spec, &out.Spec + *out = new(OTelCollectorStatefulSetPodSpec) + (*in).DeepCopyInto(*out) + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new OTelCollectorStatefulSetPodTemplateSpec. +func (in *OTelCollectorStatefulSetPodTemplateSpec) DeepCopy() *OTelCollectorStatefulSetPodTemplateSpec { + if in == nil { + return nil + } + out := new(OTelCollectorStatefulSetPodTemplateSpec) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *OTelCollectorStatefulSetSpec) DeepCopyInto(out *OTelCollectorStatefulSetSpec) { + *out = *in + if in.MinReadySeconds != nil { + in, out := &in.MinReadySeconds, &out.MinReadySeconds + *out = new(int32) + **out = **in + } + if in.Template != nil { + in, out := &in.Template, &out.Template + *out = new(OTelCollectorStatefulSetPodTemplateSpec) + (*in).DeepCopyInto(*out) + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new OTelCollectorStatefulSetSpec. +func (in *OTelCollectorStatefulSetSpec) DeepCopy() *OTelCollectorStatefulSetSpec { + if in == nil { + return nil + } + out := new(OTelCollectorStatefulSetSpec) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *OTelExporter) DeepCopyInto(out *OTelExporter) { + *out = *in + if in.TLSInsecure != nil { + in, out := &in.TLSInsecure, &out.TLSInsecure + *out = new(bool) + **out = **in + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new OTelExporter. +func (in *OTelExporter) DeepCopy() *OTelExporter { + if in == nil { + return nil + } + out := new(OTelExporter) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *OTelLogs) DeepCopyInto(out *OTelLogs) { + *out = *in + if in.Types != nil { + in, out := &in.Types, &out.Types + *out = make([]OTelLogType, len(*in)) + copy(*out, *in) + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new OTelLogs. +func (in *OTelLogs) DeepCopy() *OTelLogs { + if in == nil { + return nil + } + out := new(OTelLogs) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *OTelMetrics) DeepCopyInto(out *OTelMetrics) { + *out = *in + if in.Enabled != nil { + in, out := &in.Enabled, &out.Enabled + *out = new(OTelMetricsEnabled) + **out = **in + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new OTelMetrics. +func (in *OTelMetrics) DeepCopy() *OTelMetrics { + if in == nil { + return nil + } + out := new(OTelMetrics) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *PacketCaptureAPI) DeepCopyInto(out *PacketCaptureAPI) { *out = *in diff --git a/internal/controller/controllers.go b/internal/controller/controllers.go index 5f9f7d487e..c0b1a72343 100644 --- a/internal/controller/controllers.go +++ b/internal/controller/controllers.go @@ -202,6 +202,12 @@ func AddToManager(mgr ctrl.Manager, options options.ControllerOptions) error { }).SetupWithManager(mgr, options); err != nil { return fmt.Errorf("failed to create controller %s: %v", "PodIPRecovery", err) } + if err := (&OpenTelemetryCollectorReconciler{ + Client: mgr.GetClient(), + Scheme: mgr.GetScheme(), + }).SetupWithManager(mgr, options); err != nil { + return fmt.Errorf("failed to create controller %s: %v", "OpenTelemetryCollector", err) + } // +kubebuilder:scaffold:builder return nil } diff --git a/internal/controller/otelcollector_controller.go b/internal/controller/otelcollector_controller.go new file mode 100644 index 0000000000..ac6311c0b4 --- /dev/null +++ b/internal/controller/otelcollector_controller.go @@ -0,0 +1,39 @@ +// Copyright (c) 2026 Tigera, Inc. All rights reserved. + +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package controller + +import ( + "github.com/go-logr/logr" + + "k8s.io/apimachinery/pkg/runtime" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + + "github.com/tigera/operator/pkg/controller/options" + "github.com/tigera/operator/pkg/controller/otelcollector" +) + +type OpenTelemetryCollectorReconciler struct { + client.Client + Log logr.Logger + Scheme *runtime.Scheme +} + +// +kubebuilder:rbac:groups=operator.tigera.io,resources=logcollectors,verbs=get;list;watch;create;update;patch;delete +// +kubebuilder:rbac:groups=operator.tigera.io,resources=logcollectors/status,verbs=get;update;patch + +func (r *OpenTelemetryCollectorReconciler) SetupWithManager(mgr ctrl.Manager, opts options.ControllerOptions) error { + return otelcollector.Add(mgr, opts) +} diff --git a/pkg/common/common.go b/pkg/common/common.go index 485046c415..7ff2a67daa 100644 --- a/pkg/common/common.go +++ b/pkg/common/common.go @@ -36,6 +36,8 @@ const ( EgressAccessControlFeature = "egress-access-control" // PolicyRecommendation feature name PolicyRecommendationFeature = "policy-recommendation" + // OTelCollectorFeature gates the OTel Collector component in the license. + OTelCollectorFeature = "otel-collector" // MultipleOwnersLabel used to indicate multiple owner references. // If the render code places this label on an object, the object mergeState machinery will merge owner // references with any that already exist on the object rather than replace the owner references. Further diff --git a/pkg/common/validation/otelcollector/validation.go b/pkg/common/validation/otelcollector/validation.go new file mode 100644 index 0000000000..dc18e5ff4a --- /dev/null +++ b/pkg/common/validation/otelcollector/validation.go @@ -0,0 +1,27 @@ +// Copyright (c) 2026 Tigera, Inc. All rights reserved. + +// 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 otelcollector + +import ( + corev1 "k8s.io/api/core/v1" + + "github.com/tigera/operator/pkg/common/k8svalidation" + "k8s.io/apimachinery/pkg/util/validation/field" +) + +func ValidateOTelCollectorStatefulSetContainer(container corev1.Container) error { + errs := k8svalidation.ValidateResourceRequirements(&container.Resources, field.NewPath("spec", "template", "spec", "containers")) + return errs.ToAggregate() +} diff --git a/pkg/controller/logcollector/logcollector_controller.go b/pkg/controller/logcollector/logcollector_controller.go index 8dae009ddc..11cd8d72c2 100644 --- a/pkg/controller/logcollector/logcollector_controller.go +++ b/pkg/controller/logcollector/logcollector_controller.go @@ -186,6 +186,7 @@ func add(mgr manager.Manager, c ctrlruntime.Controller) error { if err = c.WatchObject(&operatorv1.NonClusterHost{}, &handler.EnqueueRequestForObject{}); err != nil { return fmt.Errorf("logcollector-controller failed to watch resource: %w", err) } + return nil } @@ -665,6 +666,8 @@ func (r *ReconcileLogCollector) Reconcile(ctx context.Context, request reconcile EKSLogForwarderKeyPair: eksLogForwarderKeyPair, NonClusterHost: nonclusterhost, LicenseExpired: licenseExpired, + OTelCollectorEnabled: instance.Spec.OTelCollector != nil, + OTelLogTypes: otelLogTypes(instance), } // Render the fluent-bit component for Linux. The same configuration drives // the shared and Windows components below; each applies its OS-specific @@ -918,3 +921,12 @@ func getUserCACertificate(client client.Client, name string) (certificatemanagem } return certificatemanagement.NewCertificate(name, common.OperatorNamespace(), []byte(cm.Data[corev1.TLSCertKey]), nil), nil } + +// otelLogTypes returns the log types selected for OTel export, empty when the +// otelCollector section or its logs selection is absent. +func otelLogTypes(lc *operatorv1.LogCollector) []operatorv1.OTelLogType { + if lc.Spec.OTelCollector == nil || lc.Spec.OTelCollector.Logs == nil { + return nil + } + return lc.Spec.OTelCollector.Logs.Types +} diff --git a/pkg/controller/otelcollector/controller.go b/pkg/controller/otelcollector/controller.go new file mode 100644 index 0000000000..90093875c9 --- /dev/null +++ b/pkg/controller/otelcollector/controller.go @@ -0,0 +1,327 @@ +// Copyright (c) 2026 Tigera, Inc. All rights reserved. + +// 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 otelcollector + +import ( + "context" + "fmt" + "time" + + appsv1 "k8s.io/api/apps/v1" + "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller" + "sigs.k8s.io/controller-runtime/pkg/handler" + logf "sigs.k8s.io/controller-runtime/pkg/log" + "sigs.k8s.io/controller-runtime/pkg/manager" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + + v3 "github.com/tigera/api/pkg/apis/projectcalico/v3" + + operatorv1 "github.com/tigera/operator/api/v1" + "github.com/tigera/operator/pkg/common" + "github.com/tigera/operator/pkg/common/validation" + otelvalidation "github.com/tigera/operator/pkg/common/validation/otelcollector" + "github.com/tigera/operator/pkg/controller/certificatemanager" + "github.com/tigera/operator/pkg/controller/options" + "github.com/tigera/operator/pkg/controller/status" + "github.com/tigera/operator/pkg/controller/utils" + "github.com/tigera/operator/pkg/controller/utils/imageset" + "github.com/tigera/operator/pkg/ctrlruntime" + "github.com/tigera/operator/pkg/dns" + "github.com/tigera/operator/pkg/render" + rcertificatemanagement "github.com/tigera/operator/pkg/render/certificatemanagement" + "github.com/tigera/operator/pkg/render/common/networkpolicy" + "github.com/tigera/operator/pkg/render/otelcollector" + "github.com/tigera/operator/pkg/tls/certificatemanagement" +) + +const ( + controllerName = "log-collector-otel-controller" + ResourceName = "log-collector-otel" +) + +var log = logf.Log.WithName(controllerName) + +func Add(mgr manager.Manager, opts options.ControllerOptions) error { + if !opts.EnterpriseCRDExists { + return nil + } + + licenseAPIReady := &utils.ReadyFlag{} + tierWatchReady := &utils.ReadyFlag{} + + statusManager := status.New(mgr.GetClient(), ResourceName, opts.KubernetesVersion) + reconciler := newReconciler(mgr.GetClient(), mgr.GetScheme(), statusManager, opts, licenseAPIReady, tierWatchReady) + + c, err := ctrlruntime.NewController(controllerName, mgr, controller.Options{Reconciler: reconciler}) + if err != nil { + return fmt.Errorf("failed to create %s: %w", controllerName, err) + } + + go utils.WaitToAddLicenseKeyWatch(c, opts.K8sClientset, log, licenseAPIReady) + go utils.WaitToAddTierWatch(networkpolicy.CalicoTierName, c, opts.K8sClientset, log, tierWatchReady) + go utils.WaitToAddNetworkPolicyWatches(c, opts.K8sClientset, log, []types.NamespacedName{ + {Name: otelcollector.OTelCollectorPolicyName, Namespace: otelcollector.OTelCollectorNamespace}, + }) + + if err = c.WatchObject(&operatorv1.LogCollector{}, &handler.EnqueueRequestForObject{}); err != nil { + return fmt.Errorf("%s failed to watch primary resource: %w", controllerName, err) + } + + if err = utils.AddAPIServerWatch(c); err != nil { + return fmt.Errorf("%s failed to watch APIServer resource: %w", controllerName, err) + } + + if err = utils.AddInstallationWatch(c); err != nil { + return fmt.Errorf("%s failed to watch Installation resource: %w", controllerName, err) + } + + if err = imageset.AddImageSetWatch(c); err != nil { + return fmt.Errorf("%s failed to watch ImageSet: %w", controllerName, err) + } + + if err = utils.AddTigeraStatusWatch(c, ResourceName); err != nil { + return fmt.Errorf("%s failed to watch TigeraStatus: %w", controllerName, err) + } + + if err = utils.AddPeriodicReconcile(c, utils.PeriodicReconcileTime, &handler.EnqueueRequestForObject{}); err != nil { + return fmt.Errorf("%s failed to create periodic reconcile watch: %w", controllerName, err) + } + + // Watch the workload and ConfigMap so out-of-band edits/deletes trigger reconcile. + if err = utils.AddNamespacedWatch(c, &appsv1.StatefulSet{ + TypeMeta: metav1.TypeMeta{Kind: "StatefulSet", APIVersion: "apps/v1"}, + ObjectMeta: metav1.ObjectMeta{Name: otelcollector.OTelCollectorStatefulSetName, Namespace: otelcollector.OTelCollectorNamespace}, + }, &handler.EnqueueRequestForObject{}); err != nil { + return fmt.Errorf("%s failed to watch StatefulSet: %w", controllerName, err) + } + + if err = utils.AddConfigMapWatch(c, otelcollector.OTelCollectorConfigMapName, otelcollector.OTelCollectorNamespace, &handler.EnqueueRequestForObject{}); err != nil { + return fmt.Errorf("%s failed to watch ConfigMap: %w", controllerName, err) + } + + if err = utils.AddSecretsWatch(c, otelcollector.OTelCollectorServerTLSSecretName, common.OperatorNamespace()); err != nil { + return fmt.Errorf("%s failed to watch the Secret resource(%s): %w", controllerName, otelcollector.OTelCollectorServerTLSSecretName, err) + } + + return nil +} + +func newReconciler( + cli client.Client, + schema *runtime.Scheme, + statusMgr status.StatusManager, + opts options.ControllerOptions, + licenseAPIReady *utils.ReadyFlag, + tierWatchReady *utils.ReadyFlag, +) *Reconciler { + r := &Reconciler{ + cli: cli, + scheme: schema, + status: statusMgr, + opts: opts, + licenseAPIReady: licenseAPIReady, + tierWatchReady: tierWatchReady, + } + r.status.Run(opts.ShutdownContext) + return r +} + +var _ reconcile.Reconciler = &Reconciler{} + +type Reconciler struct { + cli client.Client + scheme *runtime.Scheme + status status.StatusManager + opts options.ControllerOptions + licenseAPIReady *utils.ReadyFlag + tierWatchReady *utils.ReadyFlag +} + +func (r *Reconciler) Reconcile(ctx context.Context, request reconcile.Request) (reconcile.Result, error) { + reqLogger := log.WithValues("Request.Namespace", request.Namespace, "Request.Name", request.Name) + reqLogger.V(2).Info("Reconciling OTelCollector") + + logCollector, err := utils.GetIfExists[operatorv1.LogCollector](ctx, utils.DefaultEnterpriseInstanceKey, r.cli) + if err != nil { + r.status.SetDegraded(operatorv1.ResourceReadError, "Error querying LogCollector CR", err, reqLogger) + return reconcile.Result{}, err + } else if logCollector == nil { + r.status.OnCRNotFound() + return reconcile.Result{}, nil + } + + if logCollector.Spec.OTelCollector == nil { + r.status.OnCRNotFound() + return reconcile.Result{}, nil + } + + r.status.OnCRFound() + defer r.status.SetMetaData(&logCollector.ObjectMeta) + + variant, installationSpec, err := utils.GetInstallationSpec(ctx, r.cli) + if err != nil { + return reconcile.Result{}, err + } else if installationSpec == nil { + return reconcile.Result{}, nil + } + + if !utils.IsProjectCalicoV3Available(r.cli, r.opts, reqLogger) { + r.status.SetDegraded(operatorv1.ResourceNotReady, "Waiting for Tigera API server to be ready", nil, reqLogger) + return reconcile.Result{}, nil + } + + if !r.tierWatchReady.IsReady() { + r.status.SetDegraded(operatorv1.ResourceNotReady, "Waiting for Tier watch to be established", nil, reqLogger) + return reconcile.Result{RequeueAfter: utils.StandardRetry}, nil + } + + if err := r.cli.Get(ctx, client.ObjectKey{Name: networkpolicy.CalicoTierName}, &v3.Tier{}); err != nil { + if errors.IsNotFound(err) { + r.status.SetDegraded(operatorv1.ResourceNotReady, "Waiting for calico-system tier to be created, see the 'tiers' TigeraStatus for more information", err, reqLogger) + return reconcile.Result{RequeueAfter: utils.StandardRetry}, nil + } + r.status.SetDegraded(operatorv1.ResourceNotReady, "Error querying calico-system tier", err, reqLogger) + return reconcile.Result{}, err + } + + if !r.licenseAPIReady.IsReady() { + r.status.SetDegraded(operatorv1.ResourceNotReady, "Waiting for LicenseKeyAPI to be ready", nil, reqLogger) + return reconcile.Result{RequeueAfter: utils.StandardRetry}, nil + } + + license, err := utils.FetchLicenseKey(ctx, r.cli) + if err != nil { + if errors.IsNotFound(err) { + r.status.SetDegraded(operatorv1.ResourceNotFound, "License not found", err, reqLogger) + return reconcile.Result{RequeueAfter: utils.StandardRetry}, nil + } + r.status.SetDegraded(operatorv1.ResourceReadError, "Error querying license", err, reqLogger) + return reconcile.Result{RequeueAfter: utils.StandardRetry}, nil + } + if !utils.IsFeatureActive(license, common.OTelCollectorFeature) { + r.status.SetDegraded(operatorv1.ResourceValidationError, "Feature is not active - License does not support this feature", nil, reqLogger) + return reconcile.Result{}, nil + } + + gracePeriod := utils.ParseGracePeriod(license.Status.GracePeriod) + licenseStatus := utils.GetLicenseStatus(license, gracePeriod) + var graceRequeueAfter time.Duration + if licenseStatus == utils.LicenseStatusInGracePeriod { + reqLogger.Info("License has expired and is within the grace period. Please renew your license to avoid service disruption.") + graceRequeueAfter = time.Until(license.Status.Expiry.Add(gracePeriod)) + } + + if logCollector.Spec.OTelCollector.OTelCollectorStatefulSet != nil { + if err := validation.ValidateReplicatedPodResourceOverrides( + logCollector.Spec.OTelCollector.OTelCollectorStatefulSet, + otelvalidation.ValidateOTelCollectorStatefulSetContainer, + validation.NoContainersDefined, + ); err != nil { + r.status.SetDegraded(operatorv1.ResourceValidationError, "Invalid statefulSet overrides", err, reqLogger) + return reconcile.Result{}, err + } + } + + pullSecrets, err := utils.GetInstallationPullSecrets(installationSpec, r.cli) + if err != nil { + r.status.SetDegraded(operatorv1.ResourceReadError, "Error retrieving pull secrets", err, reqLogger) + return reconcile.Result{}, err + } + + var receiverTLSSecret certificatemanagement.KeyPairInterface + var trustedBundle certificatemanagement.TrustedBundle + + if logCollector.Spec.OTelCollector.HasDataSources() { + certMgr, err := certificatemanager.Create(r.cli, installationSpec, r.opts.ClusterDomain, common.OperatorNamespace()) + if err != nil { + r.status.SetDegraded(operatorv1.ResourceCreateError, "Unable to create the Tigera CA", err, reqLogger) + return reconcile.Result{}, err + } + + trustedBundle = certMgr.CreateTrustedBundle() + + if logCollector.Spec.OTelCollector.HasLogs() { + dnsNames := dns.GetServiceDNSNames(otelcollector.OTelCollectorServiceName, otelcollector.OTelCollectorNamespace, r.opts.ClusterDomain) + receiverTLSSecret, err = certMgr.GetOrCreateKeyPair(r.cli, otelcollector.OTelCollectorServerTLSSecretName, common.OperatorNamespace(), dnsNames) + if err != nil { + r.status.SetDegraded(operatorv1.ResourceCreateError, "Error creating OTel receiver TLS certificate", err, reqLogger) + return reconcile.Result{}, err + } + } + + certMgr.AddToStatusManager(r.status, otelcollector.OTelCollectorNamespace) + } + + cfg := &otelcollector.Configuration{ + PullSecrets: pullSecrets, + OpenShift: r.opts.DetectedProvider.IsOpenShift(), + Installation: installationSpec, + OTelCollector: logCollector.Spec.OTelCollector, + ReceiverTLSSecret: receiverTLSSecret, + TrustedCertBundle: trustedBundle, + } + + var keyPairOptions []rcertificatemanagement.KeyPairOption + if receiverTLSSecret != nil { + keyPairOptions = append(keyPairOptions, rcertificatemanagement.NewKeyPairOption(receiverTLSSecret, true, true)) + } + + otelComponent, err := otelcollector.OTelCollector(cfg) + if err != nil { + r.status.SetDegraded(operatorv1.ResourceRenderingError, "Error rendering OTel collector config", err, reqLogger) + return reconcile.Result{}, err + } + + components := []render.Component{ + otelComponent, + } + if logCollector.Spec.OTelCollector.HasDataSources() { + components = append(components, rcertificatemanagement.CertificateManagement(&rcertificatemanagement.Config{ + Namespace: otelcollector.OTelCollectorNamespace, + ServiceAccounts: []string{otelcollector.OTelCollectorServiceAccountName}, + KeyPairOptions: keyPairOptions, + TrustedBundle: trustedBundle, + })) + } + + ch := utils.NewComponentHandler(log, r.cli, r.scheme, logCollector) + if err = imageset.ApplyImageSet(ctx, r.cli, variant, components...); err != nil { + r.status.SetDegraded(operatorv1.ResourceUpdateError, "Error with images from ImageSet", err, reqLogger) + return reconcile.Result{}, err + } + + for _, component := range components { + if err := ch.CreateOrUpdateOrDelete(ctx, component, r.status); err != nil { + r.status.SetDegraded(operatorv1.ResourceUpdateError, "Error creating / updating resource", err, reqLogger) + return reconcile.Result{}, err + } + } + + if licenseStatus == utils.LicenseStatusExpired { + r.status.SetDegraded(operatorv1.ResourceValidationError, + "License is expired - OTel collector forwarding is stopped. Contact Tigera support or email licensing@tigera.io", nil, reqLogger) + return reconcile.Result{}, nil + } + + r.status.ReadyToMonitor() + r.status.ClearDegraded() + + return reconcile.Result{RequeueAfter: graceRequeueAfter}, nil +} diff --git a/pkg/controller/otelcollector/otelcollector_controller_test.go b/pkg/controller/otelcollector/otelcollector_controller_test.go new file mode 100644 index 0000000000..d3f5d0b4c7 --- /dev/null +++ b/pkg/controller/otelcollector/otelcollector_controller_test.go @@ -0,0 +1,250 @@ +// Copyright (c) 2026 Tigera, Inc. All rights reserved. + +// 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 otelcollector + +import ( + "context" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "github.com/stretchr/testify/mock" + + appsv1 "k8s.io/api/apps/v1" + rbacv1 "k8s.io/api/rbac/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + + v3 "github.com/tigera/api/pkg/apis/projectcalico/v3" + operatorv1 "github.com/tigera/operator/api/v1" + "github.com/tigera/operator/pkg/apis" + "github.com/tigera/operator/pkg/common" + "github.com/tigera/operator/pkg/controller/certificatemanager" + "github.com/tigera/operator/pkg/controller/options" + "github.com/tigera/operator/pkg/controller/status" + "github.com/tigera/operator/pkg/controller/utils" + ctrlrfake "github.com/tigera/operator/pkg/ctrlruntime/client/fake" + "github.com/tigera/operator/pkg/dns" + "github.com/tigera/operator/pkg/render/common/networkpolicy" + "github.com/tigera/operator/test" +) + +var _ = Describe("OTelCollector controller tests", func() { + var ( + cli client.Client + scheme *runtime.Scheme + ctx context.Context + mockStatus *status.MockStatus + r *Reconciler + install *operatorv1.Installation + licenseAPIReady *utils.ReadyFlag + tierWatchReady *utils.ReadyFlag + ) + + BeforeEach(func() { + scheme = runtime.NewScheme() + Expect(apis.AddToScheme(scheme, false)).ShouldNot(HaveOccurred()) + Expect(appsv1.SchemeBuilder.AddToScheme(scheme)).ShouldNot(HaveOccurred()) + Expect(rbacv1.SchemeBuilder.AddToScheme(scheme)).ShouldNot(HaveOccurred()) + + ctx = context.Background() + cli = ctrlrfake.DefaultFakeClientBuilder(scheme).Build() + + replicas := int32(2) + install = &operatorv1.Installation{ + ObjectMeta: metav1.ObjectMeta{Name: "default", Generation: 2}, + Status: operatorv1.InstallationStatus{ + Variant: operatorv1.CalicoEnterprise, + Computed: &operatorv1.InstallationSpec{}, + }, + Spec: operatorv1.InstallationSpec{ + ControlPlaneReplicas: &replicas, + Variant: operatorv1.CalicoEnterprise, + Registry: "some.registry.org/", + }, + } + Expect(cli.Create(ctx, install)).ToNot(HaveOccurred()) + + Expect(cli.Create(ctx, &v3.LicenseKey{ + ObjectMeta: metav1.ObjectMeta{Name: "default"}, + Status: v3.LicenseKeyStatus{ + Features: []string{common.OTelCollectorFeature}, + }, + })).ToNot(HaveOccurred()) + + // Create a CA secret so the certificate manager can issue keypairs. + cm, err := certificatemanager.Create(cli, &install.Spec, dns.DefaultClusterDomain, common.OperatorNamespace(), certificatemanager.AllowCACreation()) + Expect(err).ShouldNot(HaveOccurred()) + Expect(cli.Create(ctx, cm.KeyPair().Secret(common.OperatorNamespace()))).ShouldNot(HaveOccurred()) + + // Create the calico-system tier so the controller's tier check passes. + Expect(cli.Create(ctx, &v3.Tier{ + ObjectMeta: metav1.ObjectMeta{Name: networkpolicy.CalicoTierName}, + })).ToNot(HaveOccurred()) + + Expect(cli.Create(ctx, &operatorv1.APIServer{ + ObjectMeta: metav1.ObjectMeta{Name: "tigera-secure"}, + Status: operatorv1.APIServerStatus{State: operatorv1.TigeraStatusReady}, + })).ToNot(HaveOccurred()) + + licenseAPIReady = &utils.ReadyFlag{} + licenseAPIReady.MarkAsReady() + tierWatchReady = &utils.ReadyFlag{} + tierWatchReady.MarkAsReady() + + mockStatus = &status.MockStatus{} + mockStatus.On("AddStatefulSets", mock.Anything).Return() + mockStatus.On("AddCertificateSigningRequests", mock.Anything).Return() + mockStatus.On("RemoveCertificateSigningRequests", mock.Anything).Return() + mockStatus.On("IsAvailable").Return(true) + mockStatus.On("OnCRFound").Return() + mockStatus.On("OnCRNotFound").Return() + mockStatus.On("ClearDegraded") + mockStatus.On("ReadyToMonitor") + mockStatus.On("SetMetaData", mock.Anything).Return() + mockStatus.On("SetDegraded", mock.Anything, mock.AnythingOfType("string"), mock.Anything, mock.Anything).Return().Maybe() + mockStatus.On("ClearWarning", mock.AnythingOfType("string")).Return().Maybe() + + r = &Reconciler{ + cli: cli, + scheme: scheme, + status: mockStatus, + licenseAPIReady: licenseAPIReady, + tierWatchReady: tierWatchReady, + opts: options.ControllerOptions{ + DetectedProvider: operatorv1.ProviderNone, + EnterpriseCRDExists: true, + ClusterDomain: dns.DefaultClusterDomain, + }, + } + }) + + Context("CR not found", func() { + It("should call OnCRNotFound and return without error", func() { + _, err := r.Reconcile(ctx, reconcile.Request{}) + Expect(err).ShouldNot(HaveOccurred()) + mockStatus.AssertCalled(GinkgoT(), "OnCRNotFound") + }) + }) + + Context("LogCollector without OTelCollector", func() { + BeforeEach(func() { + Expect(cli.Create(ctx, &operatorv1.LogCollector{ + ObjectMeta: metav1.ObjectMeta{Name: "tigera-secure"}, + Spec: operatorv1.LogCollectorSpec{}, + })).ToNot(HaveOccurred()) + }) + + It("should call OnCRNotFound when OTelCollector is nil", func() { + _, err := r.Reconcile(ctx, reconcile.Request{}) + Expect(err).ShouldNot(HaveOccurred()) + mockStatus.AssertCalled(GinkgoT(), "OnCRNotFound") + }) + }) + + Context("happy path", func() { + BeforeEach(func() { + Expect(cli.Create(ctx, &operatorv1.LogCollector{ + ObjectMeta: metav1.ObjectMeta{Name: "tigera-secure"}, + Spec: operatorv1.LogCollectorSpec{ + OTelCollector: &operatorv1.OTelCollectorSpec{ + Logs: &operatorv1.OTelLogs{Types: []operatorv1.OTelLogType{operatorv1.OTelFlowLog}}, + Exporters: []operatorv1.OTelExporter{{Name: "backend", Endpoint: "otlp.example.com:4317"}}, + }, + }, + })).ToNot(HaveOccurred()) + }) + + It("should reconcile and create resources", func() { + _, err := r.Reconcile(ctx, reconcile.Request{}) + Expect(err).ShouldNot(HaveOccurred()) + + mockStatus.AssertCalled(GinkgoT(), "OnCRFound") + mockStatus.AssertCalled(GinkgoT(), "ReadyToMonitor") + mockStatus.AssertCalled(GinkgoT(), "ClearDegraded") + + ss := appsv1.StatefulSet{ObjectMeta: metav1.ObjectMeta{Name: "otel-collector", Namespace: "calico-system"}} + Expect(test.GetResource(cli, &ss)).To(BeNil()) + Expect(ss.Spec.Template.Spec.Containers).To(HaveLen(1)) + }) + }) + + Context("license missing", func() { + BeforeEach(func() { + Expect(cli.Delete(ctx, &v3.LicenseKey{ObjectMeta: metav1.ObjectMeta{Name: "default"}})).ToNot(HaveOccurred()) + Expect(cli.Create(ctx, &operatorv1.LogCollector{ + ObjectMeta: metav1.ObjectMeta{Name: "tigera-secure"}, + Spec: operatorv1.LogCollectorSpec{ + OTelCollector: &operatorv1.OTelCollectorSpec{ + Exporters: []operatorv1.OTelExporter{{Name: "backend", Endpoint: "otlp.example.com:4317"}}, + }, + }, + })).ToNot(HaveOccurred()) + }) + + It("should set degraded status", func() { + _, err := r.Reconcile(ctx, reconcile.Request{}) + Expect(err).ShouldNot(HaveOccurred()) + mockStatus.AssertCalled(GinkgoT(), "SetDegraded", operatorv1.ResourceNotFound, mock.AnythingOfType("string"), mock.Anything, mock.Anything) + }) + }) + + Context("license feature inactive", func() { + BeforeEach(func() { + Expect(cli.Delete(ctx, &v3.LicenseKey{ObjectMeta: metav1.ObjectMeta{Name: "default"}})).ToNot(HaveOccurred()) + Expect(cli.Create(ctx, &v3.LicenseKey{ + ObjectMeta: metav1.ObjectMeta{Name: "default"}, + Status: v3.LicenseKeyStatus{ + Features: []string{"some-other-feature"}, + }, + })).ToNot(HaveOccurred()) + + Expect(cli.Create(ctx, &operatorv1.LogCollector{ + ObjectMeta: metav1.ObjectMeta{Name: "tigera-secure"}, + Spec: operatorv1.LogCollectorSpec{ + OTelCollector: &operatorv1.OTelCollectorSpec{ + Exporters: []operatorv1.OTelExporter{{Name: "backend", Endpoint: "otlp.example.com:4317"}}, + }, + }, + })).ToNot(HaveOccurred()) + }) + + It("should set degraded status for inactive feature", func() { + _, err := r.Reconcile(ctx, reconcile.Request{}) + Expect(err).ShouldNot(HaveOccurred()) + mockStatus.AssertCalled(GinkgoT(), "SetDegraded", operatorv1.ResourceValidationError, mock.AnythingOfType("string"), mock.Anything, mock.Anything) + }) + }) + + Context("installation missing", func() { + BeforeEach(func() { + Expect(cli.Delete(ctx, &operatorv1.Installation{ObjectMeta: metav1.ObjectMeta{Name: "default"}})).ToNot(HaveOccurred()) + Expect(cli.Create(ctx, &operatorv1.LogCollector{ + ObjectMeta: metav1.ObjectMeta{Name: "tigera-secure"}, + Spec: operatorv1.LogCollectorSpec{ + OTelCollector: &operatorv1.OTelCollectorSpec{ + Exporters: []operatorv1.OTelExporter{{Name: "backend", Endpoint: "otlp.example.com:4317"}}, + }, + }, + })).ToNot(HaveOccurred()) + }) + + It("should return error when installation is missing", func() { + _, err := r.Reconcile(ctx, reconcile.Request{}) + Expect(err).Should(HaveOccurred()) + }) + }) +}) diff --git a/pkg/controller/otelcollector/otelcollector_suite_test.go b/pkg/controller/otelcollector/otelcollector_suite_test.go new file mode 100644 index 0000000000..860f89ef35 --- /dev/null +++ b/pkg/controller/otelcollector/otelcollector_suite_test.go @@ -0,0 +1,33 @@ +// Copyright (c) 2026 Tigera, Inc. All rights reserved. + +// 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 otelcollector + +import ( + "testing" + + "github.com/onsi/ginkgo/v2" + "github.com/onsi/gomega" + uzap "go.uber.org/zap" + logf "sigs.k8s.io/controller-runtime/pkg/log" + "sigs.k8s.io/controller-runtime/pkg/log/zap" +) + +func TestController(t *testing.T) { + logf.SetLogger(zap.New(zap.WriteTo(ginkgo.GinkgoWriter), zap.UseDevMode(true), zap.Level(uzap.NewAtomicLevelAt(uzap.DebugLevel)))) + gomega.RegisterFailHandler(ginkgo.Fail) + suiteConfig, reporterConfig := ginkgo.GinkgoConfiguration() + reporterConfig.JUnitReport = "../../../report/ut/otelcollector_controller_suite.xml" + ginkgo.RunSpecs(t, "pkg/controller/otelcollector Controller Suite", suiteConfig, reporterConfig) +} diff --git a/pkg/imports/crds/operator/operator.tigera.io_logcollectors.yaml b/pkg/imports/crds/operator/operator.tigera.io_logcollectors.yaml index 483dc0a148..90ac5f30e1 100644 --- a/pkg/imports/crds/operator/operator.tigera.io_logcollectors.yaml +++ b/pkg/imports/crds/operator/operator.tigera.io_logcollectors.yaml @@ -963,6 +963,1531 @@ spec: If running as a multi-tenant management cluster, the namespace in which the management cluster's tenant services are running. type: string + otelCollector: + description: + OTelCollector configures the OpenTelemetry Collector + for exporting logs and metrics via OTLP. + properties: + exporters: + description: Exporters configures the OTLP export endpoints. + items: + description: OTelExporter defines an OTLP export endpoint. + properties: + endpoint: + description: Endpoint is the OTLP endpoint URL. + type: string + name: + description: Name is a unique identifier for this exporter. + type: string + protocol: + default: grpc + description: + "Protocol specifies the OTLP transport protocol. + Default: grpc." + enum: + - grpc + - http + type: string + tlsInsecure: + description: |- + TLSInsecure disables TLS verification for this exporter. Only use for trusted in-cluster targets. + Default: false + type: boolean + required: + - endpoint + - name + type: object + type: array + logs: + description: + Logs configures which log types are exported via + OTLP. + properties: + types: + description: + "Types specifies which log types to export. Supported + values: Audit, DNS, Flows." + items: + description: + OTelLogType represents the allowable log types + for OTel export. + enum: + - Audit + - DNS + - Flows + type: string + type: array + type: object + metrics: + description: + Metrics configures whether Calico component metrics + are exported via OTLP. + properties: + enabled: + description: |- + Enabled specifies whether to scrape and export Calico component metrics via OTLP. + Default: Disabled + enum: + - Enabled + - Disabled + type: string + type: object + openTelemetryCollectorStatefulSet: + description: + OTelCollectorStatefulSet configures the OTel Collector + StatefulSet. + properties: + metadata: + description: + Metadata is a subset of a Kubernetes object's + metadata that is added to the StatefulSet. + properties: + annotations: + additionalProperties: + type: string + description: |- + Annotations is a map of arbitrary non-identifying metadata. Each of these + key/value pairs are added to the object's annotations provided the key does not + already exist in the object's annotations. + type: object + labels: + additionalProperties: + type: string + description: |- + Labels is a map of string keys and values that may match replicaset and + service selectors. Each of these key/value pairs are added to the + object's labels provided the key does not already exist in the object's labels. + type: object + type: object + spec: + description: + Spec is the specification of the OTel Collector + StatefulSet. + properties: + minReadySeconds: + description: |- + MinReadySeconds is the minimum number of seconds for which a newly created StatefulSet pod should + be ready without any of its container crashing, for it to be considered available. + If specified, this overrides any minReadySeconds value that may be set on the OTel Collector StatefulSet. + If omitted, the OTel Collector StatefulSet will use its default value for minReadySeconds. + format: int32 + maximum: 2147483647 + minimum: 0 + type: integer + template: + description: + Template describes the OTel Collector StatefulSet + pod that will be created. + properties: + metadata: + description: + Metadata is a subset of a Kubernetes + object's metadata that is added to the pod's metadata. + properties: + annotations: + additionalProperties: + type: string + description: |- + Annotations is a map of arbitrary non-identifying metadata. Each of these + key/value pairs are added to the object's annotations provided the key does not + already exist in the object's annotations. + type: object + labels: + additionalProperties: + type: string + description: |- + Labels is a map of string keys and values that may match replicaset and + service selectors. Each of these key/value pairs are added to the + object's labels provided the key does not already exist in the object's labels. + type: object + type: object + spec: + description: + Spec is the OTel Collector StatefulSet's + PodSpec. + properties: + affinity: + description: + Affinity is a group of affinity scheduling + rules for the OTel Collector pods. + properties: + nodeAffinity: + description: + Describes node affinity scheduling + rules for the pod. + properties: + preferredDuringSchedulingIgnoredDuringExecution: + description: |- + The scheduler will prefer to schedule pods to nodes that satisfy + the affinity expressions specified by this field, but it may choose + a node that violates one or more of the expressions. The node that is + most preferred is the one with the greatest sum of weights, i.e. + for each node that meets all of the scheduling requirements (resource + request, requiredDuringScheduling affinity expressions, etc.), + compute a sum by iterating through the elements of this field and adding + "weight" to the sum if the node matches the corresponding matchExpressions; the + node(s) with the highest sum are the most preferred. + items: + description: |- + An empty preferred scheduling term matches all objects with implicit weight 0 + (i.e. it's a no-op). A null preferred scheduling term matches no objects (i.e. is also a no-op). + properties: + preference: + description: + A node selector term, + associated with the corresponding + weight. + properties: + matchExpressions: + description: + A list of node + selector requirements by node's + labels. + items: + description: |- + A node selector requirement is a selector that contains values, a key, and an operator + that relates the key and values. + properties: + key: + description: + The label + key that the selector + applies to. + type: string + operator: + description: |- + Represents a key's relationship to a set of values. + Valid operators are In, NotIn, Exists, DoesNotExist. Gt, and Lt. + type: string + values: + description: |- + 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. If the operator is Gt or Lt, the values + array must have a single element, which will be interpreted as an integer. + 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 + matchFields: + description: + A list of node + selector requirements by node's + fields. + items: + description: |- + A node selector requirement is a selector that contains values, a key, and an operator + that relates the key and values. + properties: + key: + description: + The label + key that the selector + applies to. + type: string + operator: + description: |- + Represents a key's relationship to a set of values. + Valid operators are In, NotIn, Exists, DoesNotExist. Gt, and Lt. + type: string + values: + description: |- + 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. If the operator is Gt or Lt, the values + array must have a single element, which will be interpreted as an integer. + 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 + type: object + x-kubernetes-map-type: atomic + weight: + description: + Weight associated with + matching the corresponding nodeSelectorTerm, + in the range 1-100. + format: int32 + type: integer + required: + - preference + - weight + type: object + type: array + x-kubernetes-list-type: atomic + requiredDuringSchedulingIgnoredDuringExecution: + description: |- + If the affinity requirements specified by this field are not met at + scheduling time, the pod will not be scheduled onto the node. + If the affinity requirements specified by this field cease to be met + at some point during pod execution (e.g. due to an update), the system + may or may not try to eventually evict the pod from its node. + properties: + nodeSelectorTerms: + description: + Required. A list of node + selector terms. The terms are ORed. + items: + description: |- + A null or empty node selector term matches no objects. The requirements of + them are ANDed. + The TopologySelectorTerm type implements a subset of the NodeSelectorTerm. + properties: + matchExpressions: + description: + A list of node + selector requirements by node's + labels. + items: + description: |- + A node selector requirement is a selector that contains values, a key, and an operator + that relates the key and values. + properties: + key: + description: + The label + key that the selector + applies to. + type: string + operator: + description: |- + Represents a key's relationship to a set of values. + Valid operators are In, NotIn, Exists, DoesNotExist. Gt, and Lt. + type: string + values: + description: |- + 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. If the operator is Gt or Lt, the values + array must have a single element, which will be interpreted as an integer. + 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 + matchFields: + description: + A list of node + selector requirements by node's + fields. + items: + description: |- + A node selector requirement is a selector that contains values, a key, and an operator + that relates the key and values. + properties: + key: + description: + The label + key that the selector + applies to. + type: string + operator: + description: |- + Represents a key's relationship to a set of values. + Valid operators are In, NotIn, Exists, DoesNotExist. Gt, and Lt. + type: string + values: + description: |- + 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. If the operator is Gt or Lt, the values + array must have a single element, which will be interpreted as an integer. + 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 + type: object + x-kubernetes-map-type: atomic + type: array + x-kubernetes-list-type: atomic + required: + - nodeSelectorTerms + type: object + x-kubernetes-map-type: atomic + type: object + podAffinity: + description: + Describes pod affinity scheduling + rules (e.g. co-locate this pod in the same + node, zone, etc. as some other pod(s)). + properties: + preferredDuringSchedulingIgnoredDuringExecution: + description: |- + The scheduler will prefer to schedule pods to nodes that satisfy + the affinity expressions specified by this field, but it may choose + a node that violates one or more of the expressions. The node that is + most preferred is the one with the greatest sum of weights, i.e. + for each node that meets all of the scheduling requirements (resource + request, requiredDuringScheduling affinity expressions, etc.), + compute a sum by iterating through the elements of this field and adding + "weight" to the sum if the node has pods which matches the corresponding podAffinityTerm; the + node(s) with the highest sum are the most preferred. + items: + description: + The weights of all of the + matched WeightedPodAffinityTerm fields + are added per-node to find the most + preferred node(s) + properties: + podAffinityTerm: + description: + Required. A pod affinity + term, associated with the corresponding + weight. + properties: + labelSelector: + description: |- + A label query over a set of resources, in this case pods. + If it's null, this PodAffinityTerm matches with no Pods. + 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 + matchLabelKeys: + description: |- + MatchLabelKeys is a set of pod label keys to select which pods will + be taken into consideration. The keys are used to lookup values from the + incoming pod labels, those key-value labels are merged with `labelSelector` as `key in (value)` + to select the group of existing pods which pods will be taken into consideration + for the incoming pod's pod (anti) affinity. Keys that don't exist in the incoming + pod labels will be ignored. The default value is empty. + The same key is forbidden to exist in both matchLabelKeys and labelSelector. + Also, matchLabelKeys cannot be set when labelSelector isn't set. + items: + type: string + type: array + x-kubernetes-list-type: atomic + mismatchLabelKeys: + description: |- + MismatchLabelKeys is a set of pod label keys to select which pods will + be taken into consideration. The keys are used to lookup values from the + incoming pod labels, those key-value labels are merged with `labelSelector` as `key notin (value)` + to select the group of existing pods which pods will be taken into consideration + for the incoming pod's pod (anti) affinity. Keys that don't exist in the incoming + pod labels will be ignored. The default value is empty. + The same key is forbidden to exist in both mismatchLabelKeys and labelSelector. + Also, mismatchLabelKeys cannot be set when labelSelector isn't set. + items: + type: string + type: array + x-kubernetes-list-type: atomic + namespaceSelector: + description: |- + A label query over the set of namespaces that the term applies to. + The term is applied to the union of the namespaces selected by this field + and the ones listed in the namespaces field. + null selector and null or empty namespaces list means "this pod's namespace". + An empty selector ({}) matches all namespaces. + 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 + namespaces: + description: |- + namespaces specifies a static list of namespace names that the term applies to. + The term is applied to the union of the namespaces listed in this field + and the ones selected by namespaceSelector. + null or empty namespaces list and null namespaceSelector means "this pod's namespace". + items: + type: string + type: array + x-kubernetes-list-type: atomic + topologyKey: + description: |- + This pod should be co-located (affinity) or not co-located (anti-affinity) with the pods matching + the labelSelector in the specified namespaces, where co-located is defined as running on a node + whose value of the label with key topologyKey matches that of any node on which any of the + selected pods is running. + Empty topologyKey is not allowed. + type: string + required: + - topologyKey + type: object + weight: + description: |- + weight associated with matching the corresponding podAffinityTerm, + in the range 1-100. + format: int32 + type: integer + required: + - podAffinityTerm + - weight + type: object + type: array + x-kubernetes-list-type: atomic + requiredDuringSchedulingIgnoredDuringExecution: + description: |- + If the affinity requirements specified by this field are not met at + scheduling time, the pod will not be scheduled onto the node. + If the affinity requirements specified by this field cease to be met + at some point during pod execution (e.g. due to a pod label update), the + system may or may not try to eventually evict the pod from its node. + When there are multiple elements, the lists of nodes corresponding to each + podAffinityTerm are intersected, i.e. all terms must be satisfied. + items: + description: |- + Defines a set of pods (namely those matching the labelSelector + relative to the given namespace(s)) that this pod should be + co-located (affinity) or not co-located (anti-affinity) with, + where co-located is defined as running on a node whose value of + the label with key matches that of any node on which + a pod of the set of pods is running + properties: + labelSelector: + description: |- + A label query over a set of resources, in this case pods. + If it's null, this PodAffinityTerm matches with no Pods. + 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 + matchLabelKeys: + description: |- + MatchLabelKeys is a set of pod label keys to select which pods will + be taken into consideration. The keys are used to lookup values from the + incoming pod labels, those key-value labels are merged with `labelSelector` as `key in (value)` + to select the group of existing pods which pods will be taken into consideration + for the incoming pod's pod (anti) affinity. Keys that don't exist in the incoming + pod labels will be ignored. The default value is empty. + The same key is forbidden to exist in both matchLabelKeys and labelSelector. + Also, matchLabelKeys cannot be set when labelSelector isn't set. + items: + type: string + type: array + x-kubernetes-list-type: atomic + mismatchLabelKeys: + description: |- + MismatchLabelKeys is a set of pod label keys to select which pods will + be taken into consideration. The keys are used to lookup values from the + incoming pod labels, those key-value labels are merged with `labelSelector` as `key notin (value)` + to select the group of existing pods which pods will be taken into consideration + for the incoming pod's pod (anti) affinity. Keys that don't exist in the incoming + pod labels will be ignored. The default value is empty. + The same key is forbidden to exist in both mismatchLabelKeys and labelSelector. + Also, mismatchLabelKeys cannot be set when labelSelector isn't set. + items: + type: string + type: array + x-kubernetes-list-type: atomic + namespaceSelector: + description: |- + A label query over the set of namespaces that the term applies to. + The term is applied to the union of the namespaces selected by this field + and the ones listed in the namespaces field. + null selector and null or empty namespaces list means "this pod's namespace". + An empty selector ({}) matches all namespaces. + 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 + namespaces: + description: |- + namespaces specifies a static list of namespace names that the term applies to. + The term is applied to the union of the namespaces listed in this field + and the ones selected by namespaceSelector. + null or empty namespaces list and null namespaceSelector means "this pod's namespace". + items: + type: string + type: array + x-kubernetes-list-type: atomic + topologyKey: + description: |- + This pod should be co-located (affinity) or not co-located (anti-affinity) with the pods matching + the labelSelector in the specified namespaces, where co-located is defined as running on a node + whose value of the label with key topologyKey matches that of any node on which any of the + selected pods is running. + Empty topologyKey is not allowed. + type: string + required: + - topologyKey + type: object + type: array + x-kubernetes-list-type: atomic + type: object + podAntiAffinity: + description: + Describes pod anti-affinity scheduling + rules (e.g. avoid putting this pod in the + same node, zone, etc. as some other pod(s)). + properties: + preferredDuringSchedulingIgnoredDuringExecution: + description: |- + The scheduler will prefer to schedule pods to nodes that satisfy + the anti-affinity expressions specified by this field, but it may choose + a node that violates one or more of the expressions. The node that is + most preferred is the one with the greatest sum of weights, i.e. + for each node that meets all of the scheduling requirements (resource + request, requiredDuringScheduling anti-affinity expressions, etc.), + compute a sum by iterating through the elements of this field and subtracting + "weight" from the sum if the node has pods which matches the corresponding podAffinityTerm; the + node(s) with the highest sum are the most preferred. + items: + description: + The weights of all of the + matched WeightedPodAffinityTerm fields + are added per-node to find the most + preferred node(s) + properties: + podAffinityTerm: + description: + Required. A pod affinity + term, associated with the corresponding + weight. + properties: + labelSelector: + description: |- + A label query over a set of resources, in this case pods. + If it's null, this PodAffinityTerm matches with no Pods. + 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 + matchLabelKeys: + description: |- + MatchLabelKeys is a set of pod label keys to select which pods will + be taken into consideration. The keys are used to lookup values from the + incoming pod labels, those key-value labels are merged with `labelSelector` as `key in (value)` + to select the group of existing pods which pods will be taken into consideration + for the incoming pod's pod (anti) affinity. Keys that don't exist in the incoming + pod labels will be ignored. The default value is empty. + The same key is forbidden to exist in both matchLabelKeys and labelSelector. + Also, matchLabelKeys cannot be set when labelSelector isn't set. + items: + type: string + type: array + x-kubernetes-list-type: atomic + mismatchLabelKeys: + description: |- + MismatchLabelKeys is a set of pod label keys to select which pods will + be taken into consideration. The keys are used to lookup values from the + incoming pod labels, those key-value labels are merged with `labelSelector` as `key notin (value)` + to select the group of existing pods which pods will be taken into consideration + for the incoming pod's pod (anti) affinity. Keys that don't exist in the incoming + pod labels will be ignored. The default value is empty. + The same key is forbidden to exist in both mismatchLabelKeys and labelSelector. + Also, mismatchLabelKeys cannot be set when labelSelector isn't set. + items: + type: string + type: array + x-kubernetes-list-type: atomic + namespaceSelector: + description: |- + A label query over the set of namespaces that the term applies to. + The term is applied to the union of the namespaces selected by this field + and the ones listed in the namespaces field. + null selector and null or empty namespaces list means "this pod's namespace". + An empty selector ({}) matches all namespaces. + 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 + namespaces: + description: |- + namespaces specifies a static list of namespace names that the term applies to. + The term is applied to the union of the namespaces listed in this field + and the ones selected by namespaceSelector. + null or empty namespaces list and null namespaceSelector means "this pod's namespace". + items: + type: string + type: array + x-kubernetes-list-type: atomic + topologyKey: + description: |- + This pod should be co-located (affinity) or not co-located (anti-affinity) with the pods matching + the labelSelector in the specified namespaces, where co-located is defined as running on a node + whose value of the label with key topologyKey matches that of any node on which any of the + selected pods is running. + Empty topologyKey is not allowed. + type: string + required: + - topologyKey + type: object + weight: + description: |- + weight associated with matching the corresponding podAffinityTerm, + in the range 1-100. + format: int32 + type: integer + required: + - podAffinityTerm + - weight + type: object + type: array + x-kubernetes-list-type: atomic + requiredDuringSchedulingIgnoredDuringExecution: + description: |- + If the anti-affinity requirements specified by this field are not met at + scheduling time, the pod will not be scheduled onto the node. + If the anti-affinity requirements specified by this field cease to be met + at some point during pod execution (e.g. due to a pod label update), the + system may or may not try to eventually evict the pod from its node. + When there are multiple elements, the lists of nodes corresponding to each + podAffinityTerm are intersected, i.e. all terms must be satisfied. + items: + description: |- + Defines a set of pods (namely those matching the labelSelector + relative to the given namespace(s)) that this pod should be + co-located (affinity) or not co-located (anti-affinity) with, + where co-located is defined as running on a node whose value of + the label with key matches that of any node on which + a pod of the set of pods is running + properties: + labelSelector: + description: |- + A label query over a set of resources, in this case pods. + If it's null, this PodAffinityTerm matches with no Pods. + 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 + matchLabelKeys: + description: |- + MatchLabelKeys is a set of pod label keys to select which pods will + be taken into consideration. The keys are used to lookup values from the + incoming pod labels, those key-value labels are merged with `labelSelector` as `key in (value)` + to select the group of existing pods which pods will be taken into consideration + for the incoming pod's pod (anti) affinity. Keys that don't exist in the incoming + pod labels will be ignored. The default value is empty. + The same key is forbidden to exist in both matchLabelKeys and labelSelector. + Also, matchLabelKeys cannot be set when labelSelector isn't set. + items: + type: string + type: array + x-kubernetes-list-type: atomic + mismatchLabelKeys: + description: |- + MismatchLabelKeys is a set of pod label keys to select which pods will + be taken into consideration. The keys are used to lookup values from the + incoming pod labels, those key-value labels are merged with `labelSelector` as `key notin (value)` + to select the group of existing pods which pods will be taken into consideration + for the incoming pod's pod (anti) affinity. Keys that don't exist in the incoming + pod labels will be ignored. The default value is empty. + The same key is forbidden to exist in both mismatchLabelKeys and labelSelector. + Also, mismatchLabelKeys cannot be set when labelSelector isn't set. + items: + type: string + type: array + x-kubernetes-list-type: atomic + namespaceSelector: + description: |- + A label query over the set of namespaces that the term applies to. + The term is applied to the union of the namespaces selected by this field + and the ones listed in the namespaces field. + null selector and null or empty namespaces list means "this pod's namespace". + An empty selector ({}) matches all namespaces. + 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 + namespaces: + description: |- + namespaces specifies a static list of namespace names that the term applies to. + The term is applied to the union of the namespaces listed in this field + and the ones selected by namespaceSelector. + null or empty namespaces list and null namespaceSelector means "this pod's namespace". + items: + type: string + type: array + x-kubernetes-list-type: atomic + topologyKey: + description: |- + This pod should be co-located (affinity) or not co-located (anti-affinity) with the pods matching + the labelSelector in the specified namespaces, where co-located is defined as running on a node + whose value of the label with key topologyKey matches that of any node on which any of the + selected pods is running. + Empty topologyKey is not allowed. + type: string + required: + - topologyKey + type: object + type: array + x-kubernetes-list-type: atomic + type: object + type: object + containers: + description: |- + Containers is a list of OTel Collector containers. + If specified, this overrides the specified OTel Collector StatefulSet containers. + If omitted, the OTel Collector StatefulSet will use its default values for its containers. + items: + description: + OTelCollectorStatefulSetContainer + is an OTel Collector StatefulSet container. + properties: + livenessProbe: + description: |- + LivenessProbe allows customization of the liveness probe timing parameters. + The probe handler is set by the operator and cannot be overridden. + properties: + failureThreshold: + description: |- + FailureThreshold is the minimum consecutive failures for the probe + to be considered failed after having succeeded. + format: int32 + type: integer + initialDelaySeconds: + description: |- + InitialDelaySeconds is the number of seconds after the container + starts before the probe is initiated. + format: int32 + type: integer + periodSeconds: + description: + PeriodSeconds is how often + (in seconds) to perform the probe. + format: int32 + type: integer + timeoutSeconds: + description: + TimeoutSeconds is the number + of seconds after which the probe times + out. + format: int32 + type: integer + type: object + name: + description: |- + Name is an enum which identifies the OTel Collector StatefulSet container by name. + Supported values are: otel-collector + enum: + - otel-collector + type: string + readinessProbe: + description: |- + ReadinessProbe allows customization of the readiness probe timing parameters. + The probe handler is set by the operator and cannot be overridden. + properties: + failureThreshold: + description: |- + FailureThreshold is the minimum consecutive failures for the probe + to be considered failed after having succeeded. + format: int32 + type: integer + initialDelaySeconds: + description: |- + InitialDelaySeconds is the number of seconds after the container + starts before the probe is initiated. + format: int32 + type: integer + periodSeconds: + description: + PeriodSeconds is how often + (in seconds) to perform the probe. + format: int32 + type: integer + timeoutSeconds: + description: + TimeoutSeconds is the number + of seconds after which the probe times + out. + format: int32 + type: integer + type: object + resources: + description: |- + Resources allows customization of limits and requests for compute resources such as cpu and memory. + If specified, this overrides the named OTel Collector StatefulSet container's resources. + If omitted, the OTel Collector StatefulSet will use its default value for this container's resources. + properties: + claims: + description: |- + Claims lists the names of resources, defined in spec.resourceClaims, + that are used by this container. + This field depends on the + DynamicResourceAllocation feature gate. + This field is immutable. It can only be set for containers. + items: + description: + ResourceClaim references + one entry in PodSpec.ResourceClaims. + properties: + name: + description: |- + Name must match the name of one entry in pod.spec.resourceClaims of + the Pod where this field is used. It makes that resource available + inside a container. + type: string + request: + description: |- + Request is the name chosen for a request in the referenced claim. + If empty, everything from the claim is made available, otherwise + only the result of this request. + type: string + required: + - name + type: object + type: array + x-kubernetes-list-map-keys: + - name + x-kubernetes-list-type: map + limits: + additionalProperties: + anyOf: + - type: integer + - type: string + pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$ + x-kubernetes-int-or-string: true + description: |- + Limits describes the maximum amount of compute resources allowed. + More info: https://kubernetes.io/docs/concepts/configuration/manage-resources-containers/ + type: object + requests: + additionalProperties: + anyOf: + - type: integer + - type: string + pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$ + x-kubernetes-int-or-string: true + description: |- + Requests describes the minimum amount of compute resources required. + If Requests is omitted for a container, it defaults to Limits if that is explicitly specified, + otherwise to an implementation-defined value. Requests cannot exceed Limits. + More info: https://kubernetes.io/docs/concepts/configuration/manage-resources-containers/ + type: object + type: object + required: + - name + type: object + type: array + nodeSelector: + additionalProperties: + type: string + description: + NodeSelector gives more control over + the nodes where the OTel Collector pods will + run on. + type: object + priorityClassName: + description: + PriorityClassName allows to specify + a PriorityClass resource to be used. + type: string + tolerations: + description: |- + Tolerations is the OTel Collector pod's tolerations. + If specified, this overrides any tolerations that may be set on the OTel Collector StatefulSet. + If omitted, the OTel Collector StatefulSet will use its default value for tolerations. + items: + description: |- + The pod this Toleration is attached to tolerates any taint that matches + the triple using the matching operator . + properties: + effect: + description: |- + Effect indicates the taint effect to match. Empty means match all taint effects. + When specified, allowed values are NoSchedule, PreferNoSchedule and NoExecute. + type: string + key: + description: |- + Key is the taint key that the toleration applies to. Empty means match all taint keys. + If the key is empty, operator must be Exists; this combination means to match all values and all keys. + type: string + operator: + description: |- + Operator represents a key's relationship to the value. + Valid operators are Exists, Equal, Lt, and Gt. Defaults to Equal. + Exists is equivalent to wildcard for value, so that a pod can + tolerate all taints of a particular category. + Lt and Gt perform numeric comparisons (requires feature gate TaintTolerationComparisonOperators). + type: string + tolerationSeconds: + description: |- + TolerationSeconds represents the period of time the toleration (which must be + of effect NoExecute, otherwise this field is ignored) tolerates the taint. By default, + it is not set, which means tolerate the taint forever (do not evict). Zero and + negative values will be treated as 0 (evict immediately) by the system. + format: int64 + type: integer + value: + description: |- + Value is the taint value the toleration matches to. + If the operator is Exists, the value should be empty, otherwise just a regular string. + type: string + type: object + type: array + topologySpreadConstraints: + description: |- + TopologySpreadConstraints describes how a group of pods ought to spread across topology + domains. Scheduler will schedule pods in a way which abides by the constraints. + All topologySpreadConstraints are ANDed. + items: + description: + TopologySpreadConstraint specifies + how to spread matching pods among the given + topology. + properties: + labelSelector: + description: |- + LabelSelector is used to find matching pods. + Pods that match this label selector are counted to determine the number of pods + in their corresponding topology domain. + 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 + matchLabelKeys: + description: |- + MatchLabelKeys is a set of pod label keys to select the pods over which + spreading will be calculated. The keys are used to lookup values from the + incoming pod labels, those key-value labels are ANDed with labelSelector + to select the group of existing pods over which spreading will be calculated + for the incoming pod. The same key is forbidden to exist in both MatchLabelKeys and LabelSelector. + MatchLabelKeys cannot be set when LabelSelector isn't set. + Keys that don't exist in the incoming pod labels will + be ignored. A null or empty list means only match against labelSelector. + This is a beta field and requires the MatchLabelKeysInPodTopologySpread feature gate to be enabled (enabled by default). + items: + type: string + type: array + x-kubernetes-list-type: atomic + maxSkew: + description: |- + MaxSkew describes the degree to which pods may be unevenly distributed. + When `whenUnsatisfiable=DoNotSchedule`, it is the maximum permitted difference + between the number of matching pods in the target topology and the global minimum. + The global minimum is the minimum number of matching pods in an eligible domain + or zero if the number of eligible domains is less than MinDomains. + For example, in a 3-zone cluster, MaxSkew is set to 1, and pods with the same + labelSelector spread as 2/2/1: + In this case, the global minimum is 1. + | zone1 | zone2 | zone3 | + | P P | P P | P | + - if MaxSkew is 1, incoming pod can only be scheduled to zone3 to become 2/2/2; + scheduling it onto zone1(zone2) would make the ActualSkew(3-1) on zone1(zone2) + violate MaxSkew(1). + - if MaxSkew is 2, incoming pod can be scheduled onto any zone. + When `whenUnsatisfiable=ScheduleAnyway`, it is used to give higher precedence + to topologies that satisfy it. + It's a required field. Default value is 1 and 0 is not allowed. + format: int32 + type: integer + minDomains: + description: |- + MinDomains indicates a minimum number of eligible domains. + When the number of eligible domains with matching topology keys is less than minDomains, + Pod Topology Spread treats "global minimum" as 0, and then the calculation of Skew is performed. + And when the number of eligible domains with matching topology keys equals or greater than minDomains, + this value has no effect on scheduling. + As a result, when the number of eligible domains is less than minDomains, + scheduler won't schedule more than maxSkew Pods to those domains. + If value is nil, the constraint behaves as if MinDomains is equal to 1. + Valid values are integers greater than 0. + When value is not nil, WhenUnsatisfiable must be DoNotSchedule. + For example, in a 3-zone cluster, MaxSkew is set to 2, MinDomains is set to 5 and pods with the same + labelSelector spread as 2/2/2: + | zone1 | zone2 | zone3 | + | P P | P P | P P | + The number of domains is less than 5(MinDomains), so "global minimum" is treated as 0. + In this situation, new pod with the same labelSelector cannot be scheduled, + because computed skew will be 3(3 - 0) if new Pod is scheduled to any of the three zones, + it will violate MaxSkew. + format: int32 + type: integer + nodeAffinityPolicy: + description: |- + NodeAffinityPolicy indicates how we will treat Pod's nodeAffinity/nodeSelector + when calculating pod topology spread skew. Options are: + - Honor: only nodes matching nodeAffinity/nodeSelector are included in the calculations. + - Ignore: nodeAffinity/nodeSelector are ignored. All nodes are included in the calculations. + If this value is nil, the behavior is equivalent to the Honor policy. + type: string + nodeTaintsPolicy: + description: |- + NodeTaintsPolicy indicates how we will treat node taints when calculating + pod topology spread skew. Options are: + - Honor: nodes without taints, along with tainted nodes for which the incoming pod + has a toleration, are included. + - Ignore: node taints are ignored. All nodes are included. + If this value is nil, the behavior is equivalent to the Ignore policy. + type: string + topologyKey: + description: |- + TopologyKey is the key of node labels. Nodes that have a label with this key + and identical values are considered to be in the same topology. + We consider each as a "bucket", and try to put balanced number + of pods into each bucket. + We define a domain as a particular instance of a topology. + Also, we define an eligible domain as a domain whose nodes meet the requirements of + nodeAffinityPolicy and nodeTaintsPolicy. + e.g. If TopologyKey is "kubernetes.io/hostname", each Node is a domain of that topology. + And, if TopologyKey is "topology.kubernetes.io/zone", each zone is a domain of that topology. + It's a required field. + type: string + whenUnsatisfiable: + description: |- + WhenUnsatisfiable indicates how to deal with a pod if it doesn't satisfy + the spread constraint. + - DoNotSchedule (default) tells the scheduler not to schedule it. + - ScheduleAnyway tells the scheduler to schedule the pod in any location, + but giving higher precedence to topologies that would help reduce the + skew. + A constraint is considered "Unsatisfiable" for an incoming pod + if and only if every possible node assignment for that pod would violate + "MaxSkew" on some topology. + For example, in a 3-zone cluster, MaxSkew is set to 1, and pods with the same + labelSelector spread as 3/1/1: + | zone1 | zone2 | zone3 | + | P P P | P | P | + If WhenUnsatisfiable is set to DoNotSchedule, incoming pod can only be scheduled + to zone2(zone3) to become 3/2/1(3/1/2) as ActualSkew(2-1) on zone2(zone3) satisfies + MaxSkew(1). In other words, the cluster can still be imbalanced, but scheduler + won't make it *more* imbalanced. + It's a required field. + type: string + required: + - maxSkew + - topologyKey + - whenUnsatisfiable + type: object + type: array + type: object + type: object + type: object + type: object + type: object type: object status: description: Most recently observed state for Tigera log collection. diff --git a/pkg/render/logcollector/fluentbit_test.go b/pkg/render/logcollector/fluentbit_test.go index bc42622f5d..32a2cc2928 100644 --- a/pkg/render/logcollector/fluentbit_test.go +++ b/pkg/render/logcollector/fluentbit_test.go @@ -240,6 +240,28 @@ var _ = Describe("Tigera Secure Fluent Bit rendering tests", func() { Expect(ms.Spec.ClusterIP).To(Equal("None"), "metrics service should be headless to prevent kube-proxy from rendering too many iptables rules") }) + It("should render one opentelemetry output per selected OTel log type", func() { + cfg.OTelCollectorEnabled = true + cfg.OTelLogTypes = []operatorv1.OTelLogType{operatorv1.OTelFlowLog, operatorv1.OTelAuditLog} + component := logcollector.FluentBitOSSpecific(cfg, rmeta.OSTypeLinux) + resources, _ := component.Objects() + + cm := rtest.GetResource(resources, logcollector.FluentBitConfConfigMapName, render.LogCollectorNamespace, "", "v1", "ConfigMap").(*corev1.ConfigMap) + conf := cm.Data["fluent-bit.yaml"] + + // One output per selected type, matched by tag — DNS not selected, so + // no output for it. service.name is stamped at the source via the + // per-output processors, so the collector needs no classification. + Expect(strings.Count(conf, `"name": "opentelemetry"`)).To(Equal(2), "one output per selected type") + Expect(conf).To(ContainSubstring(`"match": "flows"`)) + Expect(conf).To(ContainSubstring(`"match": "audit.*"`)) + Expect(conf).To(ContainSubstring(`"name": "opentelemetry_envelope"`)) + Expect(conf).To(ContainSubstring(`"name": "content_modifier"`)) + Expect(conf).To(ContainSubstring(`"context": "otel_resource_attributes"`)) + Expect(conf).To(ContainSubstring(`"value": "flows"`)) + Expect(conf).To(ContainSubstring(`"value": "audit"`)) + }) + It("should render fluent-bit DaemonSet with resources requests/limits", func() { ca, _ := tls.MakeCA(rmeta.DefaultOperatorCASignerName()) cert, _, _ := ca.Config.GetPEMBytes() // create a valid pem block diff --git a/pkg/render/logcollector/logcollector.go b/pkg/render/logcollector/logcollector.go index 0dd9064c1e..10343e60af 100644 --- a/pkg/render/logcollector/logcollector.go +++ b/pkg/render/logcollector/logcollector.go @@ -171,6 +171,11 @@ type FluentBitConfiguration struct { // LicenseExpired indicates the license has expired and fluent-bit DaemonSet should be removed. LicenseExpired bool + + OTelCollectorEnabled bool + // OTelLogTypes selects which log types ship to the OTel collector; one + // opentelemetry output is rendered per type. + OTelLogTypes []operatorv1.OTelLogType } type fluentBitComponent struct { diff --git a/pkg/render/logcollector/pipeline.go b/pkg/render/logcollector/pipeline.go index a179213969..8b4b40193a 100644 --- a/pkg/render/logcollector/pipeline.go +++ b/pkg/render/logcollector/pipeline.go @@ -23,6 +23,7 @@ import ( "sigs.k8s.io/yaml" operatorv1 "github.com/tigera/operator/api/v1" + "github.com/tigera/operator/pkg/common" "github.com/tigera/operator/pkg/render" relasticsearch "github.com/tigera/operator/pkg/render/common/elasticsearch" rmeta "github.com/tigera/operator/pkg/render/common/meta" @@ -234,6 +235,8 @@ func (c *fluentBitComponent) addOutputs(cfg *fluentBitConfig) { c.linseedHTTPOutput(tag, c.certPath(), c.keyPath(), linseedStorageLimit(tag))) } + c.addOTelOutputs(cfg) + // Additional stores are Linux-only, matching the fluentd Windows variant // (Linseed only). if c.cfg.LogCollector.Spec.AdditionalStores != nil && c.osType == rmeta.OSTypeLinux { @@ -243,6 +246,50 @@ func (c *fluentBitComponent) addOutputs(cfg *fluentBitConfig) { } } +func (c *fluentBitComponent) addOTelOutputs(cfg *fluentBitConfig) { + if !c.cfg.OTelCollectorEnabled { + return + } + for _, t := range c.cfg.OTelLogTypes { + m, ok := otelLogTypeMatch[t] + if !ok { + continue + } + cfg.Pipeline.Outputs = append(cfg.Pipeline.Outputs, map[string]interface{}{ + "name": "opentelemetry", + "match": m.match, + "host": fmt.Sprintf("otel-collector.%s.svc", common.CalicoNamespace), + "port": 4318, + "logs_uri": "/v1/logs", + "tls": "on", + "tls.verify": "on", + "tls.ca_file": c.trustedBundlePath(), + "tls.crt_file": c.certPath(), + "tls.key_file": c.keyPath(), + "processors": map[string]interface{}{ + "logs": []map[string]interface{}{ + {"name": "opentelemetry_envelope"}, + { + "name": "content_modifier", + "context": "otel_resource_attributes", + "action": "upsert", + "key": "service.name", + "value": m.serviceName, + }, + }, + }, + }) + } +} + +// otelLogTypeMatch maps each OTel log type to {fluent-bit match pattern, service.name}. +// audit.* covers both audit.tsee and audit.kube. +var otelLogTypeMatch = map[operatorv1.OTelLogType]struct{ match, serviceName string }{ + operatorv1.OTelFlowLog: {"flows", "flows"}, + operatorv1.OTelDNSLog: {"dns", "dns"}, + operatorv1.OTelAuditLog: {"audit.*", "audit"}, +} + // linseedTags lists the tags shipped to Linseed: every tailed tag except // ids.events and compliance.reports — those are deliberately not // Linseed-bound (IDS events use a different ingestion path; compliance diff --git a/pkg/render/otelcollector/collector-config.yaml.template b/pkg/render/otelcollector/collector-config.yaml.template new file mode 100644 index 0000000000..1ad3b4a66e --- /dev/null +++ b/pkg/render/otelcollector/collector-config.yaml.template @@ -0,0 +1,85 @@ +receivers: +{{- if .HasLogs}} + otlp: + protocols: + http: + endpoint: 0.0.0.0:4318 +{{- if .ReceiverTLS}} + tls: + cert_file: {{.ReceiverCertFile}} + key_file: {{.ReceiverKeyFile}} + client_ca_file: {{.ReceiverClientCA}} +{{- end}} +{{- end}} +{{- if .MetricsEnabled}} + # Federate everything the in-cluster tigera-prometheus already scrapes + # (calico-node, typha/kube-controllers, calico-api, fluent-bit, + # elasticsearch, operator — every ServiceMonitor, including future ones) + # instead of re-implementing per-component scrape configs and their TLS + # quirks here. The request goes through Prometheus's authn-proxy: bearer + # token is the pod's ServiceAccount token, authorized by the + # services/proxy RBAC rule on the collector's ClusterRole. + prometheus: + config: + scrape_configs: + - job_name: 'tigera-prometheus-federate' + scheme: https + metrics_path: /federate + params: + 'match[]': ['{__name__=~".+"}'] + # Keep the original job/instance labels from the federated series. + honor_labels: true + authorization: + credentials_file: /var/run/secrets/kubernetes.io/serviceaccount/token + tls_config: + ca_file: {{.MetricsCAFile}} + static_configs: + - targets: ['{{.PrometheusFederateTarget}}'] +{{- end}} + +exporters: +{{- range .Exporters}} + {{.Prefix}}/{{.Name}}: + endpoint: {{.Endpoint}} +{{- if .TLSInsecure}} + tls: + insecure: true +{{- end}} +{{- end}} + +processors: + memory_limiter: + check_interval: 1s + limit_mib: {{.MemoryLimitMiB}} + spike_limit_mib: {{.MemorySpikeLimitMiB}} + batch: + send_batch_size: {{.BatchMaxSize}} + send_batch_max_size: {{.BatchMaxSize}} + +extensions: + health_check: + endpoint: 0.0.0.0:{{.HealthCheckPort}} + +service: + telemetry: + metrics: + readers: + - pull: + exporter: + prometheus: + host: "0.0.0.0" + port: {{.InternalMetricsPort}} + extensions: [health_check] + pipelines: +{{- if .HasLogs}} + logs: + receivers: [otlp] + processors: [memory_limiter, batch] + exporters: [{{.ExporterNames}}] +{{- end}} +{{- if .MetricsEnabled}} + metrics: + receivers: [prometheus] + processors: [memory_limiter, batch] + exporters: [{{.ExporterNames}}] +{{- end}} diff --git a/pkg/render/otelcollector/component.go b/pkg/render/otelcollector/component.go new file mode 100644 index 0000000000..c1e14f97e4 --- /dev/null +++ b/pkg/render/otelcollector/component.go @@ -0,0 +1,531 @@ +// Copyright (c) 2026 Tigera, Inc. All rights reserved. + +// 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 otelcollector + +import ( + "bytes" + _ "embed" + "fmt" + "net" + "net/url" + "strconv" + "strings" + "text/template" + + v3 "github.com/tigera/api/pkg/apis/projectcalico/v3" + operatorv1 "github.com/tigera/operator/api/v1" + "github.com/tigera/operator/pkg/common" + "github.com/tigera/operator/pkg/components" + "github.com/tigera/operator/pkg/render" + rcomp "github.com/tigera/operator/pkg/render/common/components" + rmeta "github.com/tigera/operator/pkg/render/common/meta" + "github.com/tigera/operator/pkg/render/common/networkpolicy" + "github.com/tigera/operator/pkg/render/common/secret" + "github.com/tigera/operator/pkg/render/common/securitycontext" + "github.com/tigera/operator/pkg/render/monitor" + "github.com/tigera/operator/pkg/tls/certificatemanagement" + + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + rbacv1 "k8s.io/api/rbac/v1" + "k8s.io/apimachinery/pkg/api/resource" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/intstr" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +const ( + OTelCollectorName = "otel-collector" + OTelCollectorNamespace = common.CalicoNamespace + OTelCollectorServiceAccountName = OTelCollectorName + OTelCollectorStatefulSetName = OTelCollectorName + OTelCollectorServiceName = OTelCollectorName + OTelCollectorConfigMapName = OTelCollectorName + OTelCollectorContainerName = "otel-collector" + OTelCollectorPolicyName = networkpolicy.CalicoComponentPolicyPrefix + OTelCollectorName + OTelCollectorClusterRoleName = OTelCollectorName + OTelCollectorServerTLSSecretName = "otel-collector-tls" + + OTLPGRPCPort = 4317 + OTLPHTTPPort = 4318 + exporterPrefixHTTP = "otlphttp" + exporterPrefixGRPC = "otlp_grpc" + HealthCheckPort = 13133 + InternalMetricsPort = 8888 + + DefaultMemoryLimit = "512Mi" + DefaultMemoryRequest = "128Mi" + DefaultMemoryLimitMiB = 409 // 80% of 512Mi + DefaultMemorySpikeLimitMiB = 100 // ~25% of limit_mib + DefaultBatchMaxSize = 2000 // keep each export under gRPC's 4MB default + + configHashAnnotation = "hash.operator.tigera.io/otel-collector-config" +) + +type Configuration struct { + PullSecrets []*corev1.Secret + OpenShift bool + Installation *operatorv1.InstallationSpec + OTelCollector *operatorv1.OTelCollectorSpec + // ReceiverTLSSecret is the server keypair for the OTLP receiver (mTLS termination). + ReceiverTLSSecret certificatemanagement.KeyPairInterface + TrustedCertBundle certificatemanagement.TrustedBundleRO +} + +type component struct { + cfg *Configuration + image string + renderedConf string +} + +func OTelCollector(cfg *Configuration) (render.Component, error) { + c := &component{cfg: cfg} + conf, err := c.collectorConfig() + if err != nil { + return nil, err + } + c.renderedConf = conf + return c, nil +} + +func (c *component) ResolveImages(is *operatorv1.ImageSet) error { + reg := c.cfg.Installation.Registry + path := c.cfg.Installation.ImagePath + prefix := c.cfg.Installation.ImagePrefix + + var err error + c.image, err = components.GetReference(components.CombinedCalicoImage(c.cfg.Installation), reg, path, prefix, is) + return err +} + +func (c *component) SupportedOSType() rmeta.OSType { + return rmeta.OSTypeLinux +} + +func (c *component) Objects() ([]client.Object, []client.Object) { + statefulSet := c.statefulSet() + if c.cfg.OTelCollector.OTelCollectorStatefulSet != nil { + rcomp.ApplyStatefulSetOverrides(statefulSet, c.cfg.OTelCollector.OTelCollectorStatefulSet) + } + + objs := []client.Object{ + c.serviceAccount(), + c.clusterRole(), + c.clusterRoleBinding(), + c.configMap(), + c.service(), + statefulSet, + c.networkPolicy(), + } + + objs = append(objs, secret.ToRuntimeObjects(secret.CopyToNamespace(OTelCollectorNamespace, c.cfg.PullSecrets...)...)...) + + return objs, nil +} + +func (c *component) Ready() bool { + return true +} + +func (c *component) serviceAccount() *corev1.ServiceAccount { + return &corev1.ServiceAccount{ + TypeMeta: metav1.TypeMeta{Kind: "ServiceAccount", APIVersion: "v1"}, + ObjectMeta: metav1.ObjectMeta{Name: OTelCollectorServiceAccountName, Namespace: OTelCollectorNamespace}, + } +} + +func (c *component) clusterRole() *rbacv1.ClusterRole { + return &rbacv1.ClusterRole{ + TypeMeta: metav1.TypeMeta{Kind: "ClusterRole", APIVersion: "rbac.authorization.k8s.io/v1"}, + ObjectMeta: metav1.ObjectMeta{ + Name: OTelCollectorClusterRoleName, + }, + Rules: []rbacv1.PolicyRule{ + // Authorizes the collector's federate scrapes at the + // tigera-prometheus authn-proxy (TokenReview + + // SubjectAccessReview on this resource) — the same rule the + // manager and guardian use to query Prometheus. + { + APIGroups: []string{""}, + Resources: []string{"services/proxy"}, + ResourceNames: []string{"calico-node-prometheus:9090"}, + Verbs: []string{"get"}, + }, + }, + } +} + +func (c *component) clusterRoleBinding() *rbacv1.ClusterRoleBinding { + return &rbacv1.ClusterRoleBinding{ + TypeMeta: metav1.TypeMeta{Kind: "ClusterRoleBinding", APIVersion: "rbac.authorization.k8s.io/v1"}, + ObjectMeta: metav1.ObjectMeta{ + Name: OTelCollectorClusterRoleName, + }, + RoleRef: rbacv1.RoleRef{ + APIGroup: "rbac.authorization.k8s.io", + Kind: "ClusterRole", + Name: OTelCollectorClusterRoleName, + }, + Subjects: []rbacv1.Subject{ + { + Kind: "ServiceAccount", + Name: OTelCollectorServiceAccountName, + Namespace: OTelCollectorNamespace, + }, + }, + } +} + +func (c *component) configMap() *corev1.ConfigMap { + return &corev1.ConfigMap{ + TypeMeta: metav1.TypeMeta{Kind: "ConfigMap", APIVersion: "v1"}, + ObjectMeta: metav1.ObjectMeta{Name: OTelCollectorConfigMapName, Namespace: OTelCollectorNamespace}, + Data: map[string]string{ + "config.yaml": c.renderedConf, + }, + } +} + +func (c *component) metricsEnabled() bool { + return c.cfg.OTelCollector.MetricsEnabled() +} + +func (c *component) hasLogs() bool { + return c.cfg.OTelCollector.HasLogs() +} + +func (c *component) receiverTLSReady() bool { + return c.hasLogs() && c.cfg.ReceiverTLSSecret != nil && c.cfg.TrustedCertBundle != nil +} + +func (c *component) metricsTLSReady() bool { + return c.metricsEnabled() && c.cfg.TrustedCertBundle != nil +} + +type configTemplateData struct { + HasLogs bool + ReceiverTLS bool + ReceiverCertFile string + ReceiverKeyFile string + ReceiverClientCA string + MetricsEnabled bool + MetricsCAFile string + // PrometheusFederateTarget is the host:port of the tigera-prometheus + // authn-proxy fronting the /federate endpoint. + PrometheusFederateTarget string + Exporters []exporterEntry + ExporterNames string + HealthCheckPort int + InternalMetricsPort int + MemoryLimitMiB int + MemorySpikeLimitMiB int + BatchMaxSize int +} + +type exporterEntry struct { + Prefix string + Name string + Endpoint string + TLSInsecure bool +} + +//go:embed collector-config.yaml.template +var collectorConfigStr string + +var collectorConfigTmpl = template.Must(template.New("config").Parse(collectorConfigStr)) + +func (c *component) collectorConfig() (string, error) { + var exporters []exporterEntry + var exporterNames []string + for _, exp := range c.cfg.OTelCollector.Exporters { + var prefix string + if exp.Protocol == operatorv1.OTelProtocolHTTP { + prefix = exporterPrefixHTTP + } else { + prefix = exporterPrefixGRPC + } + exporters = append(exporters, exporterEntry{ + Prefix: prefix, + Name: exp.Name, + Endpoint: exp.Endpoint, + TLSInsecure: exp.TLSInsecure != nil && *exp.TLSInsecure, + }) + exporterNames = append(exporterNames, fmt.Sprintf("%s/%s", prefix, exp.Name)) + } + + data := configTemplateData{ + HasLogs: c.hasLogs(), + MetricsEnabled: c.metricsEnabled(), + Exporters: exporters, + ExporterNames: strings.Join(exporterNames, ", "), + HealthCheckPort: HealthCheckPort, + InternalMetricsPort: InternalMetricsPort, + MemoryLimitMiB: DefaultMemoryLimitMiB, + MemorySpikeLimitMiB: DefaultMemorySpikeLimitMiB, + BatchMaxSize: DefaultBatchMaxSize, + } + + if c.receiverTLSReady() { + data.ReceiverTLS = true + data.ReceiverCertFile = c.cfg.ReceiverTLSSecret.VolumeMountCertificateFilePath() + data.ReceiverKeyFile = c.cfg.ReceiverTLSSecret.VolumeMountKeyFilePath() + data.ReceiverClientCA = c.cfg.TrustedCertBundle.MountPath() + } + + if c.metricsTLSReady() { + data.MetricsCAFile = c.cfg.TrustedCertBundle.MountPath() + data.PrometheusFederateTarget = fmt.Sprintf("%s.%s.svc:%d", + monitor.PrometheusServiceServiceName, common.TigeraPrometheusNamespace, monitor.PrometheusDefaultPort) + } + + var buf bytes.Buffer + if err := collectorConfigTmpl.Execute(&buf, data); err != nil { + return "", fmt.Errorf("failed to render otel collector config: %w", err) + } + return buf.String(), nil +} + +func (c *component) service() *corev1.Service { + ports := []corev1.ServicePort{ + { + Name: "otlp-http", + Port: OTLPHTTPPort, + TargetPort: intstr.FromInt32(OTLPHTTPPort), + Protocol: corev1.ProtocolTCP, + }, + { + Name: "metrics", + Port: InternalMetricsPort, + TargetPort: intstr.FromInt32(InternalMetricsPort), + Protocol: corev1.ProtocolTCP, + }, + } + + return &corev1.Service{ + TypeMeta: metav1.TypeMeta{Kind: "Service", APIVersion: "v1"}, + ObjectMeta: metav1.ObjectMeta{ + Name: OTelCollectorServiceName, + Namespace: OTelCollectorNamespace, + Labels: map[string]string{"k8s-app": OTelCollectorStatefulSetName}, + }, + Spec: corev1.ServiceSpec{ + Selector: map[string]string{"k8s-app": OTelCollectorStatefulSetName}, + Ports: ports, + }, + } +} + +func (c *component) container() corev1.Container { + volumeMounts := []corev1.VolumeMount{ + { + Name: "config", + MountPath: "/etc/otel", + ReadOnly: true, + }, + } + + if c.cfg.TrustedCertBundle != nil { + volumeMounts = append(volumeMounts, + c.cfg.TrustedCertBundle.VolumeMounts(rmeta.OSTypeLinux)..., + ) + } + + if c.cfg.ReceiverTLSSecret != nil { + volumeMounts = append(volumeMounts, + c.cfg.ReceiverTLSSecret.VolumeMount(rmeta.OSTypeLinux), + ) + } + + return corev1.Container{ + Name: OTelCollectorContainerName, + Image: c.image, + Command: []string{"/usr/bin/otelcol", "--config=/etc/otel/config.yaml"}, + Ports: []corev1.ContainerPort{ + // No otlp-grpc port: the receiver is HTTP-only (fluent-bit's + // opentelemetry output has no gRPC mode). 4317 appears only in + // the egress policy for outbound gRPC exporters. + {Name: "otlp-http", ContainerPort: OTLPHTTPPort, Protocol: corev1.ProtocolTCP}, + {Name: "health", ContainerPort: HealthCheckPort, Protocol: corev1.ProtocolTCP}, + {Name: "metrics", ContainerPort: InternalMetricsPort, Protocol: corev1.ProtocolTCP}, + }, + Resources: corev1.ResourceRequirements{ + Limits: corev1.ResourceList{ + corev1.ResourceMemory: resource.MustParse(DefaultMemoryLimit), + }, + Requests: corev1.ResourceList{ + corev1.ResourceMemory: resource.MustParse(DefaultMemoryRequest), + }, + }, + SecurityContext: securitycontext.NewNonRootContext(), + ReadinessProbe: healthProbe(), + LivenessProbe: healthProbe(), + VolumeMounts: volumeMounts, + } +} + +func healthProbe() *corev1.Probe { + return &corev1.Probe{ + ProbeHandler: corev1.ProbeHandler{ + HTTPGet: &corev1.HTTPGetAction{ + Path: "/", + Port: intstr.FromInt32(HealthCheckPort), + }, + }, + PeriodSeconds: 10, + } +} + +func (c *component) statefulSet() *appsv1.StatefulSet { + tolerations := append(c.cfg.Installation.ControlPlaneTolerations, rmeta.TolerateCriticalAddonsAndControlPlane...) + if c.cfg.Installation.KubernetesProvider.IsGKE() { + tolerations = append(tolerations, rmeta.TolerateGKEARM64NoSchedule) + } + + volumes := []corev1.Volume{ + { + Name: "config", + VolumeSource: corev1.VolumeSource{ + ConfigMap: &corev1.ConfigMapVolumeSource{ + LocalObjectReference: corev1.LocalObjectReference{Name: OTelCollectorConfigMapName}, + }, + }, + }, + } + + if c.cfg.TrustedCertBundle != nil { + volumes = append(volumes, c.cfg.TrustedCertBundle.Volume()) + } + + if c.cfg.ReceiverTLSSecret != nil { + volumes = append(volumes, c.cfg.ReceiverTLSSecret.Volume()) + } + + return &appsv1.StatefulSet{ + TypeMeta: metav1.TypeMeta{Kind: "StatefulSet", APIVersion: "apps/v1"}, + ObjectMeta: metav1.ObjectMeta{ + Name: OTelCollectorStatefulSetName, + Namespace: OTelCollectorNamespace, + }, + Spec: appsv1.StatefulSetSpec{ + Replicas: c.cfg.Installation.ControlPlaneReplicas, + ServiceName: OTelCollectorServiceName, + Selector: &metav1.LabelSelector{ + MatchLabels: map[string]string{"k8s-app": OTelCollectorStatefulSetName}, + }, + Template: corev1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{ + Name: OTelCollectorStatefulSetName, + Labels: map[string]string{"k8s-app": OTelCollectorStatefulSetName}, + Annotations: map[string]string{configHashAnnotation: rmeta.AnnotationHash(c.renderedConf)}, + }, + Spec: corev1.PodSpec{ + NodeSelector: c.cfg.Installation.ControlPlaneNodeSelector, + ServiceAccountName: OTelCollectorServiceAccountName, + Tolerations: tolerations, + ImagePullSecrets: secret.GetReferenceList(c.cfg.PullSecrets), + Containers: []corev1.Container{c.container()}, + Volumes: volumes, + }, + }, + }, + } +} + +func (c *component) networkPolicy() *v3.NetworkPolicy { + egressRules := []v3.Rule{ + { + Action: v3.Allow, + Protocol: &networkpolicy.TCPProtocol, + Destination: v3.EntityRule{ + Ports: networkpolicy.Ports(OTLPGRPCPort, OTLPHTTPPort), + }, + }, + } + + for _, exp := range c.cfg.OTelCollector.Exporters { + if port, ok := resolvePort(exp.Endpoint); ok { + egressRules = append(egressRules, v3.Rule{ + Action: v3.Allow, + Protocol: &networkpolicy.TCPProtocol, + Destination: v3.EntityRule{ + Ports: networkpolicy.Ports(uint16(port)), + }, + }) + } + } + + if c.metricsEnabled() { + // Federation scrapes go to the tigera-prometheus authn-proxy only. + egressRules = append(egressRules, v3.Rule{ + Action: v3.Allow, + Protocol: &networkpolicy.TCPProtocol, + Destination: networkpolicy.PrometheusEntityRule, + }) + } + + egressRules = networkpolicy.AppendDNSEgressRules(egressRules, c.cfg.OpenShift) + + return &v3.NetworkPolicy{ + TypeMeta: metav1.TypeMeta{Kind: "NetworkPolicy", APIVersion: "projectcalico.org/v3"}, + ObjectMeta: metav1.ObjectMeta{Name: OTelCollectorPolicyName, Namespace: OTelCollectorNamespace}, + Spec: v3.NetworkPolicySpec{ + Tier: networkpolicy.CalicoTierName, + Selector: networkpolicy.KubernetesAppSelector(OTelCollectorStatefulSetName), + Types: []v3.PolicyType{v3.PolicyTypeIngress, v3.PolicyTypeEgress}, + Ingress: []v3.Rule{ + { + Action: v3.Allow, + Protocol: &networkpolicy.TCPProtocol, + Destination: v3.EntityRule{ + Ports: networkpolicy.Ports(OTLPHTTPPort), + }, + }, + { + Action: v3.Allow, + Protocol: &networkpolicy.TCPProtocol, + Destination: v3.EntityRule{ + Ports: networkpolicy.Ports(InternalMetricsPort), + }, + }, + }, + Egress: egressRules, + }, + } +} + +// resolvePort extracts a TCP port from an endpoint string. Accepts bare +// host:port ("otlp.example.com:4317") or full URLs ("https://otlp.example.com:443"), +// falling back to the scheme's default port (443 for https, 80 for http). +func resolvePort(endpoint string) (int, bool) { + if _, portStr, err := net.SplitHostPort(endpoint); err == nil { + if p, err := strconv.Atoi(portStr); err == nil && p > 0 && p <= 65535 { + return p, true + } + } + if u, err := url.Parse(endpoint); err == nil { + if portStr := u.Port(); portStr != "" { + if p, err := strconv.Atoi(portStr); err == nil && p > 0 && p <= 65535 { + return p, true + } + } + switch u.Scheme { + case "https": + return 443, true + case "http": + return 80, true + } + } + return 0, false +} diff --git a/pkg/render/otelcollector/component_test.go b/pkg/render/otelcollector/component_test.go new file mode 100644 index 0000000000..78acfa6ac9 --- /dev/null +++ b/pkg/render/otelcollector/component_test.go @@ -0,0 +1,708 @@ +// Copyright (c) 2026 Tigera, Inc. All rights reserved. + +// 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 otelcollector_test + +import ( + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + "github.com/google/go-cmp/cmp" + + v3 "github.com/tigera/api/pkg/apis/projectcalico/v3" + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + rbacv1 "k8s.io/api/rbac/v1" + "k8s.io/apimachinery/pkg/api/resource" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/intstr" + "k8s.io/utils/ptr" + + operatorv1 "github.com/tigera/operator/api/v1" + rmeta "github.com/tigera/operator/pkg/render/common/meta" + "github.com/tigera/operator/pkg/render/common/securitycontext" + rtest "github.com/tigera/operator/pkg/render/common/test" + "github.com/tigera/operator/pkg/render/otelcollector" + "github.com/tigera/operator/pkg/tls/certificatemanagement" +) + +var _ = Describe("OTelCollector rendering", func() { + var defaultInstallation *operatorv1.InstallationSpec + + BeforeEach(func() { + defaultInstallation = &operatorv1.InstallationSpec{ + Variant: operatorv1.CalicoEnterprise, + Registry: "testregistry.com/", + KubernetesProvider: operatorv1.ProviderGKE, + ControlPlaneReplicas: ptr.To(int32(2)), + } + }) + + DescribeTable("Object counts", + func(cfg *otelcollector.Configuration, createCount, deleteCount int) { + component, err := otelcollector.OTelCollector(cfg) + Expect(err).NotTo(HaveOccurred()) + toCreate, toDelete := component.Objects() + Expect(toCreate).To(HaveLen(createCount)) + Expect(toDelete).To(HaveLen(deleteCount)) + }, + Entry("logs and metrics enabled", + &otelcollector.Configuration{ + Installation: &operatorv1.InstallationSpec{KubernetesProvider: operatorv1.ProviderGKE}, + OTelCollector: &operatorv1.OTelCollectorSpec{ + Logs: &operatorv1.OTelLogs{Types: []operatorv1.OTelLogType{operatorv1.OTelFlowLog}}, + Metrics: &operatorv1.OTelMetrics{Enabled: ptr.To(operatorv1.OTelMetricsEnable)}, + Exporters: []operatorv1.OTelExporter{{Name: "backend", Endpoint: "otlp.example.com:4317"}}, + }, + }, + 7, 0, + ), + Entry("logs only", + &otelcollector.Configuration{ + Installation: &operatorv1.InstallationSpec{KubernetesProvider: operatorv1.ProviderGKE}, + OTelCollector: &operatorv1.OTelCollectorSpec{ + Logs: &operatorv1.OTelLogs{Types: []operatorv1.OTelLogType{operatorv1.OTelAuditLog}}, + Exporters: []operatorv1.OTelExporter{{Name: "backend", Endpoint: "otlp.example.com:4317"}}, + }, + }, + 7, 0, + ), + Entry("metrics only", + &otelcollector.Configuration{ + Installation: &operatorv1.InstallationSpec{KubernetesProvider: operatorv1.ProviderGKE}, + OTelCollector: &operatorv1.OTelCollectorSpec{ + Metrics: &operatorv1.OTelMetrics{Enabled: ptr.To(operatorv1.OTelMetricsEnable)}, + Exporters: []operatorv1.OTelExporter{{Name: "backend", Endpoint: "otlp.example.com:4317"}}, + }, + }, + 7, 0, + ), + Entry("no logs, no metrics", + &otelcollector.Configuration{ + Installation: &operatorv1.InstallationSpec{KubernetesProvider: operatorv1.ProviderGKE}, + OTelCollector: &operatorv1.OTelCollectorSpec{ + Exporters: []operatorv1.OTelExporter{{Name: "backend", Endpoint: "otlp.example.com:4317"}}, + }, + }, + 7, 0, + ), + ) + + Context("StatefulSet rendering", func() { + It("should render the expected statefulset", func() { + cfg := &otelcollector.Configuration{ + Installation: defaultInstallation, + OTelCollector: &operatorv1.OTelCollectorSpec{ + Logs: &operatorv1.OTelLogs{Types: []operatorv1.OTelLogType{operatorv1.OTelFlowLog}}, + Exporters: []operatorv1.OTelExporter{{Name: "backend", Endpoint: "otlp.example.com:4317"}}, + }, + } + component, err := otelcollector.OTelCollector(cfg) + Expect(err).NotTo(HaveOccurred()) + Expect(component.ResolveImages(nil)).NotTo(HaveOccurred()) + objs, _ := component.Objects() + + expected := &appsv1.StatefulSet{ + TypeMeta: metav1.TypeMeta{Kind: "StatefulSet", APIVersion: "apps/v1"}, + ObjectMeta: metav1.ObjectMeta{ + Name: otelcollector.OTelCollectorStatefulSetName, + Namespace: otelcollector.OTelCollectorNamespace, + }, + Spec: appsv1.StatefulSetSpec{ + Replicas: ptr.To(int32(2)), + ServiceName: otelcollector.OTelCollectorServiceName, + Selector: &metav1.LabelSelector{ + MatchLabels: map[string]string{"k8s-app": otelcollector.OTelCollectorStatefulSetName}, + }, + Template: corev1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{ + Name: otelcollector.OTelCollectorStatefulSetName, + Labels: map[string]string{"k8s-app": otelcollector.OTelCollectorStatefulSetName}, + }, + Spec: corev1.PodSpec{ + ServiceAccountName: otelcollector.OTelCollectorServiceAccountName, + Tolerations: append(rmeta.TolerateCriticalAddonsAndControlPlane, rmeta.TolerateGKEARM64NoSchedule), + Containers: []corev1.Container{ + { + Name: otelcollector.OTelCollectorContainerName, + Image: "testregistry.com/tigera/calico:master", + Command: []string{"/usr/bin/otelcol", "--config=/etc/otel/config.yaml"}, + Ports: []corev1.ContainerPort{ + {Name: "otlp-http", ContainerPort: otelcollector.OTLPHTTPPort, Protocol: corev1.ProtocolTCP}, + {Name: "health", ContainerPort: otelcollector.HealthCheckPort, Protocol: corev1.ProtocolTCP}, + {Name: "metrics", ContainerPort: otelcollector.InternalMetricsPort, Protocol: corev1.ProtocolTCP}, + }, + Resources: corev1.ResourceRequirements{ + Limits: corev1.ResourceList{ + corev1.ResourceMemory: resource.MustParse(otelcollector.DefaultMemoryLimit), + }, + Requests: corev1.ResourceList{ + corev1.ResourceMemory: resource.MustParse(otelcollector.DefaultMemoryRequest), + }, + }, + SecurityContext: securitycontext.NewNonRootContext(), + ReadinessProbe: &corev1.Probe{ + ProbeHandler: corev1.ProbeHandler{ + HTTPGet: &corev1.HTTPGetAction{ + Path: "/", + Port: intstr.FromInt32(otelcollector.HealthCheckPort), + }, + }, + PeriodSeconds: 10, + }, + LivenessProbe: &corev1.Probe{ + ProbeHandler: corev1.ProbeHandler{ + HTTPGet: &corev1.HTTPGetAction{ + Path: "/", + Port: intstr.FromInt32(otelcollector.HealthCheckPort), + }, + }, + PeriodSeconds: 10, + }, + VolumeMounts: []corev1.VolumeMount{ + {Name: "config", MountPath: "/etc/otel", ReadOnly: true}, + }, + }, + }, + Volumes: []corev1.Volume{ + { + Name: "config", + VolumeSource: corev1.VolumeSource{ + ConfigMap: &corev1.ConfigMapVolumeSource{ + LocalObjectReference: corev1.LocalObjectReference{Name: otelcollector.OTelCollectorConfigMapName}, + }, + }, + }, + }, + }, + }, + }, + } + + statefulSet, err := rtest.GetResourceOfType[*appsv1.StatefulSet](objs, otelcollector.OTelCollectorStatefulSetName, otelcollector.OTelCollectorNamespace) + Expect(err).ShouldNot(HaveOccurred()) + + Expect(statefulSet.Spec.Template.Spec.Containers[0].Ports).To(ConsistOf(expected.Spec.Template.Spec.Containers[0].Ports)) + Expect(statefulSet.Spec.Template.Spec.Containers[0].VolumeMounts).To(ConsistOf(expected.Spec.Template.Spec.Containers[0].VolumeMounts)) + Expect(statefulSet.Spec.Template.Spec.Volumes).To(ConsistOf(expected.Spec.Template.Spec.Volumes)) + Expect(statefulSet.Spec.Template.ObjectMeta.Annotations).To(HaveKey("hash.operator.tigera.io/otel-collector-config")) + expected.Spec.Template.ObjectMeta.Annotations = statefulSet.Spec.Template.ObjectMeta.Annotations + Expect(statefulSet).To(Equal(expected), cmp.Diff(statefulSet, expected)) + }) + + It("should include the trusted bundle volume when metrics are enabled", func() { + trustedBundle := certificatemanagement.CreateTrustedBundle(nil) + + cfg := &otelcollector.Configuration{ + Installation: defaultInstallation, + OTelCollector: &operatorv1.OTelCollectorSpec{ + Metrics: &operatorv1.OTelMetrics{Enabled: ptr.To(operatorv1.OTelMetricsEnable)}, + Exporters: []operatorv1.OTelExporter{{Name: "backend", Endpoint: "otlp.example.com:4317"}}, + }, + TrustedCertBundle: trustedBundle, + } + component, err := otelcollector.OTelCollector(cfg) + Expect(err).NotTo(HaveOccurred()) + Expect(component.ResolveImages(nil)).NotTo(HaveOccurred()) + objs, _ := component.Objects() + + statefulSet, err := rtest.GetResourceOfType[*appsv1.StatefulSet](objs, otelcollector.OTelCollectorStatefulSetName, otelcollector.OTelCollectorNamespace) + Expect(err).ShouldNot(HaveOccurred()) + + Expect(len(statefulSet.Spec.Template.Spec.Volumes)).To(BeNumerically(">", 1)) + Expect(len(statefulSet.Spec.Template.Spec.Containers[0].VolumeMounts)).To(BeNumerically(">", 1)) + }) + + It("should include receiver TLS volumes and mounts when logs with certs are enabled", func() { + receiverKeyPair := certificatemanagement.NewKeyPair(&corev1.Secret{ObjectMeta: metav1.ObjectMeta{Name: "otel-collector-tls"}}, nil, "") + trustedBundle := certificatemanagement.CreateTrustedBundle(nil) + + cfg := &otelcollector.Configuration{ + Installation: defaultInstallation, + OTelCollector: &operatorv1.OTelCollectorSpec{ + Logs: &operatorv1.OTelLogs{Types: []operatorv1.OTelLogType{operatorv1.OTelFlowLog}}, + Exporters: []operatorv1.OTelExporter{{Name: "backend", Endpoint: "otlp.example.com:4317"}}, + }, + ReceiverTLSSecret: receiverKeyPair, + TrustedCertBundle: trustedBundle, + } + component, err := otelcollector.OTelCollector(cfg) + Expect(err).NotTo(HaveOccurred()) + Expect(component.ResolveImages(nil)).NotTo(HaveOccurred()) + objs, _ := component.Objects() + + statefulSet, err := rtest.GetResourceOfType[*appsv1.StatefulSet](objs, otelcollector.OTelCollectorStatefulSetName, otelcollector.OTelCollectorNamespace) + Expect(err).ShouldNot(HaveOccurred()) + + Expect(len(statefulSet.Spec.Template.Spec.Volumes)).To(BeNumerically(">", 1)) + Expect(len(statefulSet.Spec.Template.Spec.Containers[0].VolumeMounts)).To(BeNumerically(">", 1)) + }) + }) + + Context("ConfigMap content", func() { + It("should include otlp receiver when logs are enabled", func() { + cfg := &otelcollector.Configuration{ + Installation: defaultInstallation, + OTelCollector: &operatorv1.OTelCollectorSpec{ + Logs: &operatorv1.OTelLogs{Types: []operatorv1.OTelLogType{operatorv1.OTelFlowLog}}, + Exporters: []operatorv1.OTelExporter{{Name: "backend", Endpoint: "otlp.example.com:4317"}}, + }, + } + comp, err := otelcollector.OTelCollector(cfg) + Expect(err).NotTo(HaveOccurred()) + objs, _ := comp.Objects() + cm, err := rtest.GetResourceOfType[*corev1.ConfigMap](objs, otelcollector.OTelCollectorConfigMapName, otelcollector.OTelCollectorNamespace) + Expect(err).ShouldNot(HaveOccurred()) + + config := cm.Data["config.yaml"] + Expect(config).To(ContainSubstring("otlp:")) + Expect(config).To(ContainSubstring("0.0.0.0:4318")) + Expect(config).To(ContainSubstring("logs:")) + Expect(config).To(ContainSubstring("receivers: [otlp]")) + }) + + It("should include receiver TLS config when logs with certs are enabled", func() { + receiverKeyPair := certificatemanagement.NewKeyPair(&corev1.Secret{ObjectMeta: metav1.ObjectMeta{Name: "otel-collector-tls"}}, nil, "") + trustedBundle := certificatemanagement.CreateTrustedBundle(nil) + + cfg := &otelcollector.Configuration{ + Installation: defaultInstallation, + OTelCollector: &operatorv1.OTelCollectorSpec{ + Logs: &operatorv1.OTelLogs{Types: []operatorv1.OTelLogType{operatorv1.OTelFlowLog}}, + Exporters: []operatorv1.OTelExporter{{Name: "backend", Endpoint: "otlp.example.com:4317"}}, + }, + ReceiverTLSSecret: receiverKeyPair, + TrustedCertBundle: trustedBundle, + } + comp, err := otelcollector.OTelCollector(cfg) + Expect(err).NotTo(HaveOccurred()) + objs, _ := comp.Objects() + cm, err := rtest.GetResourceOfType[*corev1.ConfigMap](objs, otelcollector.OTelCollectorConfigMapName, otelcollector.OTelCollectorNamespace) + Expect(err).ShouldNot(HaveOccurred()) + + config := cm.Data["config.yaml"] + Expect(config).To(ContainSubstring("otlp:")) + Expect(config).To(ContainSubstring("cert_file:")) + Expect(config).To(ContainSubstring("key_file:")) + Expect(config).To(ContainSubstring("client_ca_file:")) + }) + + It("should include prometheus receiver when metrics are enabled", func() { + trustedBundle := certificatemanagement.CreateTrustedBundle(nil) + + cfg := &otelcollector.Configuration{ + Installation: defaultInstallation, + OTelCollector: &operatorv1.OTelCollectorSpec{ + Metrics: &operatorv1.OTelMetrics{Enabled: ptr.To(operatorv1.OTelMetricsEnable)}, + Exporters: []operatorv1.OTelExporter{{Name: "backend", Endpoint: "otlp.example.com:4317"}}, + }, + TrustedCertBundle: trustedBundle, + } + comp, err := otelcollector.OTelCollector(cfg) + Expect(err).NotTo(HaveOccurred()) + objs, _ := comp.Objects() + cm, err := rtest.GetResourceOfType[*corev1.ConfigMap](objs, otelcollector.OTelCollectorConfigMapName, otelcollector.OTelCollectorNamespace) + Expect(err).ShouldNot(HaveOccurred()) + + config := cm.Data["config.yaml"] + Expect(config).To(ContainSubstring("prometheus:")) + Expect(config).To(ContainSubstring("job_name: 'tigera-prometheus-federate'")) + Expect(config).To(ContainSubstring("metrics_path: /federate")) + Expect(config).To(ContainSubstring("honor_labels: true")) + Expect(config).To(ContainSubstring("credentials_file: /var/run/secrets/kubernetes.io/serviceaccount/token")) + Expect(config).To(ContainSubstring("tls_config:")) + Expect(config).To(ContainSubstring("targets: ['prometheus-http-api.tigera-prometheus.svc:9090']")) + Expect(config).To(ContainSubstring("metrics:")) + Expect(config).To(ContainSubstring("receivers: [prometheus]")) + Expect(config).To(ContainSubstring("exporters: [otlp_grpc/backend]")) + }) + + It("should not include receivers or pipelines when logs and metrics are disabled", func() { + cfg := &otelcollector.Configuration{ + Installation: defaultInstallation, + OTelCollector: &operatorv1.OTelCollectorSpec{ + Exporters: []operatorv1.OTelExporter{{Name: "backend", Endpoint: "otlp.example.com:4317"}}, + }, + } + comp, err := otelcollector.OTelCollector(cfg) + Expect(err).NotTo(HaveOccurred()) + objs, _ := comp.Objects() + cm, err := rtest.GetResourceOfType[*corev1.ConfigMap](objs, otelcollector.OTelCollectorConfigMapName, otelcollector.OTelCollectorNamespace) + Expect(err).ShouldNot(HaveOccurred()) + + config := cm.Data["config.yaml"] + Expect(config).NotTo(ContainSubstring("otlp:")) + Expect(config).NotTo(ContainSubstring("scrape_configs:")) + Expect(config).NotTo(ContainSubstring("logs:")) + Expect(config).NotTo(ContainSubstring("receivers: [prometheus]")) + }) + + It("should use otlphttp prefix for HTTP exporters", func() { + cfg := &otelcollector.Configuration{ + Installation: defaultInstallation, + OTelCollector: &operatorv1.OTelCollectorSpec{ + Logs: &operatorv1.OTelLogs{Types: []operatorv1.OTelLogType{operatorv1.OTelFlowLog}}, + Exporters: []operatorv1.OTelExporter{ + {Name: "httpbackend", Endpoint: "https://otlp.example.com:443", Protocol: operatorv1.OTelProtocolHTTP}, + }, + }, + } + comp, err := otelcollector.OTelCollector(cfg) + Expect(err).NotTo(HaveOccurred()) + objs, _ := comp.Objects() + cm, err := rtest.GetResourceOfType[*corev1.ConfigMap](objs, otelcollector.OTelCollectorConfigMapName, otelcollector.OTelCollectorNamespace) + Expect(err).ShouldNot(HaveOccurred()) + + config := cm.Data["config.yaml"] + Expect(config).To(ContainSubstring("otlphttp/httpbackend:")) + Expect(config).To(ContainSubstring("exporters: [otlphttp/httpbackend]")) + }) + + It("should use otlp_grpc prefix for gRPC exporters", func() { + cfg := &otelcollector.Configuration{ + Installation: defaultInstallation, + OTelCollector: &operatorv1.OTelCollectorSpec{ + Logs: &operatorv1.OTelLogs{Types: []operatorv1.OTelLogType{operatorv1.OTelFlowLog}}, + Exporters: []operatorv1.OTelExporter{ + {Name: "grpcbackend", Endpoint: "otlp.example.com:4317", Protocol: operatorv1.OTelProtocolGRPC}, + }, + }, + } + comp, err := otelcollector.OTelCollector(cfg) + Expect(err).NotTo(HaveOccurred()) + objs, _ := comp.Objects() + cm, err := rtest.GetResourceOfType[*corev1.ConfigMap](objs, otelcollector.OTelCollectorConfigMapName, otelcollector.OTelCollectorNamespace) + Expect(err).ShouldNot(HaveOccurred()) + + config := cm.Data["config.yaml"] + Expect(config).To(ContainSubstring("otlp_grpc/grpcbackend:")) + Expect(config).To(ContainSubstring("exporters: [otlp_grpc/grpcbackend]")) + }) + + It("should list multiple exporters in pipelines", func() { + cfg := &otelcollector.Configuration{ + Installation: defaultInstallation, + OTelCollector: &operatorv1.OTelCollectorSpec{ + Logs: &operatorv1.OTelLogs{Types: []operatorv1.OTelLogType{operatorv1.OTelFlowLog}}, + Exporters: []operatorv1.OTelExporter{ + {Name: "first", Endpoint: "first.example.com:4317"}, + {Name: "second", Endpoint: "https://second.example.com", Protocol: operatorv1.OTelProtocolHTTP}, + }, + }, + } + comp, err := otelcollector.OTelCollector(cfg) + Expect(err).NotTo(HaveOccurred()) + objs, _ := comp.Objects() + cm, err := rtest.GetResourceOfType[*corev1.ConfigMap](objs, otelcollector.OTelCollectorConfigMapName, otelcollector.OTelCollectorNamespace) + Expect(err).ShouldNot(HaveOccurred()) + + config := cm.Data["config.yaml"] + Expect(config).To(ContainSubstring("exporters: [otlp_grpc/first, otlphttp/second]")) + }) + + It("should always include the health_check extension", func() { + cfg := &otelcollector.Configuration{ + Installation: defaultInstallation, + OTelCollector: &operatorv1.OTelCollectorSpec{ + Exporters: []operatorv1.OTelExporter{{Name: "backend", Endpoint: "otlp.example.com:4317"}}, + }, + } + comp, err := otelcollector.OTelCollector(cfg) + Expect(err).NotTo(HaveOccurred()) + objs, _ := comp.Objects() + cm, err := rtest.GetResourceOfType[*corev1.ConfigMap](objs, otelcollector.OTelCollectorConfigMapName, otelcollector.OTelCollectorNamespace) + Expect(err).ShouldNot(HaveOccurred()) + + config := cm.Data["config.yaml"] + Expect(config).To(ContainSubstring("extensions:")) + Expect(config).To(ContainSubstring("health_check:")) + Expect(config).To(ContainSubstring("0.0.0.0:13133")) + Expect(config).To(ContainSubstring("extensions: [health_check]")) + }) + }) + + Context("Service", func() { + It("should expose the OTLP HTTP port", func() { + cfg := &otelcollector.Configuration{ + Installation: defaultInstallation, + OTelCollector: &operatorv1.OTelCollectorSpec{ + Exporters: []operatorv1.OTelExporter{{Name: "backend", Endpoint: "otlp.example.com:4317"}}, + }, + } + comp, err := otelcollector.OTelCollector(cfg) + Expect(err).NotTo(HaveOccurred()) + objs, _ := comp.Objects() + svc, err := rtest.GetResourceOfType[*corev1.Service](objs, otelcollector.OTelCollectorServiceName, otelcollector.OTelCollectorNamespace) + Expect(err).ShouldNot(HaveOccurred()) + Expect(svc.Spec.Ports).To(HaveLen(2)) + Expect(svc.Spec.Ports[0].Port).To(Equal(int32(otelcollector.OTLPHTTPPort))) + Expect(svc.Spec.Ports[0].Name).To(Equal("otlp-http")) + Expect(svc.Spec.Ports[1].Port).To(Equal(int32(otelcollector.InternalMetricsPort))) + Expect(svc.Spec.Ports[1].Name).To(Equal("metrics")) + Expect(svc.Spec.Selector).To(Equal(map[string]string{"k8s-app": otelcollector.OTelCollectorStatefulSetName})) + }) + }) + + Context("RBAC", func() { + It("should render the expected service account, cluster role, and cluster role binding", func() { + cfg := &otelcollector.Configuration{ + Installation: defaultInstallation, + OTelCollector: &operatorv1.OTelCollectorSpec{ + Exporters: []operatorv1.OTelExporter{{Name: "backend", Endpoint: "otlp.example.com:4317"}}, + }, + } + comp, err := otelcollector.OTelCollector(cfg) + Expect(err).NotTo(HaveOccurred()) + objs, _ := comp.Objects() + + sa, err := rtest.GetResourceOfType[*corev1.ServiceAccount](objs, otelcollector.OTelCollectorServiceAccountName, otelcollector.OTelCollectorNamespace) + Expect(err).ShouldNot(HaveOccurred()) + Expect(sa).NotTo(BeNil()) + + cr, err := rtest.GetResourceOfType[*rbacv1.ClusterRole](objs, otelcollector.OTelCollectorClusterRoleName, "") + Expect(err).ShouldNot(HaveOccurred()) + Expect(cr.Rules).To(HaveLen(1)) + Expect(cr.Rules[0].Resources).To(ConsistOf("services/proxy")) + Expect(cr.Rules[0].ResourceNames).To(ConsistOf("calico-node-prometheus:9090")) + + crb, err := rtest.GetResourceOfType[*rbacv1.ClusterRoleBinding](objs, otelcollector.OTelCollectorClusterRoleName, "") + Expect(err).ShouldNot(HaveOccurred()) + Expect(crb.RoleRef.Name).To(Equal(otelcollector.OTelCollectorClusterRoleName)) + Expect(crb.Subjects).To(HaveLen(1)) + Expect(crb.Subjects[0].Name).To(Equal(otelcollector.OTelCollectorServiceAccountName)) + }) + }) + + Context("NetworkPolicy", func() { + It("should allow ingress on the OTLP HTTP port", func() { + cfg := &otelcollector.Configuration{ + Installation: defaultInstallation, + OTelCollector: &operatorv1.OTelCollectorSpec{ + Exporters: []operatorv1.OTelExporter{{Name: "backend", Endpoint: "otlp.example.com:4317"}}, + }, + } + comp, err := otelcollector.OTelCollector(cfg) + Expect(err).NotTo(HaveOccurred()) + objs, _ := comp.Objects() + + np, err := rtest.GetResourceOfType[*v3.NetworkPolicy](objs, otelcollector.OTelCollectorPolicyName, otelcollector.OTelCollectorNamespace) + Expect(err).ShouldNot(HaveOccurred()) + Expect(np.Spec.Ingress).To(HaveLen(2)) + Expect(np.Spec.Ingress[0].Action).To(Equal(v3.Allow)) + Expect(np.Spec.Ingress[1].Action).To(Equal(v3.Allow)) + Expect(np.Spec.Types).To(ConsistOf(v3.PolicyTypeIngress, v3.PolicyTypeEgress)) + }) + + It("should parse the port from bare host:port exporter endpoints", func() { + cfg := &otelcollector.Configuration{ + Installation: defaultInstallation, + OTelCollector: &operatorv1.OTelCollectorSpec{ + Exporters: []operatorv1.OTelExporter{{Name: "backend", Endpoint: "otlp.example.com:4317"}}, + }, + } + comp, err := otelcollector.OTelCollector(cfg) + Expect(err).NotTo(HaveOccurred()) + objs, _ := comp.Objects() + + np, err := rtest.GetResourceOfType[*v3.NetworkPolicy](objs, otelcollector.OTelCollectorPolicyName, otelcollector.OTelCollectorNamespace) + Expect(err).ShouldNot(HaveOccurred()) + + var egressPorts []uint16 + for _, r := range np.Spec.Egress { + for _, p := range r.Destination.Ports { + egressPorts = append(egressPorts, p.MinPort) + } + } + Expect(egressPorts).To(ContainElement(uint16(4317)), "should include port from bare host:port endpoint") + }) + + It("should parse the port from URL-style exporter endpoints", func() { + cfg := &otelcollector.Configuration{ + Installation: defaultInstallation, + OTelCollector: &operatorv1.OTelCollectorSpec{ + Exporters: []operatorv1.OTelExporter{ + {Name: "https-backend", Endpoint: "https://otlp.example.com:9443", Protocol: operatorv1.OTelProtocolHTTP}, + }, + }, + } + comp, err := otelcollector.OTelCollector(cfg) + Expect(err).NotTo(HaveOccurred()) + objs, _ := comp.Objects() + + np, err := rtest.GetResourceOfType[*v3.NetworkPolicy](objs, otelcollector.OTelCollectorPolicyName, otelcollector.OTelCollectorNamespace) + Expect(err).ShouldNot(HaveOccurred()) + + var egressPorts []uint16 + for _, r := range np.Spec.Egress { + for _, p := range r.Destination.Ports { + egressPorts = append(egressPorts, p.MinPort) + } + } + Expect(egressPorts).To(ContainElement(uint16(9443)), "should include port from URL-style endpoint") + }) + + It("should default to 443 for https endpoints without explicit port", func() { + cfg := &otelcollector.Configuration{ + Installation: defaultInstallation, + OTelCollector: &operatorv1.OTelCollectorSpec{ + Exporters: []operatorv1.OTelExporter{ + {Name: "https-no-port", Endpoint: "https://otlp.example.com", Protocol: operatorv1.OTelProtocolHTTP}, + }, + }, + } + comp, err := otelcollector.OTelCollector(cfg) + Expect(err).NotTo(HaveOccurred()) + objs, _ := comp.Objects() + + np, err := rtest.GetResourceOfType[*v3.NetworkPolicy](objs, otelcollector.OTelCollectorPolicyName, otelcollector.OTelCollectorNamespace) + Expect(err).ShouldNot(HaveOccurred()) + + var egressPorts []uint16 + for _, r := range np.Spec.Egress { + for _, p := range r.Destination.Ports { + egressPorts = append(egressPorts, p.MinPort) + } + } + Expect(egressPorts).To(ContainElement(uint16(443)), "should default to 443 for https") + }) + + It("should add kube API server and prometheus egress rules when metrics are enabled", func() { + cfg := &otelcollector.Configuration{ + Installation: defaultInstallation, + OTelCollector: &operatorv1.OTelCollectorSpec{ + Metrics: &operatorv1.OTelMetrics{Enabled: ptr.To(operatorv1.OTelMetricsEnable)}, + Exporters: []operatorv1.OTelExporter{{Name: "backend", Endpoint: "otlp.example.com:4317"}}, + }, + } + comp, err := otelcollector.OTelCollector(cfg) + Expect(err).NotTo(HaveOccurred()) + objs, _ := comp.Objects() + + np, err := rtest.GetResourceOfType[*v3.NetworkPolicy](objs, otelcollector.OTelCollectorPolicyName, otelcollector.OTelCollectorNamespace) + Expect(err).ShouldNot(HaveOccurred()) + Expect(len(np.Spec.Egress)).To(BeNumerically(">=", 4)) + }) + }) + + Context("StatefulSet overrides", func() { + It("should apply overrides from the CR", func() { + affinity := &corev1.Affinity{ + NodeAffinity: &corev1.NodeAffinity{ + RequiredDuringSchedulingIgnoredDuringExecution: &corev1.NodeSelector{ + NodeSelectorTerms: []corev1.NodeSelectorTerm{{ + MatchExpressions: []corev1.NodeSelectorRequirement{{ + Key: "custom-key", + Operator: corev1.NodeSelectorOpExists, + }}, + }}, + }, + }, + } + containerResources := &corev1.ResourceRequirements{ + Limits: corev1.ResourceList{"cpu": resource.MustParse("500m")}, + Requests: corev1.ResourceList{"cpu": resource.MustParse("100m")}, + } + nodeSelector := map[string]string{"zone": "us-west-2a"} + tolerations := []corev1.Toleration{{Key: "dedicated", Operator: corev1.TolerationOpEqual, Value: "otel"}} + topologyConstraints := []corev1.TopologySpreadConstraint{{ + MaxSkew: 1, + TopologyKey: "topology.kubernetes.io/zone", + WhenUnsatisfiable: corev1.ScheduleAnyway, + LabelSelector: &metav1.LabelSelector{MatchLabels: map[string]string{"k8s-app": "otel-collector"}}, + }} + podLabels := map[string]string{"extra-label": "value"} + podAnnotations := map[string]string{"extra-annotation": "value"} + priorityClassName := "system-cluster-critical" + + cfg := &otelcollector.Configuration{ + Installation: defaultInstallation, + OTelCollector: &operatorv1.OTelCollectorSpec{ + Exporters: []operatorv1.OTelExporter{{Name: "backend", Endpoint: "otlp.example.com:4317"}}, + OTelCollectorStatefulSet: &operatorv1.OTelCollectorStatefulSet{ + Spec: &operatorv1.OTelCollectorStatefulSetSpec{ + Template: &operatorv1.OTelCollectorStatefulSetPodTemplateSpec{ + Metadata: &operatorv1.Metadata{ + Labels: podLabels, + Annotations: podAnnotations, + }, + Spec: &operatorv1.OTelCollectorStatefulSetPodSpec{ + Affinity: affinity, + Containers: []operatorv1.OTelCollectorStatefulSetContainer{{ + Name: "otel-collector", + Resources: containerResources, + }}, + NodeSelector: nodeSelector, + Tolerations: tolerations, + TopologySpreadConstraints: topologyConstraints, + PriorityClassName: priorityClassName, + }, + }, + }, + }, + }, + } + + component, err := otelcollector.OTelCollector(cfg) + Expect(err).NotTo(HaveOccurred()) + Expect(component.ResolveImages(nil)).NotTo(HaveOccurred()) + objs, _ := component.Objects() + + statefulSet, err := rtest.GetResourceOfType[*appsv1.StatefulSet](objs, otelcollector.OTelCollectorStatefulSetName, otelcollector.OTelCollectorNamespace) + Expect(err).ShouldNot(HaveOccurred()) + + Expect(statefulSet.Spec.Template.ObjectMeta.Labels).To(HaveKeyWithValue("extra-label", "value")) + Expect(statefulSet.Spec.Template.ObjectMeta.Labels).To(HaveKeyWithValue("k8s-app", otelcollector.OTelCollectorStatefulSetName)) + Expect(statefulSet.Spec.Template.ObjectMeta.Annotations).To(HaveKeyWithValue("extra-annotation", "value")) + Expect(statefulSet.Spec.Template.ObjectMeta.Annotations).To(HaveKey("hash.operator.tigera.io/otel-collector-config")) + Expect(statefulSet.Spec.Template.Spec.Affinity).To(Equal(affinity)) + Expect(statefulSet.Spec.Template.Spec.NodeSelector).To(Equal(nodeSelector)) + Expect(statefulSet.Spec.Template.Spec.Tolerations).To(Equal(tolerations)) + Expect(statefulSet.Spec.Template.Spec.TopologySpreadConstraints).To(Equal(topologyConstraints)) + Expect(statefulSet.Spec.Template.Spec.PriorityClassName).To(Equal(priorityClassName)) + Expect(statefulSet.Spec.Template.Spec.Containers[0].Resources).To(Equal(*containerResources)) + }) + }) + + Context("Pull secrets", func() { + It("should include pull secrets when configured", func() { + pullSecrets := []*corev1.Secret{ + {ObjectMeta: metav1.ObjectMeta{Name: "my-pull-secret", Namespace: "tigera-operator"}}, + } + cfg := &otelcollector.Configuration{ + Installation: defaultInstallation, + PullSecrets: pullSecrets, + OTelCollector: &operatorv1.OTelCollectorSpec{ + Exporters: []operatorv1.OTelExporter{{Name: "backend", Endpoint: "otlp.example.com:4317"}}, + }, + } + comp, err := otelcollector.OTelCollector(cfg) + Expect(err).NotTo(HaveOccurred()) + objs, _ := comp.Objects() + // 7 base objects + 1 copied pull secret + Expect(objs).To(HaveLen(8)) + }) + }) + + It("should support Linux OS type", func() { + component, err := otelcollector.OTelCollector(&otelcollector.Configuration{ + Installation: defaultInstallation, + OTelCollector: &operatorv1.OTelCollectorSpec{ + Exporters: []operatorv1.OTelExporter{{Name: "backend", Endpoint: "otlp.example.com:4317"}}, + }, + }) + Expect(err).NotTo(HaveOccurred()) + Expect(component.SupportedOSType()).To(Equal(rmeta.OSTypeLinux)) + }) + +}) diff --git a/pkg/render/otelcollector/suite_test.go b/pkg/render/otelcollector/suite_test.go new file mode 100644 index 0000000000..f2768136b1 --- /dev/null +++ b/pkg/render/otelcollector/suite_test.go @@ -0,0 +1,29 @@ +// Copyright (c) 2026 Tigera, Inc. All rights reserved. + +// 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 otelcollector_test + +import ( + "testing" + + "github.com/onsi/ginkgo/v2" + "github.com/onsi/gomega" +) + +func TestRender(t *testing.T) { + gomega.RegisterFailHandler(ginkgo.Fail) + suiteConfig, reporterConfig := ginkgo.GinkgoConfiguration() + reporterConfig.JUnitReport = "../../../report/ut/otelcollector_render_suite.xml" + ginkgo.RunSpecs(t, "pkg/render/otelcollector Suite", suiteConfig, reporterConfig) +}