LOG-7535: Add MaxUnavailable to API and enable kube caching

Author: jcantrill

api/observability/v1/clusterlogforwarder_types.go

@@ -17,6 +17,7 @@ package v1
 import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
 )
 
 // ClusterLogForwarderSpec defines the desired state of ClusterLogForwarder
@@ -128,6 +129,13 @@ type CollectorSpec struct {
 	// +kubebuilder:validation:Optional
 	// +operator-sdk:csv:customresourcedefinitions:type=spec,displayName="Affinity"
 	Affinity *corev1.Affinity `json:"affinity,omitempty"`
+
+	// Define the maxUnavailable pod rollout strategy
+	//
+	// Value can be a number (e.g., 50) or a percentage string (e.g., "50%").
+	// +kubebuilder:validation:Type=string
+	// +kubebuilder:validation:Pattern="^([0-9]{1,2}|100)$|^([0-9]{1,2}|100)%$"
+	MaxUnavailable *intstr.IntOrString `json:"maxUnavailable,omitempty"`
 }
 
 // PipelineSpec links a set of inputs and transformations to a set of outputs.

api/observability/v1/conditions.go

@@ -35,9 +35,6 @@ const (
 	// ConditionTypeMaxUnavailable validates the value of the max-unavailable-rollout annotation
 	ConditionTypeMaxUnavailable = GroupName + "/MaxUnavailableAnnotation"
 
-	// ConditionTypeUseKubeCache validates the value of the use-apiserver-cache annotation
-	ConditionTypeUseKubeCache = GroupName + "/UseKubeCacheAnnotation"
-
 	// ConditionTypeReady indicates the service is ready.
 	//
 	// Ready=True means the operands are running and providing some service.

api/observability/v1/zz_generated.deepcopy.go

@@ -948,6 +948,17 @@ spec:
                             type: array
                         type: object
                     type: object
+                  maxUnavailable:
+                    anyOf:
+                    - type: integer
+                    - type: string
+                    description: |-
+                      Define the maxUnavailable pod rollout strategy
+
+                      Value can be a number (e.g., 50) or a percentage string (e.g., "50%").
+                    pattern: ^([0-9]{1,2}|100)$|^([0-9]{1,2}|100)%$
+                    type: string
+                    x-kubernetes-int-or-string: true
                   nodeSelector:
                     additionalProperties:
                       type: string

bundle/manifests/observability.openshift.io_clusterlogforwarders.yaml

@@ -948,6 +948,17 @@ spec:
                             type: array
                         type: object
                     type: object
+                  maxUnavailable:
+                    anyOf:
+                    - type: integer
+                    - type: string
+                    description: |-
+                      Define the maxUnavailable pod rollout strategy
+
+                      Value can be a number (e.g., 50) or a percentage string (e.g., "50%").
+                    pattern: ^([0-9]{1,2}|100)$|^([0-9]{1,2}|100)%$
+                    type: string
+                    x-kubernetes-int-or-string: true
                   nodeSelector:
                     additionalProperties:
                       type: string

config/crd/bases/observability.openshift.io_clusterlogforwarders.yaml

@@ -74,6 +74,9 @@ Type:: object
 |Property|Type|Description
 
 |affinity|object|  Define scheduling rules that influence pod placement based on node or pod affinity/anti-affinity constraints.
+|maxUnavailable|object|  Define the maxUnavailable pod rollout strategy
+
+Value can be a number (e.g., 50) or a percentage string (e.g., "50%").
 |nodeSelector|object|  Define nodes for scheduling the pods.
 
 |resources|object|  The resource requirements for the collector
@@ -876,6 +879,19 @@ Type:: object
 
 Type:: array
 
+=== .spec.collector.maxUnavailable
+
+Type:: object
+
+[options="header"]
+|======================
+|Property|Type|Description
+
+|IntVal|int|  
+|StrVal|string|  
+|Type|int|  
+|======================
+
 === .spec.collector.nodeSelector
 
 Type:: object

docs/reference/operator/api_observability_v1.adoc

@@ -2,9 +2,10 @@ package collector
 
 import (
 	"fmt"
-	"k8s.io/apimachinery/pkg/util/intstr"
 	"strings"
 
+	"k8s.io/apimachinery/pkg/util/intstr"
+
 	log "github.com/ViaQ/logerr/v2/log/static"
 	"github.com/openshift/cluster-logging-operator/internal/auth"
 	"github.com/openshift/cluster-logging-operator/internal/collector/common"
@@ -22,6 +23,9 @@ import (
 )
 
 const (
+	//DefaultMaxUnavailable is the maxUnavailable collector setting when not defined by spec.collector.maxUnavailable
+	DefaultMaxUnavailable = "100%"
+
 	defaultAudience                 = "openshift"
 	clusterLoggingPriorityClassName = "system-node-critical"
 	MetricsPort                     = int32(24231)
@@ -69,9 +73,8 @@ type Factory struct {
 	PodLabelVisitor        PodLabelVisitor
 	ResourceNames          *factory.ForwarderResourceNames
 	isDaemonset            bool
-	LogLevel               string
-	UseKubeCache           bool
-	MaxUnavailable         string
+	//LogLevel               string
+	annotations map[string]string
 }
 
 // CollectorResourceRequirements returns the resource requirements for a given collector implementation
@@ -89,12 +92,22 @@ func (f *Factory) NodeSelector() map[string]string {
 func (f *Factory) Tolerations() []v1.Toleration {
 	return f.CollectorSpec.Tolerations
 }
-
 func (f *Factory) Affinity() *v1.Affinity {
 	return f.CollectorSpec.Affinity
 }
+func (f *Factory) MaxUnavailable() intstr.IntOrString {
+	if f.CollectorSpec.MaxUnavailable != nil {
+		return *f.CollectorSpec.MaxUnavailable
+	}
+	if f.annotations != nil {
+		if value, found := f.annotations[constants.AnnotationMaxUnavailable]; found {
+			return intstr.Parse(value)
+		}
+	}
+	return intstr.Parse(DefaultMaxUnavailable)
+}
 
-func New(confHash, clusterID string, collectorSpec *obs.CollectorSpec, secrets internalobs.Secrets, configMaps map[string]*v1.ConfigMap, forwarderSpec obs.ClusterLogForwarderSpec, resNames *factory.ForwarderResourceNames, isDaemonset bool, logLevel string, useCache bool, maxUnavailable string) *Factory {
+func New(confHash, clusterID string, collectorSpec *obs.CollectorSpec, secrets internalobs.Secrets, configMaps map[string]*v1.ConfigMap, forwarderSpec obs.ClusterLogForwarderSpec, resNames *factory.ForwarderResourceNames, isDaemonset bool, annotations map[string]string) *Factory {
 	if collectorSpec == nil {
 		collectorSpec = &obs.CollectorSpec{}
 	}
@@ -113,16 +126,14 @@ func New(confHash, clusterID string, collectorSpec *obs.CollectorSpec, secrets i
 		ResourceNames:   resNames,
 		PodLabelVisitor: vector.PodLogExcludeLabel,
 		isDaemonset:     isDaemonset,
-		LogLevel:        logLevel,
-		UseKubeCache:    useCache,
-		MaxUnavailable:  maxUnavailable,
+		annotations:     annotations,
 	}
 	return factory
 }
 
 func (f *Factory) NewDaemonSet(namespace, name string, trustedCABundle *v1.ConfigMap, tlsProfileSpec configv1.TLSProfileSpec) *apps.DaemonSet {
 	podSpec := f.NewPodSpec(trustedCABundle, f.ForwarderSpec, f.ClusterID, tlsProfileSpec, namespace)
-	ds := factory.NewDaemonSet(namespace, name, name, constants.CollectorName, constants.VectorName, f.MaxUnavailable, *podSpec, f.CommonLabelInitializer, f.PodLabelVisitor)
+	ds := factory.NewDaemonSet(namespace, name, name, constants.CollectorName, constants.VectorName, f.MaxUnavailable(), *podSpec, f.CommonLabelInitializer, f.PodLabelVisitor)
 	ds.Spec.Template.Annotations[constants.AnnotationSecretHash] = f.Secrets.Hash64a()
 	return ds
 }
@@ -143,6 +154,7 @@ func (f *Factory) NewPodSpec(trustedCABundle *v1.ConfigMap, spec obs.ClusterLogF
 		TerminationGracePeriodSeconds: utils.GetPtr[int64](10),
 		Tolerations:                   append(constants.DefaultTolerations(), f.Tolerations()...),
 		Affinity:                      f.Affinity(),
+
 		Volumes: []v1.Volume{
 			{Name: metricsVolumeName, VolumeSource: v1.VolumeSource{Secret: &v1.SecretVolumeSource{SecretName: f.ResourceNames.SecretMetrics}}},
 			{Name: tmpVolumeName, VolumeSource: v1.VolumeSource{EmptyDir: &v1.EmptyDirVolumeSource{Medium: v1.StorageMediumMemory}}},
@@ -172,7 +184,7 @@ func (f *Factory) NewPodSpec(trustedCABundle *v1.ConfigMap, spec obs.ClusterLogF
 
 	addTrustedCABundle(collector, podSpec, trustedCABundle)
 
-	f.Visit(collector, podSpec, f.ResourceNames, namespace, f.LogLevel)
+	f.Visit(collector, podSpec, f.ResourceNames, namespace, LogLevel(f.annotations))
 
 	podSpec.Containers = []v1.Container{
 		*collector,
@@ -413,3 +425,10 @@ func hasTrustedCABundle(configMap *v1.ConfigMap) (string, bool) {
 	caBundle, ok := configMap.Data[constants.TrustedCABundleKey]
 	return caBundle, ok && caBundle != ""
 }
+
+func LogLevel(annotations map[string]string) string {
+	if level, ok := annotations[constants.AnnotationVectorLogLevel]; ok {
+		return level
+	}
+	return "warn"
+}

internal/collector/collector.go

@@ -1,6 +1,9 @@
 package collector
 
 import (
+	"os"
+	"path"
+
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	obs "github.com/openshift/cluster-logging-operator/api/observability/v1"
@@ -14,14 +17,40 @@ import (
 	"github.com/openshift/cluster-logging-operator/internal/tls"
 	"github.com/openshift/cluster-logging-operator/internal/utils"
 	. "github.com/openshift/cluster-logging-operator/test/matchers"
-	"os"
-	"path"
-
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
 )
 
+var _ = Describe("Factory#MaxUnavalable", func() {
+	var (
+		factory Factory
+	)
+	BeforeEach(func() {
+		factory = Factory{
+			CollectorSpec: obs.CollectorSpec{},
+			annotations:   make(map[string]string),
+		}
+	})
+
+	Context("when evaluating MaxUnavailable", func() {
+		It("should apply the default when nothing is not defined", func() {
+			Expect(factory.MaxUnavailable().StrVal).To(Equal(DefaultMaxUnavailable))
+		})
+		It("should prefer spec over the deprecated annotation", func() {
+			exp := intstr.Parse("30%")
+			factory.CollectorSpec.MaxUnavailable = &exp
+			Expect(factory.MaxUnavailable()).To(Equal(exp))
+		})
+		It("should honor the deprecated annotation", func() {
+			exp := intstr.Parse("30%")
+			factory.annotations[constants.AnnotationMaxUnavailable] = exp.StrVal
+			Expect(factory.MaxUnavailable()).To(Equal(exp))
+		})
+	})
+})
+
 var _ = Describe("Factory#Daemonset", func() {
 	var (
 		podSpec   v1.PodSpec
@@ -99,7 +128,9 @@ var _ = Describe("Factory#Daemonset", func() {
 
 			It("should set VECTOR_LOG env variable with debug value", func() {
 				logLevelDebug := "debug"
-				factory.LogLevel = logLevelDebug
+				factory.annotations = map[string]string{
+					constants.AnnotationVectorLogLevel: logLevelDebug,
+				}
 
 				podSpec = *factory.NewPodSpec(nil, obs.ClusterLogForwarderSpec{}, "1234", tls.GetClusterTLSProfileSpec(nil), constants.OpenshiftNS)
 				collector = podSpec.Containers[0]

internal/collector/collector_test.go

@@ -18,16 +18,7 @@ const (
 
 	AnnotationSecretHash = "observability.openshift.io/secret-hash"
 
-	// AnnotationKubeCache is used to enable caching for requests to the kube-apiserver using vector kubernetes_logs source.
-	// Tech-Preview feature
-	//
-	// While enabling cache can significantly reduce Kubernetes control plane
-	// memory pressure, the trade-off is a chance of receiving stale data.
-	AnnotationKubeCache = "observability.openshift.io/use-apiserver-cache"
-
-	// AnnotationMaxUnavailable configures the maximum number of DaemonSet pods that can be unavailable during a rolling update.
-	// Tech-Preview feature
-	//
+	// AnnotationMaxUnavailable (Deprecated) configures the maximum number of DaemonSet pods that can be unavailable during a rolling update.
 	// This can be an absolute number (e.g., 1) or a percentage (e.g., 10%). Default is 100%.
 	AnnotationMaxUnavailable = "observability.openshift.io/max-unavailable-rollout"
 )

internal/constants/annotations.go

@@ -23,7 +23,6 @@ import (
 	"github.com/openshift/cluster-logging-operator/internal/runtime/serviceaccount"
 	"github.com/openshift/cluster-logging-operator/internal/tls"
 	"github.com/openshift/cluster-logging-operator/internal/utils"
-	"github.com/openshift/cluster-logging-operator/internal/validations/observability"
 	corev1 "k8s.io/api/core/v1"
 	networkingv1 "k8s.io/api/networking/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -46,11 +45,6 @@ func ReconcileCollector(context internalcontext.ForwarderContext, pollInterval,
 		options = context.AdditionalContext
 	}
 
-	// Set kubeapi and rollout options based on annotation (LOG-7196)
-	// TODO: replace with API fields
-	SetKubeCacheOption(context.Forwarder.Annotations, options)
-	SetMaxUnavailableRolloutOption(context.Forwarder.Annotations, options)
-
 	if internalobs.Outputs(context.Forwarder.Spec.Outputs).NeedServiceAccountToken() {
 		// temporarily create SA token until collector is capable of dynamically reloading a projected serviceaccount token
 		var sa *corev1.ServiceAccount
@@ -116,9 +110,7 @@ func ReconcileCollector(context internalcontext.ForwarderContext, pollInterval,
 		context.Forwarder.Spec,
 		resourceNames,
 		isDaemonSet,
-		LogLevel(context.Forwarder.Annotations),
-		factory.IncludesKubeCacheOption(options),
-		factory.GetMaxUnavailableValue(options),
+		context.Forwarder.Annotations,
 	)
 
 	if err = collectorFactory.ReconcileCollectorConfig(context.Client, context.Reader, context.Forwarder.Namespace, collectorConfig, ownerRef); err != nil {
@@ -164,7 +156,7 @@ func ReconcileCollector(context internalcontext.ForwarderContext, pollInterval,
 func GenerateConfig(k8Client client.Client, clf obs.ClusterLogForwarder, resourceNames factory.ForwarderResourceNames, secrets internalobs.Secrets, op framework.Options) (config string, err error) {
 	tlsProfile, _ := tls.FetchAPIServerTlsProfile(k8Client)
 	op[framework.ClusterTLSProfileSpec] = tls.GetClusterTLSProfileSpec(tlsProfile)
-	//EvaluateAnnotationsForEnabledCapabilities(clusterRequest.Forwarder, op)
+	EvaluateAnnotationsForEnabledCapabilities(clf.Annotations, op)
 	g := forwardergenerator.New()
 	generatedConfig, err := g.GenerateConf(secrets, clf.Spec, clf.Namespace, clf.Name, resourceNames, op)
 
@@ -188,41 +180,6 @@ func EvaluateAnnotationsForEnabledCapabilities(annotations map[string]string, op
 			if strings.ToLower(value) == "true" {
 				options[generatorhelpers.EnableDebugOutput] = "true"
 			}
-		case constants.AnnotationKubeCache:
-			// Matching the validate_annotations logic
-			if observability.IsEnabledValue(value) {
-				options[framework.UseKubeCacheOption] = "true"
-			}
-		case constants.AnnotationMaxUnavailable:
-			// Matching the validate_annotations logic
-			if observability.IsPercentOrWholeNumber(value) {
-				options[framework.MaxUnavailableOption] = value
-			}
-		}
-	}
-}
-
-func LogLevel(annotations map[string]string) string {
-	if level, ok := annotations[constants.AnnotationVectorLogLevel]; ok {
-		return level
-	}
-	return "warn"
-}
-
-func SetKubeCacheOption(annotations map[string]string, options framework.Options) {
-	if value, found := annotations[constants.AnnotationKubeCache]; found {
-		if observability.IsEnabledValue(value) {
-			log.V(3).Info("Kube cache annotation found")
-			options[framework.UseKubeCacheOption] = "true"
-		}
-	}
-}
-
-func SetMaxUnavailableRolloutOption(annotations map[string]string, options framework.Options) {
-	if value, found := annotations[constants.AnnotationMaxUnavailable]; found {
-		if observability.IsPercentOrWholeNumber(value) {
-			log.V(3).Info("Max Unavailable annotation found")
-			options[framework.MaxUnavailableOption] = value
 		}
 	}
 }