Skip to content

Commit d155d40

Browse files
leandroberettaleandroberetta
committed
feedback addressed
1 parent 2061148 commit d155d40

16 files changed

Lines changed: 280 additions & 179 deletions

‎api/flowcollector/v1beta2/flowcollector_types.go‎

Lines changed: 10 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -754,8 +754,11 @@ type FlowCollectorFLP struct {
754754
Service *ProcessorServiceConfig `json:"service,omitempty"`
755755

756756
// `centralizedInformers` configuration for centralized Kubernetes informers that push cache updates to flowlogs-pipeline processors.
757-
// This reduces load on the Kubernetes API server by having a single component (flowlogs-pipeline-informers) query the API instead of N FLP processors.
758-
// When enabled, a dedicated `flowlogs-pipeline-informers` deployment is created that watches Kubernetes resources and pushes updates via gRPC.
757+
// This reduces load on the Kubernetes API server by having a single component query the API instead of N FLP processors.
758+
// When enabled, a dedicated deployment is created that watches Kubernetes resources and pushes updates via gRPC.
759+
// Benefits: Reduced API server load on large clusters with many FLP replicas.
760+
// Drawbacks: More complex deployment (additional component), higher resource usage on small clusters.
761+
// Recommended only for clusters with many FLP replicas (>3) or when API server load is a concern.
759762
// +optional
760763
CentralizedInformers *FlowCollectorCentralizedInformers `json:"centralizedInformers,omitempty"`
761764

@@ -769,14 +772,15 @@ type FlowCollectorFLP struct {
769772
// `FlowCollectorCentralizedInformers` defines the configuration for centralized Kubernetes informers
770773
type FlowCollectorCentralizedInformers struct {
771774
// `enabled` controls whether to deploy centralized Kubernetes informers.
772-
// When `true`, a dedicated `flowlogs-pipeline-informers` deployment watches K8s resources and pushes cache updates via gRPC to FLP processors.
773-
// When `false`, each FLP processor uses local informers (previous behavior).
775+
// When `true`, a dedicated deployment watches K8s resources and pushes cache updates via gRPC to FLP processors, reducing API server load.
776+
// When `false` (default), each FLP processor uses local informers.
777+
// Enable only on large clusters or when API server load is a concern, as it adds deployment complexity.
774778
// +kubebuilder:default:=false
775779
Enabled *bool `json:"enabled,omitempty"`
776780

777781
// `replicas` defines the number of replicas for the flowlogs-pipeline-informers deployment.
778782
// For high availability, a minimum of 2 replicas is required when `enabled` is `true`.
779-
// +kubebuilder:validation:Minimum=1
783+
// +kubebuilder:validation:Minimum=2
780784
// +kubebuilder:default:=2
781785
Replicas *int32 `json:"replicas,omitempty"`
782786

@@ -824,7 +828,7 @@ type AdvancedCentralizedInformersConfig struct {
824828
// `processorPort` defines the gRPC port where flowlogs-pipeline processors listen for k8s cache updates.
825829
// +kubebuilder:validation:Minimum=1
826830
// +kubebuilder:validation:Maximum=65535
827-
// +kubebuilder:default:=9090
831+
// +kubebuilder:default:=9402
828832
// +optional
829833
ProcessorPort *int32 `json:"processorPort,omitempty"`
830834
}

‎api/flowcollector/v1beta2/flowcollector_validation_port_conflicts_test.go‎

Lines changed: 147 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,147 @@
1+
package v1beta2
2+
3+
import (
4+
"context"
5+
"testing"
6+
7+
"github.com/stretchr/testify/assert"
8+
"k8s.io/utils/ptr"
9+
)
10+
11+
func TestPortConflictValidation(t *testing.T) {
12+
tests := []struct {
13+
name string
14+
spec FlowCollectorSpec
15+
expectError bool
16+
errorContains string
17+
}{
18+
{
19+
name: "FLP port conflicts with k8scache port when informers enabled",
20+
spec: FlowCollectorSpec{
21+
Processor: FlowCollectorFLP{
22+
Advanced: &AdvancedProcessorConfig{
23+
Port: ptr.To(int32(9402)),
24+
},
25+
CentralizedInformers: &FlowCollectorCentralizedInformers{
26+
Enabled: ptr.To(true),
27+
},
28+
},
29+
},
30+
expectError: true,
31+
errorContains: "spec.processor.advanced.port 9402 conflicts with reserved k8scache port 9402",
32+
},
33+
{
34+
name: "Health port conflicts with k8scache port when informers enabled",
35+
spec: FlowCollectorSpec{
36+
Processor: FlowCollectorFLP{
37+
Advanced: &AdvancedProcessorConfig{
38+
HealthPort: ptr.To(int32(9402)),
39+
},
40+
CentralizedInformers: &FlowCollectorCentralizedInformers{
41+
Enabled: ptr.To(true),
42+
},
43+
},
44+
},
45+
expectError: true,
46+
errorContains: "spec.processor.advanced.healthPort 9402 conflicts with reserved k8scache port 9402",
47+
},
48+
{
49+
name: "Metrics port conflicts with k8scache port when informers enabled",
50+
spec: FlowCollectorSpec{
51+
Processor: FlowCollectorFLP{
52+
Metrics: FLPMetrics{
53+
Server: MetricsServerConfig{
54+
Port: ptr.To(int32(9402)),
55+
},
56+
},
57+
CentralizedInformers: &FlowCollectorCentralizedInformers{
58+
Enabled: ptr.To(true),
59+
},
60+
},
61+
},
62+
expectError: true,
63+
errorContains: "spec.processor.metrics.server.port 9402 conflicts with reserved k8scache port 9402",
64+
},
65+
{
66+
name: "Profile port conflicts with k8scache port when informers enabled",
67+
spec: FlowCollectorSpec{
68+
Processor: FlowCollectorFLP{
69+
Advanced: &AdvancedProcessorConfig{
70+
ProfilePort: ptr.To(int32(9402)),
71+
},
72+
CentralizedInformers: &FlowCollectorCentralizedInformers{
73+
Enabled: ptr.To(true),
74+
},
75+
},
76+
},
77+
expectError: true,
78+
errorContains: "spec.processor.advanced.profilePort 9402 conflicts with reserved k8scache port 9402",
79+
},
80+
{
81+
name: "Port 9402 is allowed when informers disabled",
82+
spec: FlowCollectorSpec{
83+
Processor: FlowCollectorFLP{
84+
Advanced: &AdvancedProcessorConfig{
85+
Port: ptr.To(int32(9402)),
86+
},
87+
CentralizedInformers: &FlowCollectorCentralizedInformers{
88+
Enabled: ptr.To(false),
89+
},
90+
},
91+
},
92+
expectError: false,
93+
},
94+
{
95+
name: "Port 9402 is allowed when informers is nil",
96+
spec: FlowCollectorSpec{
97+
Processor: FlowCollectorFLP{
98+
Advanced: &AdvancedProcessorConfig{
99+
Port: ptr.To(int32(9402)),
100+
},
101+
CentralizedInformers: nil,
102+
},
103+
},
104+
expectError: false,
105+
},
106+
{
107+
name: "Valid configuration with no port conflicts",
108+
spec: FlowCollectorSpec{
109+
Processor: FlowCollectorFLP{
110+
Advanced: &AdvancedProcessorConfig{
111+
Port: ptr.To(int32(2055)),
112+
HealthPort: ptr.To(int32(8080)),
113+
ProfilePort: ptr.To(int32(6060)),
114+
},
115+
Metrics: FLPMetrics{
116+
Server: MetricsServerConfig{
117+
Port: ptr.To(int32(9102)),
118+
},
119+
},
120+
CentralizedInformers: &FlowCollectorCentralizedInformers{
121+
Enabled: ptr.To(true),
122+
},
123+
},
124+
},
125+
expectError: false,
126+
},
127+
}
128+
129+
for _, tt := range tests {
130+
t.Run(tt.name, func(t *testing.T) {
131+
fc := &FlowCollector{
132+
Spec: tt.spec,
133+
}
134+
135+
_, err := fc.Validate(context.Background(), fc)
136+
137+
if tt.expectError {
138+
assert.Error(t, err)
139+
if tt.errorContains != "" {
140+
assert.Contains(t, err.Error(), tt.errorContains)
141+
}
142+
} else {
143+
assert.NoError(t, err)
144+
}
145+
})
146+
}
147+
}

‎api/flowcollector/v1beta2/flowcollector_validation_webhook.go‎

Lines changed: 46 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -262,7 +262,7 @@ func (v *validator) validateFLP() {
262262
v.validateFLPMetricsForAlerts()
263263
v.validateFLPMetricsIncludeLists()
264264
v.validateFLPTLS()
265-
v.validateInformers()
265+
v.validatePortConflicts()
266266
}
267267

268268
func (v *validator) validateScheduling() {
@@ -467,26 +467,55 @@ func (v *validator) validateFLPTLS() {
467467
}
468468
}
469469

470-
func (v *validator) validateInformers() {
471-
if v.fc.Processor.CentralizedInformers == nil {
470+
func (v *validator) validatePortConflicts() {
471+
// Only check port conflicts when centralized informers are enabled (when k8scache port is actually used)
472+
if !v.fc.Processor.IsCentralizedInformersEnabled() {
472473
return
473474
}
474475

475-
// Check if enabled
476-
enabled := v.fc.Processor.CentralizedInformers.Enabled != nil && *v.fc.Processor.CentralizedInformers.Enabled
476+
// k8scache port is hardcoded to 9402 when centralized informers are enabled
477+
const k8scachePort = 9402
477478

478-
if enabled {
479-
// When enabled, replicas must be at least 2 for high availability
480-
replicas := int32(2) // default
481-
if v.fc.Processor.CentralizedInformers.Replicas != nil {
482-
replicas = *v.fc.Processor.CentralizedInformers.Replicas
483-
}
484-
if replicas < 2 {
485-
v.errors = append(
486-
v.errors,
487-
fmt.Errorf("spec.processor.informers.replicas must be at least 2 when informers are enabled (got %d). Centralized informers require high availability to avoid losing the entire flow collection pipeline in case of failure", replicas),
488-
)
489-
}
479+
// Get advanced processor config with defaults
480+
var port, healthPort, profilePort *int32
481+
metricsPort := v.fc.Processor.GetMetricsPort()
482+
483+
if v.fc.Processor.Advanced != nil {
484+
port = v.fc.Processor.Advanced.Port
485+
healthPort = v.fc.Processor.Advanced.HealthPort
486+
profilePort = v.fc.Processor.Advanced.ProfilePort
487+
}
488+
489+
// Check FLP port
490+
if port != nil && *port == k8scachePort {
491+
v.errors = append(
492+
v.errors,
493+
fmt.Errorf("spec.processor.advanced.port %d conflicts with reserved k8scache port %d used by centralized informers", *port, k8scachePort),
494+
)
495+
}
496+
497+
// Check health port
498+
if healthPort != nil && *healthPort == k8scachePort {
499+
v.errors = append(
500+
v.errors,
501+
fmt.Errorf("spec.processor.advanced.healthPort %d conflicts with reserved k8scache port %d used by centralized informers", *healthPort, k8scachePort),
502+
)
503+
}
504+
505+
// Check metrics port
506+
if metricsPort == k8scachePort {
507+
v.errors = append(
508+
v.errors,
509+
fmt.Errorf("spec.processor.metrics.server.port %d conflicts with reserved k8scache port %d used by centralized informers", metricsPort, k8scachePort),
510+
)
511+
}
512+
513+
// Check profile port (optional)
514+
if profilePort != nil && *profilePort == k8scachePort {
515+
v.errors = append(
516+
v.errors,
517+
fmt.Errorf("spec.processor.advanced.profilePort %d conflicts with reserved k8scache port %d used by centralized informers", *profilePort, k8scachePort),
518+
)
490519
}
491520
}
492521

‎api/flowcollector/v1beta2/helper.go‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -237,6 +237,25 @@ func (spec *FlowCollectorFLP) IsUnmanagedFLPReplicas() bool {
237237
return spec.KafkaConsumerAutoscaler.IsHPAEnabled()
238238
}
239239

240+
func (spec *FlowCollectorFLP) IsCentralizedInformersEnabled() bool {
241+
return spec.CentralizedInformers != nil && spec.CentralizedInformers.Enabled != nil && *spec.CentralizedInformers.Enabled
242+
}
243+
244+
func (spec *FlowCollectorCentralizedInformers) GetTLSType() TLSConfigType {
245+
if spec == nil || spec.TLS == nil {
246+
return TLSAuto
247+
}
248+
return spec.TLS.Type
249+
}
250+
251+
func (spec *FlowCollectorCentralizedInformers) UsesOpenShiftServiceCA(isOpenShift bool) bool {
252+
if !isOpenShift {
253+
return false
254+
}
255+
tlsType := spec.GetTLSType()
256+
return tlsType == TLSAuto
257+
}
258+
240259
func (spec *FlowCollectorConsolePlugin) IsUnmanagedConsolePluginReplicas() bool {
241260
if spec.UnmanagedReplicas {
242261
return true

‎bundle/manifests/flows.netobserv.io_flowcollectors.yaml‎

Lines changed: 10 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -5453,8 +5453,11 @@ spec:
54535453
centralizedInformers:
54545454
description: |-
54555455
`centralizedInformers` configuration for centralized Kubernetes informers that push cache updates to flowlogs-pipeline processors.
5456-
This reduces load on the Kubernetes API server by having a single component (flowlogs-pipeline-informers) query the API instead of N FLP processors.
5457-
When enabled, a dedicated `flowlogs-pipeline-informers` deployment is created that watches Kubernetes resources and pushes updates via gRPC.
5456+
This reduces load on the Kubernetes API server by having a single component query the API instead of N FLP processors.
5457+
When enabled, a dedicated deployment is created that watches Kubernetes resources and pushes updates via gRPC.
5458+
Benefits: Reduced API server load on large clusters with many FLP replicas.
5459+
Drawbacks: More complex deployment (additional component), higher resource usage on small clusters.
5460+
Recommended only for clusters with many FLP replicas (>3) or when API server load is a concern.
54585461
properties:
54595462
advanced:
54605463
description: '`advanced` allows setting some technical parameters
@@ -5467,7 +5470,7 @@ spec:
54675470
minimum: 1
54685471
type: integer
54695472
processorPort:
5470-
default: 9090
5473+
default: 9402
54715474
description: '`processorPort` defines the gRPC port where
54725475
flowlogs-pipeline processors listen for k8s cache updates.'
54735476
format: int32
@@ -5497,16 +5500,17 @@ spec:
54975500
default: false
54985501
description: |-
54995502
`enabled` controls whether to deploy centralized Kubernetes informers.
5500-
When `true`, a dedicated `flowlogs-pipeline-informers` deployment watches K8s resources and pushes cache updates via gRPC to FLP processors.
5501-
When `false`, each FLP processor uses local informers (previous behavior).
5503+
When `true`, a dedicated deployment watches K8s resources and pushes cache updates via gRPC to FLP processors, reducing API server load.
5504+
When `false` (default), each FLP processor uses local informers.
5505+
Enable only on large clusters or when API server load is a concern, as it adds deployment complexity.
55025506
type: boolean
55035507
replicas:
55045508
default: 2
55055509
description: |-
55065510
`replicas` defines the number of replicas for the flowlogs-pipeline-informers deployment.
55075511
For high availability, a minimum of 2 replicas is required when `enabled` is `true`.
55085512
format: int32
5509-
minimum: 1
5513+
minimum: 2
55105514
type: integer
55115515
resources:
55125516
default:

‎config/crd/bases/flows.netobserv.io_flowcollectors.yaml‎

Lines changed: 10 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -5039,8 +5039,11 @@ spec:
50395039
centralizedInformers:
50405040
description: |-
50415041
`centralizedInformers` configuration for centralized Kubernetes informers that push cache updates to flowlogs-pipeline processors.
5042-
This reduces load on the Kubernetes API server by having a single component (flowlogs-pipeline-informers) query the API instead of N FLP processors.
5043-
When enabled, a dedicated `flowlogs-pipeline-informers` deployment is created that watches Kubernetes resources and pushes updates via gRPC.
5042+
This reduces load on the Kubernetes API server by having a single component query the API instead of N FLP processors.
5043+
When enabled, a dedicated deployment is created that watches Kubernetes resources and pushes updates via gRPC.
5044+
Benefits: Reduced API server load on large clusters with many FLP replicas.
5045+
Drawbacks: More complex deployment (additional component), higher resource usage on small clusters.
5046+
Recommended only for clusters with many FLP replicas (>3) or when API server load is a concern.
50445047
properties:
50455048
advanced:
50465049
description: '`advanced` allows setting some technical parameters of the informers component.'
@@ -5051,7 +5054,7 @@ spec:
50515054
minimum: 1
50525055
type: integer
50535056
processorPort:
5054-
default: 9090
5057+
default: 9402
50555058
description: '`processorPort` defines the gRPC port where flowlogs-pipeline processors listen for k8s cache updates.'
50565059
format: int32
50575060
maximum: 65535
@@ -5077,16 +5080,17 @@ spec:
50775080
default: false
50785081
description: |-
50795082
`enabled` controls whether to deploy centralized Kubernetes informers.
5080-
When `true`, a dedicated `flowlogs-pipeline-informers` deployment watches K8s resources and pushes cache updates via gRPC to FLP processors.
5081-
When `false`, each FLP processor uses local informers (previous behavior).
5083+
When `true`, a dedicated deployment watches K8s resources and pushes cache updates via gRPC to FLP processors, reducing API server load.
5084+
When `false` (default), each FLP processor uses local informers.
5085+
Enable only on large clusters or when API server load is a concern, as it adds deployment complexity.
50825086
type: boolean
50835087
replicas:
50845088
default: 2
50855089
description: |-
50865090
`replicas` defines the number of replicas for the flowlogs-pipeline-informers deployment.
50875091
For high availability, a minimum of 2 replicas is required when `enabled` is `true`.
50885092
format: int32
5089-
minimum: 1
5093+
minimum: 2
50905094
type: integer
50915095
resources:
50925096
default:

0 commit comments

Comments
 (0)