Skip to content

Commit 741b60a

Browse files
csatib02csatib02
authored
Merge pull request #213 from kube-logging/fix/gate-custom-components
fix: gate custom components if not using own image
2 parents 31cc1b5 + 296da90 commit 741b60a

9 files changed

Lines changed: 190 additions & 31 deletions

File tree

‎.claude/settings.json‎

Lines changed: 1 addition & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,6 @@
33
"companyAnnouncements": [
44
"Welcome to the Telemetry-controller project!\nIf you have any questions or need assistance, feel free to reach out to the maintainers or the community.\nWe are excited to have you on board and look forward to your contributions."
55
],
6-
"model": "sonnet",
76
"cleanupPeriodDays": 14,
87
"env": {
98
"DISABLE_TELEMETRY": "1",
@@ -78,7 +77,6 @@
7877
"Read(./**/tmp/**)",
7978
"Bash(rm -rf *)",
8079
"Bash(rm -fr *)",
81-
"Bash(git push *)",
8280
"Bash(*exec*)",
8381
"Bash(sudo *)",
8482
"Bash(*chmod 777*)",
@@ -95,10 +93,7 @@
9593
"enableAllProjectMcpServers": true,
9694
"enabledPlugins": {
9795
"context7@claude-plugins-official": true,
98-
"feature-dev@claude-plugins-official": true,
9996
"code-simplifier@claude-plugins-official": true,
100-
"security-guidance@claude-plugins-official": true,
101-
"claude-md-management@claude-plugins-official": true,
102-
"gopls-lsp@claude-plugins-official": true
97+
"security-guidance@claude-plugins-official": true
10398
}
10499
}

‎.gitignore‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,3 +38,6 @@ go.work.sum
3838
devspace.yaml
3939

4040
.envrc
41+
42+
**/settings.local.json
43+
**/CLAUDE.local.md

‎pkg/resources/manager/collector_manager.go‎

Lines changed: 18 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -46,8 +46,9 @@ import (
4646
)
4747

4848
const (
49-
otelCollectorKind = "OpenTelemetryCollector"
50-
axoflowOtelCollectorImageRef = "ghcr.io/axoflow/axoflow-otel-collector/axoflow-otel-collector:0.152.0-axoflow.1"
49+
otelCollectorKind = "OpenTelemetryCollector"
50+
axoflowOtelCollectorImageRepo = "ghcr.io/axoflow/axoflow-otel-collector"
51+
axoflowOtelCollectorImageRef = axoflowOtelCollectorImageRepo + "/axoflow-otel-collector:0.152.0-axoflow.1"
5152
)
5253

5354
var (
@@ -121,6 +122,11 @@ func (c *CollectorManager) BuildConfigInputForCollector(ctx context.Context, col
121122
}
122123
}
123124

125+
var collectorImage string
126+
if collector.Spec.OtelCommonFields != nil {
127+
collectorImage = collector.Spec.OtelCommonFields.Image
128+
}
129+
124130
return otelcolconfgen.OtelColConfigInput{
125131
ResourceRelations: components.ResourceRelations{
126132
Tenants: tenants,
@@ -130,9 +136,10 @@ func (c *CollectorManager) BuildConfigInputForCollector(ctx context.Context, col
130136
TenantSubscriptionMap: tenantSubscriptionMap,
131137
SubscriptionOutputMap: subscriptionOutputMap,
132138
},
133-
Debug: utils.DerefOrZero(collector.Spec.Debug),
134-
DryRunMode: utils.DerefOrZero(collector.Spec.DryRunMode),
135-
MemoryLimiter: *collector.Spec.MemoryLimiter,
139+
Debug: utils.DerefOrZero(collector.Spec.Debug),
140+
DryRunMode: utils.DerefOrZero(collector.Spec.DryRunMode),
141+
MemoryLimiter: *collector.Spec.MemoryLimiter,
142+
IsAxoflowDistribution: isAxoflowDistribution(collectorImage),
136143
}, nil
137144
}
138145

@@ -593,3 +600,9 @@ func setOtelCommonFieldsDefaults(otelCommonFields *otelv1beta1.OpenTelemetryComm
593600
func authSecretName(collectorName string) string {
594601
return fmt.Sprintf("otelcollector-%s-auth", collectorName)
595602
}
603+
604+
func isAxoflowDistribution(image string) bool {
605+
return image == "" ||
606+
strings.HasPrefix(image, axoflowOtelCollectorImageRepo+"/") ||
607+
strings.HasPrefix(image, axoflowOtelCollectorImageRepo+":")
608+
}

‎pkg/resources/manager/collector_manager_test.go‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -195,3 +195,24 @@ func TestSetOtelCommonFieldsDefaults(t *testing.T) {
195195
})
196196
}
197197
}
198+
199+
func TestIsAxoflowDistribution(t *testing.T) {
200+
tests := []struct {
201+
name string
202+
image string
203+
expected bool
204+
}{
205+
{name: "Empty image uses axoflow default", image: "", expected: true},
206+
{name: "Default axoflow image", image: axoflowOtelCollectorImageRef, expected: true},
207+
{name: "Custom axoflow tag", image: axoflowOtelCollectorImageRepo + "/axoflow-otel-collector:dev", expected: true},
208+
{name: "Upstream collector image", image: "otel/opentelemetry-collector-contrib:0.152.0", expected: false},
209+
{name: "Unrelated custom image", image: "ghcr.io/example/custom-collector:1.0.0", expected: false},
210+
{name: "Lookalike repo is not axoflow distribution", image: axoflowOtelCollectorImageRepo + "-fork:latest", expected: false},
211+
}
212+
213+
for _, tt := range tests {
214+
t.Run(tt.name, func(t *testing.T) {
215+
assert.Equal(t, tt.expected, isAxoflowDistribution(tt.image))
216+
})
217+
}
218+
}

‎pkg/resources/otel_conf_gen/otel_conf_gen.go‎

Lines changed: 12 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,9 @@ type OtelColConfigInput struct {
4141
MemoryLimiter v1alpha1.MemoryLimiter
4242
Debug bool
4343
DryRunMode bool
44+
45+
// IsAxoflowDistribution gates components only available in that distribution.
46+
IsAxoflowDistribution bool
4447
}
4548

4649
func (cfgInput *OtelColConfigInput) IsEmpty() bool {
@@ -75,7 +78,7 @@ func (cfgInput *OtelColConfigInput) generateExporters(ctx context.Context) map[s
7578
maps.Copy(exporters, exporter.GenerateMetricsExporters())
7679
maps.Copy(exporters, exporter.GenerateOTLPGRPCExporters(ctx, cfgInput.ResourceRelations))
7780
maps.Copy(exporters, exporter.GenerateOTLPHTTPExporters(ctx, cfgInput.ResourceRelations))
78-
maps.Copy(exporters, exporter.GenerateFluentforwardExporters(ctx, cfgInput.ResourceRelations))
81+
maps.Copy(exporters, exporter.GenerateFluentforwardExporters(ctx, cfgInput.ResourceRelations, cfgInput.IsAxoflowDistribution))
7982
maps.Copy(exporters, exporter.GenerateFileExporter(ctx, cfgInput.ResourceRelations))
8083
maps.Copy(exporters, exporter.GenerateElasticsearchExporters(ctx, cfgInput.ResourceRelations))
8184
maps.Copy(exporters, exporter.GenerateAWSS3Exporters(ctx, cfgInput.ResourceRelations))
@@ -177,7 +180,9 @@ func (cfgInput *OtelColConfigInput) generateConnectors() map[string]any {
177180

178181
if !cfgInput.DryRunMode {
179182
maps.Copy(connectors, connector.GenerateCountConnectors())
180-
maps.Copy(connectors, connector.GenerateBytesConnectors())
183+
if cfgInput.IsAxoflowDistribution {
184+
maps.Copy(connectors, connector.GenerateBytesConnectors())
185+
}
181186
}
182187

183188
for _, tenant := range cfgInput.Tenants {
@@ -216,7 +221,7 @@ func (cfgInput *OtelColConfigInput) generateNamedPipelines() map[string]*otelv1b
216221
}
217222

218223
if !cfgInput.DryRunMode {
219-
maps.Copy(namedPipelines, pipeline.GenerateMetricsPipelines())
224+
maps.Copy(namedPipelines, pipeline.GenerateMetricsPipelines(cfgInput.IsAxoflowDistribution))
220225
}
221226

222227
for _, tenant := range tenants {
@@ -261,7 +266,10 @@ func (cfgInput *OtelColConfigInput) generateNamedPipelines() map[string]*otelv1b
261266
}
262267
// GetExporterNameForOutput returns a non-empty name for any supported exporter.
263268
if exporterName := components.GetExporterNameForOutput(output.Output); exporterName != "" {
264-
exporters = append(exporters, exporterName, outputCountConnectorName, outputBytesConnectorName)
269+
exporters = append(exporters, exporterName, outputCountConnectorName)
270+
if cfgInput.IsAxoflowDistribution {
271+
exporters = append(exporters, outputBytesConnectorName)
272+
}
265273
}
266274
}
267275

‎pkg/resources/otel_conf_gen/otel_conf_gen_test.go‎

Lines changed: 47 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -104,6 +104,7 @@ func TestOtelColConfComplex(t *testing.T) {
104104
},
105105
}
106106
inputCfg := OtelColConfigInput{
107+
IsAxoflowDistribution: true,
107108
ResourceRelations: components.ResourceRelations{
108109
Subscriptions: subscriptions,
109110
Tenants: []v1alpha1.Tenant{
@@ -464,7 +465,8 @@ func TestOtelColConfigInput_generateNamedPipelines(t *testing.T) {
464465
{
465466
name: "Single tenant with no subscriptions",
466467
cfgInput: OtelColConfigInput{
467-
DryRunMode: false,
468+
DryRunMode: false,
469+
IsAxoflowDistribution: true,
468470
ResourceRelations: components.ResourceRelations{
469471
Bridges: nil,
470472
OutputsWithSecretData: nil,
@@ -508,9 +510,53 @@ func TestOtelColConfigInput_generateNamedPipelines(t *testing.T) {
508510
),
509511
},
510512
},
513+
{
514+
name: "Non-axoflow distribution omits the bytes pipeline",
515+
cfgInput: OtelColConfigInput{
516+
DryRunMode: false,
517+
IsAxoflowDistribution: false,
518+
ResourceRelations: components.ResourceRelations{
519+
Bridges: nil,
520+
OutputsWithSecretData: nil,
521+
TenantSubscriptionMap: map[string][]v1alpha1.NamespacedName{
522+
"tenant1": {
523+
{
524+
Namespace: "ns1",
525+
Name: "sub1",
526+
},
527+
},
528+
},
529+
SubscriptionOutputMap: map[v1alpha1.NamespacedName][]v1alpha1.NamespacedName{
530+
{
531+
Namespace: "ns1",
532+
Name: "sub1",
533+
}: {},
534+
},
535+
},
536+
},
537+
expectedPipelines: map[string]*otelv1beta1.Pipeline{
538+
"logs/tenant_tenant1": pipeline.GenerateRootPipeline([]v1alpha1.Tenant{}, "tenant1", false),
539+
"logs/tenant_tenant1_subscription_ns1_sub1": pipeline.GeneratePipeline(
540+
[]string{"routing/tenant_tenant1_subscriptions"},
541+
[]string{"attributes/subscription_sub1"},
542+
[]string{"routing/subscription_ns1_sub1_outputs"},
543+
),
544+
"metrics/output": pipeline.GeneratePipeline(
545+
[]string{"count/output_metrics"},
546+
[]string{"deltatocumulative", "attributes/metricattributes"},
547+
[]string{"prometheus/message_metrics_exporter"},
548+
),
549+
"metrics/tenant": pipeline.GeneratePipeline(
550+
[]string{"count/tenant_metrics"},
551+
[]string{"deltatocumulative", "attributes/metricattributes"},
552+
[]string{"prometheus/message_metrics_exporter"},
553+
),
554+
},
555+
},
511556
{
512557
name: "Three tenants two bridges",
513558
cfgInput: OtelColConfigInput{
559+
IsAxoflowDistribution: true,
514560
ResourceRelations: components.ResourceRelations{
515561
Tenants: []v1alpha1.Tenant{
516562
{

‎pkg/resources/otel_conf_gen/pipeline/components/exporter/fluent_forward_exporter.go‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -45,7 +45,7 @@ type FluentForwardWrapper struct {
4545
Kubernetes *v1alpha1.KubernetesMetadata `json:"kubernetes_metadata,omitempty"`
4646
}
4747

48-
func (w *FluentForwardWrapper) mapToFluentForwardWrapper(userConfig *v1alpha1.Fluentforward) {
48+
func (w *FluentForwardWrapper) mapToFluentForwardWrapper(userConfig *v1alpha1.Fluentforward, isAxoflowDistribution bool) {
4949
w.QueueConfig = &queueWrapper{}
5050
w.RetryConfig = &backOffWrapper{}
5151
w.QueueConfig.setDefaultQueueSettings(userConfig.QueueConfig)
@@ -63,20 +63,20 @@ func (w *FluentForwardWrapper) mapToFluentForwardWrapper(userConfig *v1alpha1.Fl
6363
if userConfig.DefaultLabelsEnabled != nil {
6464
w.DefaultLabelsEnabled = userConfig.DefaultLabelsEnabled
6565
}
66-
if userConfig.Kubernetes != nil {
66+
if isAxoflowDistribution && userConfig.Kubernetes != nil {
6767
w.Kubernetes = userConfig.Kubernetes
6868
}
6969
w.TCPClientSettings = userConfig.TCPClientSettings
7070
}
7171

72-
func GenerateFluentforwardExporters(ctx context.Context, resourceRelations components.ResourceRelations) map[string]any {
72+
func GenerateFluentforwardExporters(ctx context.Context, resourceRelations components.ResourceRelations, isAxoflowDistribution bool) map[string]any {
7373
logger := log.FromContext(ctx)
7474

7575
result := make(map[string]any)
7676
for _, output := range resourceRelations.OutputsWithSecretData {
7777
if output.Output.Spec.Fluentforward != nil {
7878
internalConfig := FluentForwardWrapper{}
79-
internalConfig.mapToFluentForwardWrapper(output.Output.Spec.Fluentforward)
79+
internalConfig.mapToFluentForwardWrapper(output.Output.Spec.Fluentforward, isAxoflowDistribution)
8080
tenant, err := resourceRelations.FindTenantForOutput(output.Output.NamespacedName())
8181
if err != nil {
8282
logger.Error(err, "failed to find tenant for output, skipping", "output", output.Output.NamespacedName().String())

‎pkg/resources/otel_conf_gen/pipeline/components/exporter/fluent_forward_exporter_test.go‎

Lines changed: 76 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -30,12 +30,14 @@ const testTenantName = "tenant1"
3030

3131
func TestGenerateFluentforwardExporters(t *testing.T) {
3232
tests := []struct {
33-
name string
34-
resourceRelations components.ResourceRelations
35-
expectedResult map[string]any
33+
name string
34+
resourceRelations components.ResourceRelations
35+
isAxoflowDistribution bool
36+
expectedResult map[string]any
3637
}{
3738
{
38-
name: "Valid config",
39+
name: "Valid config",
40+
isAxoflowDistribution: true,
3941
resourceRelations: components.ResourceRelations{
4042
Tenants: []v1alpha1.Tenant{
4143
{
@@ -125,7 +127,8 @@ func TestGenerateFluentforwardExporters(t *testing.T) {
125127
},
126128
},
127129
{
128-
name: "All fields set, tls settings omitted",
130+
name: "All fields set, tls settings omitted",
131+
isAxoflowDistribution: true,
129132
resourceRelations: components.ResourceRelations{
130133
Tenants: []v1alpha1.Tenant{
131134
{
@@ -233,11 +236,78 @@ func TestGenerateFluentforwardExporters(t *testing.T) {
233236
},
234237
},
235238
},
239+
{
240+
name: "kubernetes_metadata dropped for non-axoflow distribution",
241+
isAxoflowDistribution: false,
242+
resourceRelations: components.ResourceRelations{
243+
Tenants: []v1alpha1.Tenant{
244+
{
245+
ObjectMeta: metav1.ObjectMeta{
246+
Name: testTenantName,
247+
},
248+
},
249+
},
250+
OutputsWithSecretData: []components.OutputWithSecretData{
251+
{
252+
Output: v1alpha1.Output{
253+
ObjectMeta: metav1.ObjectMeta{
254+
Name: "output3",
255+
Namespace: "default",
256+
},
257+
Spec: v1alpha1.OutputSpec{
258+
Fluentforward: &v1alpha1.Fluentforward{
259+
TCPClientSettings: v1alpha1.TCPClientSettings{
260+
Endpoint: &v1alpha1.Endpoint{
261+
TCPAddr: new("http://example.com"),
262+
},
263+
},
264+
Kubernetes: &v1alpha1.KubernetesMetadata{Key: "key", IncludePodLabels: true},
265+
},
266+
},
267+
},
268+
},
269+
},
270+
TenantSubscriptionMap: map[string][]v1alpha1.NamespacedName{
271+
testTenantName: {
272+
{
273+
Name: "sub1",
274+
Namespace: "default",
275+
},
276+
},
277+
},
278+
SubscriptionOutputMap: map[v1alpha1.NamespacedName][]v1alpha1.NamespacedName{
279+
{
280+
Name: "sub1",
281+
Namespace: "default",
282+
}: {
283+
{
284+
Name: "output3",
285+
Namespace: "default",
286+
},
287+
},
288+
},
289+
},
290+
expectedResult: map[string]any{
291+
"fluentforwardexporter/default_output3": map[string]any{
292+
"endpoint": map[string]any{
293+
"tcp_addr": "http://example.com",
294+
},
295+
"sending_queue": map[string]any{
296+
"enabled": true,
297+
"queue_size": float64(1000),
298+
},
299+
"retry_on_failure": map[string]any{
300+
"enabled": true,
301+
"max_elapsed_time": "0s",
302+
},
303+
},
304+
},
305+
},
236306
}
237307

238308
for _, tt := range tests {
239309
t.Run(tt.name, func(t *testing.T) {
240-
assert.Equal(t, tt.expectedResult, GenerateFluentforwardExporters(context.TODO(), tt.resourceRelations))
310+
assert.Equal(t, tt.expectedResult, GenerateFluentforwardExporters(context.TODO(), tt.resourceRelations, tt.isAxoflowDistribution))
241311
})
242312
}
243313
}

‎pkg/resources/otel_conf_gen/pipeline/pipeline.go‎

Lines changed: 8 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -57,7 +57,7 @@ func GenerateRootPipeline(tenants []v1alpha1.Tenant, tenantName string, dryRunMo
5757
return GeneratePipeline([]string{receiverName}, []string{"k8sattributes", fmt.Sprintf("attributes/tenant_%s", tenantName), "filter/exclude"}, []string{exporterName, tenantCountConnectorName})
5858
}
5959

60-
func GenerateMetricsPipelines() map[string]*otelv1beta1.Pipeline {
60+
func GenerateMetricsPipelines(includeOutputBytes bool) map[string]*otelv1beta1.Pipeline {
6161
metricsPipelines := make(map[string]*otelv1beta1.Pipeline)
6262
metricsPipelines["metrics/tenant"] = &otelv1beta1.Pipeline{
6363
Receivers: []string{"count/tenant_metrics"},
@@ -70,10 +70,13 @@ func GenerateMetricsPipelines() map[string]*otelv1beta1.Pipeline {
7070
Exporters: []string{exporter.DefaultPrometheusExporterID},
7171
}
7272

73-
metricsPipelines["metrics/output_bytes"] = &otelv1beta1.Pipeline{
74-
Receivers: []string{"bytes/exporter"},
75-
Processors: []string{processor.DefaultDeltaToCumulativeProcessorID, "attributes/metricattributes"},
76-
Exporters: []string{exporter.DefaultPrometheusExporterID},
73+
// The bytesconnector feeding this pipeline is only available in the axoflow-otel-collector distribution.
74+
if includeOutputBytes {
75+
metricsPipelines["metrics/output_bytes"] = &otelv1beta1.Pipeline{
76+
Receivers: []string{"bytes/exporter"},
77+
Processors: []string{processor.DefaultDeltaToCumulativeProcessorID, "attributes/metricattributes"},
78+
Exporters: []string{exporter.DefaultPrometheusExporterID},
79+
}
7780
}
7881

7982
return metricsPipelines

0 commit comments

Comments
 (0)