Skip to content

Commit 497fed7

Browse files
li-zeyuanandrzej-stencel
li-zeyuan
and
andrzej-stencel
authored
[remotetapprocessor] use 'time/rate' to limit traffic (#32481)
bug: The remotetapprocessor `limit` configure doesn't work. how to fix: use `time/rate` to limit traffic. Resolves #32385 --------- Co-authored-by: Andrzej Stencel <[email protected]>
1 parent f4a3147 commit 497fed7

File tree

4 files changed

+223
-15
lines changed

4 files changed

+223
-15
lines changed

‎.chloggen/fix-remotetap-limit.yaml‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,30 @@
1+
# Use this changelog template to create an entry for release notes.
2+
3+
# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
4+
change_type: breaking
5+
6+
# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
7+
component: remotetapprocessor
8+
9+
# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`).
10+
note: Make the `limit` configuration work properly.
11+
12+
# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
13+
issues: [32385]
14+
15+
# (Optional) One or more lines of additional information to render under the primary note.
16+
# These lines will be padded with 2 spaces and then inserted directly into the document.
17+
# Use pipe (|) for multiline entries.
18+
subtext: |
19+
The `limit` configuration was ignored previously, but now it works according to the configuration and documentation.
20+
Nothing is required of users.
21+
See the remotetapprocessor's `README.md` for details.
22+
23+
# If your change doesn't affect end users or the exported elements of any package,
24+
# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
25+
# Optional: The change log or logs in which this entry should be included.
26+
# e.g. '[user]' or '[user, api]'
27+
# Include 'user' if the change is relevant to end users.
28+
# Include 'api' if there is a change to a library API.
29+
# Default: '[user]'
30+
change_logs: []

‎processor/remotetapprocessor/processor.go‎

Lines changed: 27 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ import (
1919
"go.opentelemetry.io/collector/processor"
2020
"go.uber.org/zap"
2121
"golang.org/x/net/websocket"
22+
"golang.org/x/time/rate"
2223
)
2324

2425
type wsprocessor struct {
@@ -27,6 +28,7 @@ type wsprocessor struct {
2728
server *http.Server
2829
shutdownWG sync.WaitGroup
2930
cs *channelSet
31+
limiter *rate.Limiter
3032
}
3133

3234
var logMarshaler = &plog.JSONMarshaler{}
@@ -38,6 +40,7 @@ func newProcessor(settings processor.CreateSettings, config *Config) *wsprocesso
3840
config: config,
3941
telemetrySettings: settings.TelemetrySettings,
4042
cs: newChannelSet(),
43+
limiter: rate.NewLimiter(config.Limit, int(config.Limit)),
4144
}
4245
}
4346

@@ -98,31 +101,40 @@ func (w *wsprocessor) Shutdown(ctx context.Context) error {
98101
}
99102

100103
func (w *wsprocessor) ConsumeMetrics(_ context.Context, md pmetric.Metrics) (pmetric.Metrics, error) {
101-
b, err := metricMarshaler.MarshalMetrics(md)
102-
if err != nil {
103-
w.telemetrySettings.Logger.Debug("Error serializing to JSON", zap.Error(err))
104-
} else {
105-
w.cs.writeBytes(b)
104+
if w.limiter.Allow() {
105+
b, err := metricMarshaler.MarshalMetrics(md)
106+
if err != nil {
107+
w.telemetrySettings.Logger.Debug("Error serializing to JSON", zap.Error(err))
108+
} else {
109+
w.cs.writeBytes(b)
110+
}
106111
}
112+
107113
return md, nil
108114
}
109115

110116
func (w *wsprocessor) ConsumeLogs(_ context.Context, ld plog.Logs) (plog.Logs, error) {
111-
b, err := logMarshaler.MarshalLogs(ld)
112-
if err != nil {
113-
w.telemetrySettings.Logger.Debug("Error serializing to JSON", zap.Error(err))
114-
} else {
115-
w.cs.writeBytes(b)
117+
if w.limiter.Allow() {
118+
b, err := logMarshaler.MarshalLogs(ld)
119+
if err != nil {
120+
w.telemetrySettings.Logger.Debug("Error serializing to JSON", zap.Error(err))
121+
} else {
122+
w.cs.writeBytes(b)
123+
}
116124
}
125+
117126
return ld, nil
118127
}
119128

120129
func (w *wsprocessor) ConsumeTraces(_ context.Context, td ptrace.Traces) (ptrace.Traces, error) {
121-
b, err := traceMarshaler.MarshalTraces(td)
122-
if err != nil {
123-
w.telemetrySettings.Logger.Debug("Error serializing to JSON", zap.Error(err))
124-
} else {
125-
w.cs.writeBytes(b)
130+
if w.limiter.Allow() {
131+
b, err := traceMarshaler.MarshalTraces(td)
132+
if err != nil {
133+
w.telemetrySettings.Logger.Debug("Error serializing to JSON", zap.Error(err))
134+
} else {
135+
w.cs.writeBytes(b)
136+
}
126137
}
138+
127139
return td, nil
128140
}

‎processor/remotetapprocessor/processor_test.go‎

Lines changed: 163 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,163 @@
1+
// Copyright The OpenTelemetry Authors
2+
// SPDX-License-Identifier: Apache-2.0
3+
4+
package remotetapprocessor
5+
6+
import (
7+
"context"
8+
"sync"
9+
"testing"
10+
11+
"github.com/stretchr/testify/assert"
12+
"go.opentelemetry.io/collector/pdata/plog"
13+
"go.opentelemetry.io/collector/pdata/pmetric"
14+
"go.opentelemetry.io/collector/pdata/ptrace"
15+
"go.opentelemetry.io/collector/processor/processortest"
16+
"golang.org/x/time/rate"
17+
)
18+
19+
func TestConsumeMetrics(t *testing.T) {
20+
metric := pmetric.NewMetrics()
21+
metric.ResourceMetrics().AppendEmpty().ScopeMetrics().AppendEmpty().Metrics().AppendEmpty().SetName("foo")
22+
23+
cases := []struct {
24+
name string
25+
limit int
26+
}{
27+
{name: "limit_0", limit: 0},
28+
{name: "limit_1", limit: 1},
29+
{name: "limit_10", limit: 10},
30+
{name: "limit_50", limit: 50},
31+
}
32+
33+
for _, c := range cases {
34+
t.Run(c.name, func(t *testing.T) {
35+
conf := &Config{
36+
Limit: rate.Limit(c.limit),
37+
}
38+
39+
processor := newProcessor(processortest.NewNopCreateSettings(), conf)
40+
41+
ch := make(chan []byte)
42+
idx := processor.cs.add(ch)
43+
receiveNum := 0
44+
wg := &sync.WaitGroup{}
45+
wg.Add(1)
46+
go func() {
47+
defer wg.Done()
48+
for range ch {
49+
receiveNum++
50+
}
51+
}()
52+
53+
for i := 0; i < c.limit*2; i++ {
54+
// send metric to chan c.limit*2 per sec.
55+
metric2, err := processor.ConsumeMetrics(context.Background(), metric)
56+
assert.Nil(t, err)
57+
assert.Equal(t, metric, metric2)
58+
}
59+
60+
processor.cs.closeAndRemove(idx)
61+
wg.Wait()
62+
assert.Equal(t, receiveNum, c.limit)
63+
64+
})
65+
}
66+
}
67+
68+
func TestConsumeLogs(t *testing.T) {
69+
log := plog.NewLogs()
70+
log.ResourceLogs().AppendEmpty().ScopeLogs().AppendEmpty().LogRecords().AppendEmpty().Body().SetStr("foo")
71+
72+
cases := []struct {
73+
name string
74+
limit int
75+
}{
76+
{name: "limit_0", limit: 0},
77+
{name: "limit_1", limit: 1},
78+
{name: "limit_10", limit: 10},
79+
{name: "limit_50", limit: 50},
80+
}
81+
82+
for _, c := range cases {
83+
t.Run(c.name, func(t *testing.T) {
84+
conf := &Config{
85+
Limit: rate.Limit(c.limit),
86+
}
87+
88+
processor := newProcessor(processortest.NewNopCreateSettings(), conf)
89+
90+
ch := make(chan []byte)
91+
idx := processor.cs.add(ch)
92+
receiveNum := 0
93+
wg := &sync.WaitGroup{}
94+
wg.Add(1)
95+
go func() {
96+
defer wg.Done()
97+
for range ch {
98+
receiveNum++
99+
}
100+
}()
101+
102+
// send log to chan c.limit*2 per sec.
103+
for i := 0; i < c.limit*2; i++ {
104+
log2, err := processor.ConsumeLogs(context.Background(), log)
105+
assert.Nil(t, err)
106+
assert.Equal(t, log, log2)
107+
}
108+
109+
processor.cs.closeAndRemove(idx)
110+
wg.Wait()
111+
t.Log(receiveNum)
112+
assert.Equal(t, receiveNum, c.limit)
113+
})
114+
}
115+
}
116+
117+
func TestConsumeTraces(t *testing.T) {
118+
trace := ptrace.NewTraces()
119+
trace.ResourceSpans().AppendEmpty().ScopeSpans().AppendEmpty().Spans().AppendEmpty().SetName("foo")
120+
121+
cases := []struct {
122+
name string
123+
limit int
124+
}{
125+
{name: "limit_0", limit: 0},
126+
{name: "limit_1", limit: 1},
127+
{name: "limit_10", limit: 10},
128+
{name: "limit_50", limit: 50},
129+
}
130+
131+
for _, c := range cases {
132+
t.Run(c.name, func(t *testing.T) {
133+
conf := &Config{
134+
Limit: rate.Limit(c.limit),
135+
}
136+
137+
processor := newProcessor(processortest.NewNopCreateSettings(), conf)
138+
139+
ch := make(chan []byte)
140+
idx := processor.cs.add(ch)
141+
receiveNum := 0
142+
wg := &sync.WaitGroup{}
143+
wg.Add(1)
144+
go func() {
145+
defer wg.Done()
146+
for range ch {
147+
receiveNum++
148+
}
149+
}()
150+
151+
for i := 0; i < c.limit*2; i++ {
152+
// send trace to chan c.limit*2 per sec.
153+
trace2, err := processor.ConsumeTraces(context.Background(), trace)
154+
assert.Nil(t, err)
155+
assert.Equal(t, trace, trace2)
156+
}
157+
158+
processor.cs.closeAndRemove(idx)
159+
wg.Wait()
160+
assert.Equal(t, receiveNum, c.limit)
161+
})
162+
}
163+
}

‎processor/remotetapprocessor/server_test.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@ func TestSocketConnectionLogs(t *testing.T) {
2525
ServerConfig: confighttp.ServerConfig{
2626
Endpoint: "localhost:12001",
2727
},
28+
Limit: 1,
2829
}
2930
logSink := &consumertest.LogsSink{}
3031
processor, err := NewFactory().CreateLogsProcessor(context.Background(), processortest.NewNopCreateSettings(), cfg,
@@ -62,6 +63,7 @@ func TestSocketConnectionMetrics(t *testing.T) {
6263
ServerConfig: confighttp.ServerConfig{
6364
Endpoint: "localhost:12002",
6465
},
66+
Limit: 1,
6567
}
6668
metricsSink := &consumertest.MetricsSink{}
6769
processor, err := NewFactory().CreateMetricsProcessor(context.Background(), processortest.NewNopCreateSettings(), cfg,
@@ -97,6 +99,7 @@ func TestSocketConnectionTraces(t *testing.T) {
9799
ServerConfig: confighttp.ServerConfig{
98100
Endpoint: "localhost:12003",
99101
},
102+
Limit: 1,
100103
}
101104
tracesSink := &consumertest.TracesSink{}
102105
processor, err := NewFactory().CreateTracesProcessor(context.Background(), processortest.NewNopCreateSettings(), cfg,

0 commit comments

Comments
 (0)