Skip to content

Commit 4443476

Browse files
pepovpepov
authored
Merge pull request #1402 from kube-logging/syslog-ng-output-config
syslog-ng aggregator fixes
2 parents b25cfef + 6d15842 commit 4443476

File tree

6 files changed

+178
-15
lines changed

6 files changed

+178
-15
lines changed

‎pkg/resources/fluentbit/configsecret.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -337,7 +337,7 @@ func (r *Reconciler) configSecret() (runtime.Object, reconciler.DesiredState, er
337337

338338
if r.Logging.Spec.SyslogNGSpec != nil {
339339
input.SyslogNGOutput = &syslogNGOutputConfig{}
340-
input.SyslogNGOutput.Host = fmt.Sprintf("%s.%s.svc.cluster.local", r.Logging.QualifiedName(syslogng.ServiceName), r.Logging.Spec.ControlNamespace)
340+
input.SyslogNGOutput.Host = fmt.Sprintf("%s.%s.svc%s", r.Logging.QualifiedName(syslogng.ServiceName), r.Logging.Spec.ControlNamespace, r.Logging.ClusterDomainAsSuffix())
341341
input.SyslogNGOutput.Port = syslogng.ServicePort
342342
input.SyslogNGOutput.JSONDateKey = "ts"
343343
input.SyslogNGOutput.JSONDateFormat = "iso8601"

‎pkg/sdk/logging/model/syslogng/config/config.go‎

Lines changed: 19 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -15,12 +15,14 @@
1515
package config
1616

1717
import (
18-
"errors"
1918
"io"
2019
"reflect"
2120

21+
"emperror.dev/errors"
2222
"github.com/cisco-open/operator-tools/pkg/secret"
2323
"github.com/siliconbrain/go-seqs/seqs"
24+
"k8s.io/apimachinery/pkg/types"
25+
"sigs.k8s.io/controller-runtime/pkg/client"
2426

2527
"github.com/kube-logging/logging-operator/pkg/sdk/logging/api/v1beta1"
2628
"github.com/kube-logging/logging-operator/pkg/sdk/logging/model/syslogng/config/render"
@@ -62,6 +64,8 @@ func configRenderer(in Input) (render.Renderer, error) {
6264
return nil, errors.New("missing syslog-ng spec")
6365
}
6466

67+
var errs error
68+
6569
// TODO: this should happen at the spec level, in something like `SyslogNGSpec.FinalGlobalOptions() GlobalOptions`
6670
if in.Logging.Spec.SyslogNGSpec.Metrics != nil {
6771
setDefault(&in.Logging.Spec.SyslogNGSpec.GlobalOptions, &v1beta1.GlobalOptions{})
@@ -79,7 +83,12 @@ func configRenderer(in Input) (render.Renderer, error) {
7983
globalOptions := renderAny(in.Logging.Spec.SyslogNGSpec.GlobalOptions, in.SecretLoaderFactory.SecretLoaderForNamespace(in.Logging.Namespace))
8084

8185
destinationDefs := make([]render.Renderer, 0, len(in.ClusterOutputs)+len(in.Outputs))
86+
clusterOutputRefs := make(map[string]types.NamespacedName, len(in.ClusterOutputs))
8287
for _, co := range in.ClusterOutputs {
88+
clusterOutputRefs[co.Name] = types.NamespacedName{
89+
Namespace: co.Namespace,
90+
Name: co.Name,
91+
}
8392
destinationDefs = append(destinationDefs, renderClusterOutput(co, in.SecretLoaderFactory))
8493
}
8594
for _, o := range in.Outputs {
@@ -88,10 +97,16 @@ func configRenderer(in Input) (render.Renderer, error) {
8897

8998
logDefs := make([]render.Renderer, 0, len(in.ClusterFlows)+len(in.Flows))
9099
for _, cf := range in.ClusterFlows {
91-
logDefs = append(logDefs, renderClusterFlow(sourceName, cf, in.SecretLoaderFactory))
100+
if err := validateClusterOutputs(clusterOutputRefs, client.ObjectKeyFromObject(&cf).String(), cf.Spec.GlobalOutputRefs); err != nil {
101+
errs = errors.Append(errs, err)
102+
}
103+
logDefs = append(logDefs, renderClusterFlow(clusterOutputRefs, sourceName, cf, in.SecretLoaderFactory))
92104
}
93105
for _, f := range in.Flows {
94-
logDefs = append(logDefs, renderFlow(in.Logging.Spec.ControlNamespace, sourceName, keyDelim(in.Logging.Spec.SyslogNGSpec.JSONKeyDelimiter), f, in.SecretLoaderFactory))
106+
if err := validateClusterOutputs(clusterOutputRefs, client.ObjectKeyFromObject(&f).String(), f.Spec.GlobalOutputRefs); err != nil {
107+
errs = errors.Append(errs, err)
108+
}
109+
logDefs = append(logDefs, renderFlow(clusterOutputRefs, sourceName, keyDelim(in.Logging.Spec.SyslogNGSpec.JSONKeyDelimiter), f, in.SecretLoaderFactory))
95110
}
96111

97112
if in.Logging.Spec.SyslogNGSpec.JSONKeyPrefix == "" {
@@ -132,7 +147,7 @@ func configRenderer(in Input) (render.Renderer, error) {
132147
func(rnd render.Renderer) bool { return rnd != nil },
133148
),
134149
render.Line(render.Empty),
135-
)), nil
150+
)), errs
136151
}
137152

138153
func versionStmt(version string) render.Renderer {

‎pkg/sdk/logging/model/syslogng/config/config_test.go‎

Lines changed: 131 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -395,6 +395,137 @@ destination "output_default_my-output" {
395395
};
396396
`,
397397
},
398+
"clusteroutput with flow ref": {
399+
input: Input{
400+
Logging: v1beta1.Logging{
401+
ObjectMeta: metav1.ObjectMeta{
402+
Name: "test",
403+
},
404+
Spec: v1beta1.LoggingSpec{
405+
SyslogNGSpec: &v1beta1.SyslogNGSpec{},
406+
ControlNamespace: "logging",
407+
},
408+
},
409+
Flows: []v1beta1.SyslogNGFlow{
410+
{
411+
ObjectMeta: metav1.ObjectMeta{
412+
Namespace: "default",
413+
Name: "test-flow",
414+
},
415+
Spec: v1beta1.SyslogNGFlowSpec{
416+
GlobalOutputRefs: []string{
417+
"clusterout",
418+
},
419+
},
420+
},
421+
},
422+
ClusterOutputs: []v1beta1.SyslogNGClusterOutput{
423+
{
424+
ObjectMeta: metav1.ObjectMeta{
425+
Name: "clusterout",
426+
Namespace: "logging",
427+
},
428+
Spec: v1beta1.SyslogNGClusterOutputSpec{
429+
SyslogNGOutputSpec: v1beta1.SyslogNGOutputSpec{
430+
Syslog: &output.SyslogOutput{
431+
Host: "127.0.0.1",
432+
},
433+
},
434+
},
435+
},
436+
},
437+
SecretLoaderFactory: &TestSecretLoaderFactory{},
438+
SourcePort: 601,
439+
},
440+
wantOut: `@version: current
441+
442+
@include "scl.conf"
443+
444+
source "main_input" {
445+
channel {
446+
source {
447+
network(flags("no-parse") port(601) transport("tcp"));
448+
};
449+
parser {
450+
json-parser(prefix("json."));
451+
};
452+
};
453+
};
454+
455+
destination "clusteroutput_logging_clusterout" {
456+
syslog("127.0.0.1" persist_name("clusteroutput_logging_clusterout"));
457+
};
458+
459+
filter "flow_default_test-flow_ns_filter" {
460+
match("default" value("json.kubernetes.namespace_name") type("string"));
461+
};
462+
log {
463+
source("main_input");
464+
filter("flow_default_test-flow_ns_filter");
465+
destination("clusteroutput_logging_clusterout");
466+
};
467+
`,
468+
},
469+
"flow referencing non-existent cluster output": {
470+
input: Input{
471+
Logging: v1beta1.Logging{
472+
ObjectMeta: metav1.ObjectMeta{
473+
Name: "test",
474+
},
475+
Spec: v1beta1.LoggingSpec{
476+
SyslogNGSpec: &v1beta1.SyslogNGSpec{},
477+
ControlNamespace: "logging",
478+
},
479+
},
480+
Flows: []v1beta1.SyslogNGFlow{
481+
{
482+
ObjectMeta: metav1.ObjectMeta{
483+
Namespace: "default",
484+
Name: "test-flow",
485+
},
486+
Spec: v1beta1.SyslogNGFlowSpec{
487+
GlobalOutputRefs: []string{
488+
"clusterout",
489+
},
490+
},
491+
},
492+
},
493+
ClusterOutputs: nil,
494+
SecretLoaderFactory: &TestSecretLoaderFactory{},
495+
SourcePort: 601,
496+
},
497+
wantErr: true,
498+
},
499+
"clusterFlow referencing non-existent cluster output": {
500+
input: Input{
501+
Logging: v1beta1.Logging{
502+
ObjectMeta: metav1.ObjectMeta{
503+
Name: "test",
504+
},
505+
Spec: v1beta1.LoggingSpec{
506+
SyslogNGSpec: &v1beta1.SyslogNGSpec{},
507+
ControlNamespace: "logging",
508+
},
509+
},
510+
ClusterFlows: []v1beta1.SyslogNGClusterFlow{
511+
{
512+
ObjectMeta: metav1.ObjectMeta{
513+
Namespace: "default",
514+
Name: "test-flow",
515+
},
516+
Spec: v1beta1.SyslogNGClusterFlowSpec{
517+
GlobalOutputRefs: []string{
518+
"clusterout",
519+
},
520+
},
521+
},
522+
},
523+
ClusterOutputs: nil,
524+
SecretLoaderFactory: &TestSecretLoaderFactory{},
525+
SourcePort: 601,
526+
},
527+
wantErr: true,
528+
},
398529
"parser": {
399530
input: Input{
400531
Logging: v1beta1.Logging{

‎pkg/sdk/logging/model/syslogng/config/flow.go‎

Lines changed: 22 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -20,16 +20,28 @@ import (
2020
"strconv"
2121
"strings"
2222

23+
"emperror.dev/errors"
2324
"github.com/cisco-open/operator-tools/pkg/secret"
25+
"github.com/siliconbrain/go-seqs/seqs"
26+
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
27+
"k8s.io/apimachinery/pkg/types"
28+
2429
"github.com/kube-logging/logging-operator/pkg/sdk/logging/api/v1beta1"
2530
"github.com/kube-logging/logging-operator/pkg/sdk/logging/model/syslogng/config/model"
2631
"github.com/kube-logging/logging-operator/pkg/sdk/logging/model/syslogng/config/render"
2732
filter "github.com/kube-logging/logging-operator/pkg/sdk/logging/model/syslogng/filter"
28-
"github.com/siliconbrain/go-seqs/seqs"
29-
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
3033
)
3134

32-
func renderClusterFlow(sourceName string, f v1beta1.SyslogNGClusterFlow, secretLoaderFactory SecretLoaderFactory) render.Renderer {
35+
func validateClusterOutputs(clusterOutputRefs map[string]types.NamespacedName, flow string, globalOutputRefs []string) error {
36+
return seqs.Reduce(seqs.FromSlice(globalOutputRefs), nil, func(err error, ref string) error {
37+
if _, ok := clusterOutputRefs[ref]; !ok {
38+
return errors.Append(err, errors.Errorf("cluster output reference %s for flow %s cannot be found", ref, flow))
39+
}
40+
return err
41+
})
42+
}
43+
44+
func renderClusterFlow(clusterOutputRefs map[string]types.NamespacedName, sourceName string, f v1beta1.SyslogNGClusterFlow, secretLoaderFactory SecretLoaderFactory) render.Renderer {
3345
baseName := fmt.Sprintf("clusterflow_%s_%s", f.Namespace, f.Name)
3446
matchName := fmt.Sprintf("%s_match", baseName)
3547
filterDefs := seqs.MapWithIndex(seqs.FromSlice(f.Spec.Filters), func(idx int, flt v1beta1.SyslogNGFilter) render.Renderer {
@@ -48,12 +60,14 @@ func renderClusterFlow(sourceName string, f v1beta1.SyslogNGClusterFlow, secretL
4860
return parenDefStmt(filterKind(flt), render.Literal(filterID(flt, idx, baseName)))
4961
}),
5062
)),
51-
seqs.ToSlice(seqs.Map(seqs.FromSlice(f.Spec.GlobalOutputRefs), func(ref string) string { return clusterOutputDestName(f.Namespace, ref) })),
63+
seqs.ToSlice(seqs.Map(seqs.FromSlice(f.Spec.GlobalOutputRefs), func(ref string) string {
64+
return clusterOutputDestName(clusterOutputRefs[ref].Namespace, ref)
65+
})),
5266
),
5367
)
5468
}
5569

56-
func renderFlow(controlNS string, sourceName string, keyDelim string, f v1beta1.SyslogNGFlow, secretLoaderFactory SecretLoaderFactory) render.Renderer {
70+
func renderFlow(clusterOutputRefs map[string]types.NamespacedName, sourceName string, keyDelim string, f v1beta1.SyslogNGFlow, secretLoaderFactory SecretLoaderFactory) render.Renderer {
5771
baseName := fmt.Sprintf("flow_%s_%s", f.Namespace, f.Name)
5872
matchName := fmt.Sprintf("%s_match", baseName)
5973
nsFilterName := fmt.Sprintf("%s_ns_filter", baseName)
@@ -80,7 +94,9 @@ func renderFlow(controlNS string, sourceName string, keyDelim string, f v1beta1.
8094
}),
8195
)),
8296
seqs.ToSlice(seqs.Concat(
83-
seqs.Map(seqs.FromSlice(f.Spec.GlobalOutputRefs), func(ref string) string { return clusterOutputDestName(f.Namespace, ref) }),
97+
seqs.Map(seqs.FromSlice(f.Spec.GlobalOutputRefs), func(ref string) string {
98+
return clusterOutputDestName(clusterOutputRefs[ref].Namespace, ref)
99+
}),
84100
seqs.Map(seqs.FromSlice(f.Spec.LocalOutputRefs), func(ref string) string { return outputDestName(f.Namespace, ref) }),
85101
)),
86102
),

‎pkg/sdk/logging/model/syslogng/config/flow_test.go‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -18,11 +18,12 @@ import (
1818
"strings"
1919
"testing"
2020

21-
"github.com/kube-logging/logging-operator/pkg/sdk/logging/api/v1beta1"
22-
"github.com/kube-logging/logging-operator/pkg/sdk/logging/model/syslogng/config/render"
2321
"github.com/stretchr/testify/assert"
2422
"github.com/stretchr/testify/require"
2523
v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
24+
25+
"github.com/kube-logging/logging-operator/pkg/sdk/logging/api/v1beta1"
26+
"github.com/kube-logging/logging-operator/pkg/sdk/logging/model/syslogng/config/render"
2627
)
2728

2829
func TestRenderClusterFlow(t *testing.T) {
@@ -65,7 +66,7 @@ source("test_input");
6566
testCase := testCase
6667
t.Run(name, func(t *testing.T) {
6768
out := strings.Builder{}
68-
require.NoError(t, renderClusterFlow("test_input", testCase.clusterFlow, nil)(render.RenderContext{
69+
require.NoError(t, renderClusterFlow(nil, "test_input", testCase.clusterFlow, nil)(render.RenderContext{
6970
Out: &out,
7071
}))
7172
assert.Equal(t, testCase.expected, out.String())

‎pkg/sdk/logging/model/syslogng/config/go.mod‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ module github.com/kube-logging/logging-operator/pkg/sdk/logging/model/syslogng/c
33
go 1.20
44

55
require (
6+
emperror.dev/errors v0.8.1
67
github.com/cisco-open/operator-tools v0.30.0
78
github.com/kube-logging/logging-operator/pkg/sdk v0.9.1
89
github.com/siliconbrain/go-seqs v0.5.0
@@ -14,7 +15,6 @@ require (
1415
)
1516

1617
require (
17-
emperror.dev/errors v0.8.1 // indirect
1818
github.com/beorn7/perks v1.0.1 // indirect
1919
github.com/cespare/xxhash/v2 v2.2.0 // indirect
2020
github.com/davecgh/go-spew v1.1.1 // indirect

0 commit comments

Comments
 (0)