[remotetapprocessor] use 'time/rate' to limit traffic

.chloggen/fix-remotetap-limit.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: remotetapprocessor
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Use `time/rate` to limit traffic.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [32385]
+
+# (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: []

processor/remotetapprocessor/README.md

@@ -24,8 +24,8 @@ any open WebSockets is rate limited by an adjustable amount.
 
 The WebSocket processor has two configurable fields: `port` and `limit`:
 
-- `port`: The port on which the WebSocket processor listens. Optional. Defaults
-  to `12001`.
+- `endpoint`: The endpoint on which the WebSocket processor listens. Optional. Defaults
+  to `0.0.0.0:12001`.
   The `component.UseLocalHostAsDefaultHost` feature gate changes this to localhost:12001. This will become the default in a future release.
 
 - `limit`: The rate limit over the WebSocket in messages per second. Can be a
@@ -35,6 +35,6 @@ Example configuration:
 
 ```yaml
 websocket:
-  port: 12001
+  endpoint: 0.0.0.0:12001
   limit: 1 # rate limit 1 msg/sec
 ```

processor/remotetapprocessor/processor.go

@@ -12,13 +12,15 @@ import (
 	"sync"
 	"time"
 
+	"go.uber.org/zap"
+	"golang.org/x/net/websocket"
+	"golang.org/x/time/rate"
+
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.opentelemetry.io/collector/pdata/ptrace"
 	"go.opentelemetry.io/collector/processor"
-	"go.uber.org/zap"
-	"golang.org/x/net/websocket"
 )
 
 type wsprocessor struct {
@@ -27,6 +29,7 @@ type wsprocessor struct {
 	server            *http.Server
 	shutdownWG        sync.WaitGroup
 	cs                *channelSet
+	limiter           *rate.Limiter
 }
 
 var logMarshaler = &plog.JSONMarshaler{}
@@ -38,6 +41,7 @@ func newProcessor(settings processor.CreateSettings, config *Config) *wsprocesso
 		config:            config,
 		telemetrySettings: settings.TelemetrySettings,
 		cs:                newChannelSet(),
+		limiter:           rate.NewLimiter(config.Limit, int(config.Limit)),
 	}
 }
 
@@ -89,31 +93,40 @@ func (w *wsprocessor) Shutdown(ctx context.Context) error {
 }
 
 func (w *wsprocessor) ConsumeMetrics(_ context.Context, md pmetric.Metrics) (pmetric.Metrics, error) {
-	b, err := metricMarshaler.MarshalMetrics(md)
-	if err != nil {
-		w.telemetrySettings.Logger.Debug("Error serializing to JSON", zap.Error(err))
-	} else {
-		w.cs.writeBytes(b)
+	if w.limiter.Allow() {
+		b, err := metricMarshaler.MarshalMetrics(md)
+		if err != nil {
+			w.telemetrySettings.Logger.Debug("Error serializing to JSON", zap.Error(err))
+		} else {
+			w.cs.writeBytes(b)
+		}
 	}
+
 	return md, nil
 }
 
 func (w *wsprocessor) ConsumeLogs(_ context.Context, ld plog.Logs) (plog.Logs, error) {
-	b, err := logMarshaler.MarshalLogs(ld)
-	if err != nil {
-		w.telemetrySettings.Logger.Debug("Error serializing to JSON", zap.Error(err))
-	} else {
-		w.cs.writeBytes(b)
+	if w.limiter.Allow() {
+		b, err := logMarshaler.MarshalLogs(ld)
+		if err != nil {
+			w.telemetrySettings.Logger.Debug("Error serializing to JSON", zap.Error(err))
+		} else {
+			w.cs.writeBytes(b)
+		}
 	}
+
 	return ld, nil
 }
 
 func (w *wsprocessor) ConsumeTraces(_ context.Context, td ptrace.Traces) (ptrace.Traces, error) {
-	b, err := traceMarshaler.MarshalTraces(td)
-	if err != nil {
-		w.telemetrySettings.Logger.Debug("Error serializing to JSON", zap.Error(err))
-	} else {
-		w.cs.writeBytes(b)
+	if w.limiter.Allow() {
+		b, err := traceMarshaler.MarshalTraces(td)
+		if err != nil {
+			w.telemetrySettings.Logger.Debug("Error serializing to JSON", zap.Error(err))
+		} else {
+			w.cs.writeBytes(b)
+		}
 	}
+
 	return td, nil
 }

processor/remotetapprocessor/server_test.go

@@ -25,6 +25,7 @@ func TestSocketConnectionLogs(t *testing.T) {
 		ServerConfig: confighttp.ServerConfig{
 			Endpoint: "localhost:12001",
 		},
+		Limit: 1,
 	}
 	logSink := &consumertest.LogsSink{}
 	processor, err := NewFactory().CreateLogsProcessor(context.Background(), processortest.NewNopCreateSettings(), cfg,
@@ -62,6 +63,7 @@ func TestSocketConnectionMetrics(t *testing.T) {
 		ServerConfig: confighttp.ServerConfig{
 			Endpoint: "localhost:12002",
 		},
+		Limit: 1,
 	}
 	metricsSink := &consumertest.MetricsSink{}
 	processor, err := NewFactory().CreateMetricsProcessor(context.Background(), processortest.NewNopCreateSettings(), cfg,
@@ -97,6 +99,7 @@ func TestSocketConnectionTraces(t *testing.T) {
 		ServerConfig: confighttp.ServerConfig{
 			Endpoint: "localhost:12003",
 		},
+		Limit: 1,
 	}
 	tracesSink := &consumertest.TracesSink{}
 	processor, err := NewFactory().CreateTracesProcessor(context.Background(), processortest.NewNopCreateSettings(), cfg,