Skip to content

Commit 9ff9615

Browse files
cbandycbandy
committed
Allow tuning of logs batching behavior
Batches are an important component to shipping logs efficiently. Issue: PGO-2176
1 parent 22f7989 commit 9ff9615

20 files changed

+468
-60
lines changed

‎config/crd/bases/postgres-operator.crunchydata.com_pgadmins.yaml‎

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1948,6 +1948,46 @@ spec:
19481948
description: Logs is the place for users to configure the log
19491949
collection.
19501950
properties:
1951+
batches:
1952+
description: Log records are exported in small batches. Set
1953+
this field to change their size and frequency.
1954+
properties:
1955+
maxDelay:
1956+
default: 200ms
1957+
description: |-
1958+
Maximum time to wait before exporting a log record. Higher numbers
1959+
allow more records to be deduplicated and compressed before export.
1960+
format: duration
1961+
maxLength: 20
1962+
minLength: 1
1963+
pattern: ^((PT)?( *[0-9]+ *(?i:(ms|s|m)|(milli|sec|min)s?))+|0)$
1964+
type: string
1965+
x-kubernetes-validations:
1966+
- rule: duration("0") <= self && self <= duration("5m")
1967+
maxRecords:
1968+
description: |-
1969+
Maximum number of records to include in an exported batch. When present,
1970+
batches this size are sent without any further delay.
1971+
format: int32
1972+
minimum: 1
1973+
type: integer
1974+
minRecords:
1975+
default: 8192
1976+
description: |-
1977+
Number of records to wait for before exporting a batch. Higher numbers
1978+
allow more records to be deduplicated and compressed before export.
1979+
format: int32
1980+
minimum: 0
1981+
type: integer
1982+
type: object
1983+
x-kubernetes-map-type: atomic
1984+
x-kubernetes-validations:
1985+
- message: to disable batching, both minRecords and maxDelay
1986+
must be zero
1987+
rule: (has(self.minRecords) && self.minRecords == 0) ==
1988+
(has(self.maxDelay) && self.maxDelay == duration('0'))
1989+
- message: minRecords cannot be larger than maxRecords
1990+
rule: '!has(self.maxRecords) || self.minRecords <= self.maxRecords'
19511991
exporters:
19521992
description: |-
19531993
Exporters allows users to specify which exporters they want to use in

‎config/crd/bases/postgres-operator.crunchydata.com_postgresclusters.yaml‎

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11524,6 +11524,46 @@ spec:
1152411524
description: Logs is the place for users to configure the log
1152511525
collection.
1152611526
properties:
11527+
batches:
11528+
description: Log records are exported in small batches. Set
11529+
this field to change their size and frequency.
11530+
properties:
11531+
maxDelay:
11532+
default: 200ms
11533+
description: |-
11534+
Maximum time to wait before exporting a log record. Higher numbers
11535+
allow more records to be deduplicated and compressed before export.
11536+
format: duration
11537+
maxLength: 20
11538+
minLength: 1
11539+
pattern: ^((PT)?( *[0-9]+ *(?i:(ms|s|m)|(milli|sec|min)s?))+|0)$
11540+
type: string
11541+
x-kubernetes-validations:
11542+
- rule: duration("0") <= self && self <= duration("5m")
11543+
maxRecords:
11544+
description: |-
11545+
Maximum number of records to include in an exported batch. When present,
11546+
batches this size are sent without any further delay.
11547+
format: int32
11548+
minimum: 1
11549+
type: integer
11550+
minRecords:
11551+
default: 8192
11552+
description: |-
11553+
Number of records to wait for before exporting a batch. Higher numbers
11554+
allow more records to be deduplicated and compressed before export.
11555+
format: int32
11556+
minimum: 0
11557+
type: integer
11558+
type: object
11559+
x-kubernetes-map-type: atomic
11560+
x-kubernetes-validations:
11561+
- message: to disable batching, both minRecords and maxDelay
11562+
must be zero
11563+
rule: (has(self.minRecords) && self.minRecords == 0) ==
11564+
(has(self.maxDelay) && self.maxDelay == duration('0'))
11565+
- message: minRecords cannot be larger than maxRecords
11566+
rule: '!has(self.maxRecords) || self.minRecords <= self.maxRecords'
1152711567
exporters:
1152811568
description: |-
1152911569
Exporters allows users to specify which exporters they want to use in

‎internal/collector/config.go‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -114,6 +114,29 @@ func NewConfig(spec *v1beta1.InstrumentationSpec) *Config {
114114
Pipelines: map[PipelineID]Pipeline{},
115115
}
116116

117+
// Configure a batch processor for logs according to the API spec.
118+
// Use API defaults for any unspecified fields.
119+
{
120+
var batches v1beta1.OpenTelemetryLogsBatchSpec
121+
if spec != nil && spec.Logs != nil && spec.Logs.Batches != nil {
122+
spec.Logs.Batches.DeepCopyInto(&batches)
123+
}
124+
batches.Default()
125+
126+
// https://pkg.go.dev/go.opentelemetry.io/collector/processor/batchprocessor#section-readme
127+
processor := map[string]any{}
128+
if batches.MaxDelay != nil {
129+
processor["timeout"] = batches.MaxDelay.AsDuration().Duration.String()
130+
}
131+
if batches.MaxRecords != nil {
132+
processor["send_batch_max_size"] = *batches.MaxRecords
133+
}
134+
if batches.MinRecords != nil {
135+
processor["send_batch_size"] = *batches.MinRecords
136+
}
137+
config.Processors[LogsBatchProcessor] = processor
138+
}
139+
117140
// If there are exporters defined in the spec, add them to the config.
118141
if spec != nil && spec.Config != nil && spec.Config.Exporters != nil {
119142
for k, v := range spec.Config.Exporters {

‎internal/collector/config_test.go‎

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,8 @@ import (
99

1010
"gotest.tools/v3/assert"
1111

12+
"github.com/crunchydata/postgres-operator/internal/testing/cmp"
13+
"github.com/crunchydata/postgres-operator/internal/testing/require"
1214
"github.com/crunchydata/postgres-operator/pkg/apis/postgres-operator.crunchydata.com/v1beta1"
1315
)
1416

@@ -27,6 +29,9 @@ processors:
2729
timeout: 1s
2830
batch/200ms:
2931
timeout: 200ms
32+
batch/logs:
33+
send_batch_size: 8192
34+
timeout: 200ms
3035
groupbyattrs/compact: {}
3136
receivers: {}
3237
service:
@@ -55,13 +60,55 @@ processors:
5560
timeout: 1s
5661
batch/200ms:
5762
timeout: 200ms
63+
batch/logs:
64+
send_batch_size: 8192
65+
timeout: 200ms
5866
groupbyattrs/compact: {}
5967
receivers: {}
6068
service:
6169
extensions: []
6270
pipelines: {}
6371
`)
6472
})
73+
74+
t.Run("LogsBatches", func(t *testing.T) {
75+
var spec *v1beta1.InstrumentationSpec
76+
require.UnmarshalInto(t, &spec, `{
77+
logs: {
78+
batches: {
79+
maxDelay: 5min 12sec,
80+
maxRecords: 123,
81+
minRecords: 45,
82+
},
83+
},
84+
}`)
85+
86+
result, err := NewConfig(spec).ToYAML()
87+
assert.NilError(t, err)
88+
assert.Assert(t, cmp.Contains(result, `
89+
batch/logs:
90+
send_batch_max_size: 123
91+
send_batch_size: 45
92+
timeout: 5m12s
93+
`))
94+
95+
t.Run("Disable", func(t *testing.T) {
96+
var spec *v1beta1.InstrumentationSpec
97+
require.UnmarshalInto(t, &spec, `{
98+
logs: {
99+
batches: { minRecords: 0, maxDelay: "0" },
100+
},
101+
}`)
102+
103+
result, err := NewConfig(spec).ToYAML()
104+
assert.NilError(t, err)
105+
assert.Assert(t, cmp.Contains(result, `
106+
batch/logs:
107+
send_batch_size: 0
108+
timeout: 0s
109+
`))
110+
})
111+
})
65112
}
66113

67114
func TestGenerateLogrotateConfig(t *testing.T) {

‎internal/collector/naming.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ package collector
66

77
const CompactingProcessor = "groupbyattrs/compact"
88
const DebugExporter = "debug"
9+
const LogsBatchProcessor = "batch/logs"
910
const OneSecondBatchProcessor = "batch/1s"
1011
const SubSecondBatchProcessor = "batch/200ms"
1112
const Prometheus = "prometheus"

‎internal/collector/patroni.go‎

Lines changed: 10 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ package collector
66

77
import (
88
"context"
9+
"slices"
910

1011
"github.com/crunchydata/postgres-operator/internal/feature"
1112
"github.com/crunchydata/postgres-operator/internal/naming"
@@ -16,6 +17,11 @@ func EnablePatroniLogging(ctx context.Context,
1617
inCluster *v1beta1.PostgresCluster,
1718
outConfig *Config,
1819
) {
20+
var spec *v1beta1.InstrumentationLogsSpec
21+
if inCluster != nil && inCluster.Spec.Instrumentation != nil {
22+
spec = inCluster.Spec.Instrumentation.Logs
23+
}
24+
1925
if feature.Enabled(ctx, feature.OpenTelemetryLogs) {
2026
directory := naming.PatroniPGDataLogPath
2127

@@ -103,13 +109,9 @@ func EnablePatroniLogging(ctx context.Context,
103109

104110
// If there are exporters to be added to the logs pipelines defined in
105111
// the spec, add them to the pipeline. Otherwise, add the DebugExporter.
106-
var exporters []ComponentID
107-
if inCluster.Spec.Instrumentation != nil &&
108-
inCluster.Spec.Instrumentation.Logs != nil &&
109-
inCluster.Spec.Instrumentation.Logs.Exporters != nil {
110-
exporters = inCluster.Spec.Instrumentation.Logs.Exporters
111-
} else {
112-
exporters = []ComponentID{DebugExporter}
112+
exporters := []ComponentID{DebugExporter}
113+
if spec != nil && spec.Exporters != nil {
114+
exporters = slices.Clone(spec.Exporters)
113115
}
114116

115117
outConfig.Pipelines["logs/patroni"] = Pipeline{
@@ -118,7 +120,7 @@ func EnablePatroniLogging(ctx context.Context,
118120
Processors: []ComponentID{
119121
"resource/patroni",
120122
"transform/patroni_logs",
121-
SubSecondBatchProcessor,
123+
LogsBatchProcessor,
122124
CompactingProcessor,
123125
},
124126
Exporters: exporters,

‎internal/collector/patroni_test.go‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,9 @@ processors:
4343
timeout: 1s
4444
batch/200ms:
4545
timeout: 200ms
46+
batch/logs:
47+
send_batch_size: 8192
48+
timeout: 200ms
4649
groupbyattrs/compact: {}
4750
resource/patroni:
4851
attributes:
@@ -89,7 +92,7 @@ service:
8992
processors:
9093
- resource/patroni
9194
- transform/patroni_logs
92-
- batch/200ms
95+
- batch/logs
9396
- groupbyattrs/compact
9497
receivers:
9598
- filelog/patroni_jsonlog
@@ -130,6 +133,9 @@ processors:
130133
timeout: 1s
131134
batch/200ms:
132135
timeout: 200ms
136+
batch/logs:
137+
send_batch_size: 8192
138+
timeout: 200ms
133139
groupbyattrs/compact: {}
134140
resource/patroni:
135141
attributes:
@@ -176,7 +182,7 @@ service:
176182
processors:
177183
- resource/patroni
178184
- transform/patroni_logs
179-
- batch/200ms
185+
- batch/logs
180186
- groupbyattrs/compact
181187
receivers:
182188
- filelog/patroni_jsonlog

‎internal/collector/pgadmin.go‎

Lines changed: 5 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ package collector
66

77
import (
88
"context"
9+
"slices"
910

1011
corev1 "k8s.io/api/core/v1"
1112

@@ -89,11 +90,9 @@ func EnablePgAdminLogging(ctx context.Context, spec *v1beta1.InstrumentationSpec
8990

9091
// If there are exporters to be added to the logs pipelines defined in
9192
// the spec, add them to the pipeline. Otherwise, add the DebugExporter.
92-
var exporters []ComponentID
93+
exporters := []ComponentID{DebugExporter}
9394
if spec != nil && spec.Logs != nil && spec.Logs.Exporters != nil {
94-
exporters = spec.Logs.Exporters
95-
} else {
96-
exporters = []ComponentID{DebugExporter}
95+
exporters = slices.Clone(spec.Logs.Exporters)
9796
}
9897

9998
otelConfig.Pipelines["logs/pgadmin"] = Pipeline{
@@ -102,7 +101,7 @@ func EnablePgAdminLogging(ctx context.Context, spec *v1beta1.InstrumentationSpec
102101
Processors: []ComponentID{
103102
"resource/pgadmin",
104103
"transform/pgadmin_log",
105-
SubSecondBatchProcessor,
104+
LogsBatchProcessor,
106105
CompactingProcessor,
107106
},
108107
Exporters: exporters,
@@ -114,7 +113,7 @@ func EnablePgAdminLogging(ctx context.Context, spec *v1beta1.InstrumentationSpec
114113
Processors: []ComponentID{
115114
"resource/pgadmin",
116115
"transform/pgadmin_log",
117-
SubSecondBatchProcessor,
116+
LogsBatchProcessor,
118117
CompactingProcessor,
119118
},
120119
Exporters: exporters,

‎internal/collector/pgadmin_test.go‎

Lines changed: 10 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,9 @@ collector.yaml: |
5151
timeout: 1s
5252
batch/200ms:
5353
timeout: 200ms
54+
batch/logs:
55+
send_batch_size: 8192
56+
timeout: 200ms
5457
groupbyattrs/compact: {}
5558
resource/pgadmin:
5659
attributes:
@@ -99,7 +102,7 @@ collector.yaml: |
99102
processors:
100103
- resource/pgadmin
101104
- transform/pgadmin_log
102-
- batch/200ms
105+
- batch/logs
103106
- groupbyattrs/compact
104107
receivers:
105108
- filelog/gunicorn
@@ -109,7 +112,7 @@ collector.yaml: |
109112
processors:
110113
- resource/pgadmin
111114
- transform/pgadmin_log
112-
- batch/200ms
115+
- batch/logs
113116
- groupbyattrs/compact
114117
receivers:
115118
- filelog/pgadmin
@@ -163,6 +166,9 @@ collector.yaml: |
163166
timeout: 1s
164167
batch/200ms:
165168
timeout: 200ms
169+
batch/logs:
170+
send_batch_size: 8192
171+
timeout: 200ms
166172
groupbyattrs/compact: {}
167173
resource/pgadmin:
168174
attributes:
@@ -211,7 +217,7 @@ collector.yaml: |
211217
processors:
212218
- resource/pgadmin
213219
- transform/pgadmin_log
214-
- batch/200ms
220+
- batch/logs
215221
- groupbyattrs/compact
216222
receivers:
217223
- filelog/gunicorn
@@ -221,7 +227,7 @@ collector.yaml: |
221227
processors:
222228
- resource/pgadmin
223229
- transform/pgadmin_log
224-
- batch/200ms
230+
- batch/logs
225231
- groupbyattrs/compact
226232
receivers:
227233
- filelog/pgadmin

0 commit comments

Comments
 (0)