feat(vector): add support for raising file descriptor limits

Add annotation-driven support for Vector's VECTOR_RAISE_FD_LIMIT env var,
enabling the collector to raise file descriptor soft limits at startup.
This prevents "Too many open files" errors when monitoring large numbers
of log files concurrently.

Key changes:
- New annotation: observability.openshift.io/raise-fd-limit (default: false)
- Sets VECTOR_RAISE_FD_LIMIT env var on the collector container
- Vector reads the env var natively, no entrypoint script changes needed
- Input validation rejects invalid values (only "true"/"false" allowed)
- Status condition reports validation errors to the user

Default matches Vector's behavior (false = don't raise fd limits).
Users must explicitly set the annotation to "true" to opt in.

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>

Author: vparfonov

api/observability/v1/conditions.go

@@ -35,6 +35,9 @@ const (
 	// ConditionTypeMaxUnavailable validates the value of the max-unavailable-rollout annotation
 	ConditionTypeMaxUnavailable = GroupName + "/MaxUnavailableAnnotation"
 
+	// ConditionTypeRaiseFdLimit validates the value of the raise-fd-limit annotation
+	ConditionTypeRaiseFdLimit = GroupName + "/RaiseFdLimitAnnotation"
+
 	// ConditionTypeReady indicates the service is ready.
 	//
 	// Ready=True means the operands are running and providing some service.
@@ -89,6 +92,9 @@ const (
 	// ReasonKubeCacheSupported indicates the support for the use-apiserver-cache annotation value
 	ReasonKubeCacheSupported = "KubeCacheAnnotationSupported"
 
+	// ReasonRaiseFdLimitSupported indicates the support for the raise-fd-limit annotation value
+	ReasonRaiseFdLimitSupported = "RaiseFdLimitAnnotationSupported"
+
 	// ReasonReconciliationComplete when the operator has initialized, validated, and deployed the resources for the workload
 	ReasonReconciliationComplete = "ReconciliationComplete"
 

docs/features/raise_fd_limit.adoc

@@ -0,0 +1,38 @@
+== Vector Collector: Raising File Descriptor Limits
+
+This feature enables Vector to automatically raise its file descriptor soft limit to match the hard limit at startup. This prevents "Too many open files" errors when Vector is monitoring a large number of log files concurrently.
+
+Systems often default to restrictive file descriptor soft limits (e.g. 1024 on Linux), which can cause Vector to fail when processing many log sources simultaneously. When enabled, the operator sets the `VECTOR_RAISE_FD_LIMIT` environment variable, which Vector reads natively to raise the soft limit at startup without requiring manual sysadmin intervention.
+
+Annotation: `observability.openshift.io/raise-fd-limit`.
+
+The Cluster Logging Operator does not enable this feature by default (`false`), matching Vector's default behavior.
+
+Supported values are:
+
+. `true` - raise the file descriptor soft limit to match the hard limit
+. `false` (default) - do not modify file descriptor limits
+
+=== Example
+.Enable Raising File Descriptor Limits
+[source]
+----
+apiVersion: "observability.openshift.io/v1"
+kind: ClusterLogForwarder
+metadata:
+  name: instance
+  namespace: openshift-logging
+  annotations:
+    observability.openshift.io/raise-fd-limit: "true"
+spec:
+  outputs:
+  - name: devel
+    type: elasticsearch
+  pipelines:
+   - name: devel-logs
+     inputRefs:
+     - application
+     outputRefs:
+     - devel
+----
+This configuration will enable Vector to raise its file descriptor soft limit at startup.

internal/collector/collector.go

@@ -208,6 +208,7 @@ func (f *Factory) NewCollectorContainer(inputs internalobs.Inputs, outputs inter
 		{Name: "OPENSHIFT_CLUSTER_ID", Value: clusterID},
 		{Name: "POD_IP", ValueFrom: &v1.EnvVarSource{FieldRef: &v1.ObjectFieldSelector{APIVersion: "v1", FieldPath: "status.podIP"}}},
 		{Name: "POD_IPS", ValueFrom: &v1.EnvVarSource{FieldRef: &v1.ObjectFieldSelector{APIVersion: "v1", FieldPath: "status.podIPs"}}},
+		{Name: "VECTOR_RAISE_FD_LIMIT", Value: RaiseFdLimit(f.annotations)},
 	}
 	collector.Env = append(collector.Env, utils.GetProxyEnvVars()...)
 
@@ -405,3 +406,13 @@ func LogLevel(annotations map[string]string) string {
 	}
 	return "warn"
 }
+
+func RaiseFdLimit(annotations map[string]string) string {
+	if value, ok := annotations[constants.AnnotationVectorRaiseFdLimit]; ok {
+		if value == "true" || value == "false" {
+			return value
+		}
+		log.V(0).Info("Invalid value for annotation, using default", "annotation", constants.AnnotationVectorRaiseFdLimit, "value", value, "default", "false")
+	}
+	return "false"
+}

internal/collector/collector_test.go

@@ -138,6 +138,30 @@ var _ = Describe("Factory#Daemonset", func() {
 				Expect(collector.Env).To(IncludeEnvVar(v1.EnvVar{Name: "VECTOR_LOG", Value: logLevelDebug}))
 			})
 
+			It("should set VECTOR_RAISE_FD_LIMIT to false by default", func() {
+				Expect(collector.Env).To(IncludeEnvVar(v1.EnvVar{Name: "VECTOR_RAISE_FD_LIMIT", Value: "false"}))
+			})
+
+			It("should set VECTOR_RAISE_FD_LIMIT to true when annotation is set", func() {
+				factory.annotations = map[string]string{
+					constants.AnnotationVectorRaiseFdLimit: "true",
+				}
+
+				podSpec = *factory.NewPodSpec(nil, obs.ClusterLogForwarderSpec{}, "1234", tls.GetClusterTLSProfileSpec(nil), constants.OpenshiftNS)
+				collector = podSpec.Containers[0]
+				Expect(collector.Env).To(IncludeEnvVar(v1.EnvVar{Name: "VECTOR_RAISE_FD_LIMIT", Value: "true"}))
+			})
+
+			It("should default VECTOR_RAISE_FD_LIMIT to false when annotation has invalid value", func() {
+				factory.annotations = map[string]string{
+					constants.AnnotationVectorRaiseFdLimit: "yes",
+				}
+
+				podSpec = *factory.NewPodSpec(nil, obs.ClusterLogForwarderSpec{}, "1234", tls.GetClusterTLSProfileSpec(nil), constants.OpenshiftNS)
+				collector = podSpec.Containers[0]
+				Expect(collector.Env).To(IncludeEnvVar(v1.EnvVar{Name: "VECTOR_RAISE_FD_LIMIT", Value: "false"}))
+			})
+
 			Context("the volume mounts", func() {
 				It("should mount all output configmaps", func() {
 					Expect(collector.VolumeMounts).To(IncludeVolumeMount(

internal/constants/annotations.go

@@ -13,6 +13,10 @@ const (
 	// CLO's default log level for vector is `warn`: https://issues.redhat.com/browse/LOG-3435
 	AnnotationVectorLogLevel = "observability.openshift.io/log-level"
 
+	// AnnotationVectorRaiseFdLimit controls whether Vector raises its file descriptor soft limit at startup.
+	// Valid values are "true" or "false". Default is "false" (matching Vector's default).
+	AnnotationVectorRaiseFdLimit = "observability.openshift.io/raise-fd-limit"
+
 	AnnotationSecretHash    = "observability.openshift.io/secret-hash"
 	AnnotationConfigMapHash = "observability.openshift.io/configmap-hash"
 

internal/validations/observability/validate.go

@@ -14,6 +14,7 @@ var (
 	clfValidators = []func(internalcontext.ForwarderContext){
 		validateLogLevelAnnotation,
 		validateMaxUnavailableAnnotation,
+		validateRaiseFdLimitAnnotation,
 		ValidatePermissions,
 		inputs.Validate,
 		outputs.Validate,

internal/validations/observability/validate_annotations.go

@@ -19,6 +19,7 @@ const (
 var (
 	compiledMaxUnavailableRegex = regexp.MustCompile(validMaxUnavailableRegex)
 	allowedLogLevels            = sets.NewString("trace", "debug", "info", "warn", "error", "off")
+	allowedRaiseFdLimitValues   = sets.NewString("true", "false")
 	enabledValues               = sets.NewString("true", "enabled")
 )
 
@@ -53,6 +54,18 @@ func validateLogLevelAnnotation(context internalcontext.ForwarderContext) {
 	internalobs.RemoveConditionByType(&context.Forwarder.Status.Conditions, obs.ConditionTypeLogLevel)
 }
 
+func validateRaiseFdLimitAnnotation(context internalcontext.ForwarderContext) {
+	if value, ok := context.Forwarder.Annotations[constants.AnnotationVectorRaiseFdLimit]; ok {
+		if !allowedRaiseFdLimitValues.Has(value) {
+			condition := internalobs.NewCondition(obs.ConditionTypeRaiseFdLimit, obs.ConditionFalse, obs.ReasonRaiseFdLimitSupported, "")
+			condition.Message = fmt.Sprintf("raise-fd-limit value %q must be one of [true, false]", value)
+			internalobs.SetCondition(&context.Forwarder.Status.Conditions, condition)
+			return
+		}
+	}
+	internalobs.RemoveConditionByType(&context.Forwarder.Status.Conditions, obs.ConditionTypeRaiseFdLimit)
+}
+
 func IsEnabledValue(val string) bool {
 	return enabledValues.Has(strings.ToLower(val))
 }

internal/validations/observability/validate_annotations_test.go

@@ -49,6 +49,27 @@ var _ = Describe("[internal][validations] validate clusterlogforwarder annotatio
 			Entry("should pass with level off", "off"))
 	})
 
+	Context("#validateRaiseFdLimit", func() {
+		It("should pass validation if no annotations are set", func() {
+			validateRaiseFdLimitAnnotation(context)
+			Expect(clf.Status.Conditions).To(BeEmpty())
+		})
+
+		It("should fail validation if value is not true or false", func() {
+			clf.Annotations = map[string]string{constants.AnnotationVectorRaiseFdLimit: "yes"}
+			validateRaiseFdLimitAnnotation(context)
+			Expect(clf.Status.Conditions).To(HaveCondition(obs.ConditionTypeRaiseFdLimit, false, obs.ReasonRaiseFdLimitSupported, ".*must be one of.*"))
+		})
+
+		DescribeTable("valid raise-fd-limit values", func(value string) {
+			clf.Annotations = map[string]string{constants.AnnotationVectorRaiseFdLimit: value}
+			validateRaiseFdLimitAnnotation(context)
+			Expect(clf.Status.Conditions).To(BeEmpty())
+		},
+			Entry("should pass with value true", "true"),
+			Entry("should pass with value false", "false"))
+	})
+
 	Context("#validateMaxUnavailable", func() {
 		It("should pass validation if no annotations are set", func() {
 			validateMaxUnavailableAnnotation(context)

test/functional/misc/raise_fd_limit_test.go

@@ -0,0 +1,33 @@
+package misc
+
+import (
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	obs "github.com/openshift/cluster-logging-operator/api/observability/v1"
+	"github.com/openshift/cluster-logging-operator/test/framework/functional"
+	testruntime "github.com/openshift/cluster-logging-operator/test/runtime/observability"
+)
+
+var _ = Describe("[Functional][Misc][RaiseFdLimit] Vector raise-fd-limit", func() {
+
+	var framework *functional.CollectorFunctionalFramework
+
+	BeforeEach(func() {
+		framework = functional.NewCollectorFunctionalFramework()
+		testruntime.NewClusterLogForwarderBuilder(framework.Forwarder).
+			FromInput(obs.InputTypeInfrastructure).
+			ToHttpOutput()
+	})
+
+	AfterEach(func() {
+		framework.Cleanup()
+	})
+
+	It("should start successfully with VECTOR_RAISE_FD_LIMIT env var", func() {
+		Expect(framework.Deploy()).To(BeNil())
+
+		logs, err := framework.ReadCollectorLogs()
+		Expect(err).To(BeNil())
+		Expect(logs).To(ContainSubstring("Vector has started."))
+	})
+})