diff --git a/.chloggen/fix-testbed-loadgen-counters.yaml b/.chloggen/fix-testbed-loadgen-counters.yaml new file mode 100644 index 000000000000..e73e3377baff --- /dev/null +++ b/.chloggen/fix-testbed-loadgen-counters.yaml @@ -0,0 +1,27 @@ +# Use this changelog template to create an entry for release notes. + +# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix' +change_type: bug_fix + +# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver) +component: testbed + +# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`). +note: Fixes incorrect count for sent data items in load generator. + +# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists. +issues: [34057] + +# (Optional) One or more lines of additional information to render under the primary note. +# These lines will be padded with 2 spaces and then inserted directly into the document. +# Use pipe (|) for multiline entries. +subtext: + +# If your change doesn't affect end users or the exported elements of any package, +# you should instead start your pull request title with [chore] or use the "Skip Changelog" label. +# Optional: The change log or logs in which this entry should be included. +# e.g. '[user]' or '[user, api]' +# Include 'user' if the change is relevant to end users. +# Include 'api' if there is a change to a library API. +# Default: '[user]' +change_logs: [user] diff --git a/testbed/testbed/load_generator.go b/testbed/testbed/load_generator.go index b6c5deae090d..79820a254276 100644 --- a/testbed/testbed/load_generator.go +++ b/testbed/testbed/load_generator.go @@ -245,23 +245,25 @@ func (ps *ProviderSender) generateTrace() error { } for { - select { - case <-ps.stopSignal: + // Generated data MUST be consumed once since the data counters + // are updated by the provider and not consuming the generated + // data will lead to accounting errors. + err := traceSender.ConsumeTraces(context.Background(), traceData) + if err == nil { return nil - default: - err := traceSender.ConsumeTraces(context.Background(), traceData) - if err == nil { - return nil - } - - if !consumererror.IsPermanent(err) { - ps.nonPermanentErrors.Add(uint64(traceData.SpanCount())) - continue - } + } + if consumererror.IsPermanent(err) { ps.permanentErrors.Add(uint64(traceData.SpanCount())) return fmt.Errorf("cannot send traces: %w", err) } + ps.nonPermanentErrors.Add(uint64(traceData.SpanCount())) + + select { + case <-ps.stopSignal: + return nil + default: + } } } @@ -274,23 +276,25 @@ func (ps *ProviderSender) generateMetrics() error { } for { - select { - case <-ps.stopSignal: + // Generated data MUST be consumed once since the data counters + // are updated by the provider and not consuming the generated + // data will lead to accounting errors. + err := metricSender.ConsumeMetrics(context.Background(), metricData) + if err == nil { return nil - default: - err := metricSender.ConsumeMetrics(context.Background(), metricData) - if err == nil { - return nil - } - - if !consumererror.IsPermanent(err) { - ps.nonPermanentErrors.Add(uint64(metricData.DataPointCount())) - continue - } + } + if consumererror.IsPermanent(err) { ps.permanentErrors.Add(uint64(metricData.DataPointCount())) return fmt.Errorf("cannot send metrics: %w", err) } + ps.nonPermanentErrors.Add(uint64(metricData.DataPointCount())) + + select { + case <-ps.stopSignal: + return nil + default: + } } } @@ -303,22 +307,24 @@ func (ps *ProviderSender) generateLog() error { } for { - select { - case <-ps.stopSignal: + // Generated data MUST be consumed once since the data counters + // are updated by the provider and not consuming the generated + // data will lead to accounting errors. + err := logSender.ConsumeLogs(context.Background(), logData) + if err == nil { return nil - default: - err := logSender.ConsumeLogs(context.Background(), logData) - if err == nil { - return nil - } - - if !consumererror.IsPermanent(err) { - ps.nonPermanentErrors.Add(uint64(logData.LogRecordCount())) - continue - } + } + if consumererror.IsPermanent(err) { ps.permanentErrors.Add(uint64(logData.LogRecordCount())) return fmt.Errorf("cannot send logs: %w", err) } + ps.nonPermanentErrors.Add(uint64(logData.LogRecordCount())) + + select { + case <-ps.stopSignal: + return nil + default: + } } }