[processor/websocketprocessor] add skeleton and initial README

open-telemetry · dmitryax · May 16, 2023 · Feb 23, 2023 · Apr 12, 2023 · Apr 12, 2023
commit 7507c5cb85b169757c048af614a1c73e0ade0694
@@ -0,0 +1,16 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: new_component
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: WebSocket processor
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add WebSocket processor skeleton
+
+# One or more tracking issues related to the change
+issues: [19633]
+
+# (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:
@@ -155,6 +155,7 @@ processor/spanmetricsprocessor/                          @open-telemetry/collect
 processor/spanprocessor/                                 @open-telemetry/collector-contrib-approvers @boostchicken
 processor/tailsamplingprocessor/                         @open-telemetry/collector-contrib-approvers @jpkrohling
 processor/transformprocessor/                            @open-telemetry/collector-contrib-approvers @TylerHelmuth @kentquirk @bogdandrutu @evan-bradley
+processor/websocketprocessor/                            @open-telemetry/collector-contrib-approvers @pmcollins
 
 receiver/azureeventhubreceiver/                          @open-telemetry/collector-contrib-approvers @atoulme @djaglowski
 receiver/activedirectorydsreceiver/                      @open-telemetry/collector-contrib-approvers @djaglowski @binaryfissiongames

@@ -145,6 +145,7 @@ body:
       - processor/spanmetrics
       - processor/tailsampling
       - processor/transform
+      - processor/websocket
       - receiver/activedirectoryds
       - receiver/aerospike
       - receiver/apache

@@ -139,6 +139,7 @@ body:
       - processor/spanmetrics
       - processor/tailsampling
       - processor/transform
+      - processor/websocket
       - receiver/activedirectoryds
       - receiver/aerospike
       - receiver/apache

@@ -139,6 +139,7 @@ body:
       - processor/spanmetrics
       - processor/tailsampling
       - processor/transform
+      - processor/websocket
       - receiver/activedirectoryds
       - receiver/aerospike
       - receiver/apache

@@ -677,6 +677,11 @@ updates:
     schedule:
       interval: "weekly"
       day: "wednesday"
+  - package-ecosystem: "gomod"
+    directory: "/processor/websocketprocessor"
+    schedule:
+      interval: "weekly"
+      day: "wednesday"
   - package-ecosystem: "gomod"
     directory: "/receiver/activedirectorydsreceiver"
     schedule:

diff --git a/cmd/otelcontribcol/builder-config.yaml b/cmd/otelcontribcol/builder-config.yaml
@@ -100,7 +100,6 @@ processors:
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/processor/spanprocessor v0.77.0
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/processor/tailsamplingprocessor v0.77.0
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor v0.77.0
-  - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/processor/websocketprocessor v0.71.0
 
 receivers:
   - gomod: go.opentelemetry.io/collector/receiver/otlpreceiver v0.77.0
@@ -279,7 +278,6 @@ replaces:
   - github.com/open-telemetry/opentelemetry-collector-contrib/receiver/k8seventsreceiver => ../../receiver/k8seventsreceiver
   - github.com/open-telemetry/opentelemetry-collector-contrib/receiver/k8sclusterreceiver => ../../receiver/k8sclusterreceiver
   - github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor => ../../processor/transformprocessor
-  - github.com/open-telemetry/opentelemetry-collector-contrib/processor/websocketprocessor => ../../processor/websocketprocessor
   - github.com/open-telemetry/opentelemetry-collector-contrib/processor/filterprocessor => ../../processor/filterprocessor
   - github.com/open-telemetry/opentelemetry-collector-contrib/extension/observer/hostobserver => ../../extension/observer/hostobserver
   - github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kafkareceiver => ../../receiver/kafkareceiver

diff --git a/cmd/otelcontribcol/components.go b/cmd/otelcontribcol/components.go
diff --git a/cmd/otelcontribcol/components_test.go b/cmd/otelcontribcol/components_test.go
diff --git a/cmd/otelcontribcol/go.mod b/cmd/otelcontribcol/go.mod
@@ -2,7 +2,7 @@
 
 module github.com/open-telemetry/opentelemetry-collector-contrib/cmd/otelcontribcol
 
-go 1.18
+go 1.19
 
 require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/connector/countconnector v0.77.0
@@ -93,7 +93,6 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/spanprocessor v0.77.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/tailsamplingprocessor v0.77.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor v0.77.0
-	github.com/open-telemetry/opentelemetry-collector-contrib/processor/websocketprocessor v0.71.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/activedirectorydsreceiver v0.77.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/aerospikereceiver v0.77.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/apachereceiver v0.77.0
@@ -837,8 +836,6 @@ replace github.com/open-telemetry/opentelemetry-collector-contrib/receiver/k8scl
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor => ../../processor/transformprocessor
 
-replace github.com/open-telemetry/opentelemetry-collector-contrib/processor/websocketprocessor => ../../processor/websocketprocessor
-
 replace github.com/open-telemetry/opentelemetry-collector-contrib/processor/filterprocessor => ../../processor/filterprocessor
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/extension/observer/hostobserver => ../../extension/observer/hostobserver

diff --git a/cmd/otelcontribcol/main.go b/cmd/otelcontribcol/main.go
diff --git a/cmd/otelcontribcol/main_windows.go b/cmd/otelcontribcol/main_windows.go
diff --git a/go.mod b/go.mod
@@ -81,7 +81,6 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/spanprocessor v0.77.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/tailsamplingprocessor v0.77.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor v0.77.0
-	github.com/open-telemetry/opentelemetry-collector-contrib/processor/websocketprocessor v0.0.0-00010101000000-000000000000
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/activedirectorydsreceiver v0.77.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/aerospikereceiver v0.77.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/apachereceiver v0.77.0
@@ -878,8 +877,6 @@ replace github.com/open-telemetry/opentelemetry-collector-contrib/processor/tail
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor => ./processor/transformprocessor
 
-replace github.com/open-telemetry/opentelemetry-collector-contrib/processor/websocketprocessor => ./processor/websocketprocessor
-
 replace github.com/open-telemetry/opentelemetry-collector-contrib/receiver/activedirectorydsreceiver => ./receiver/activedirectorydsreceiver
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/receiver/aerospikereceiver => ./receiver/aerospikereceiver

@@ -0,0 +1 @@
+include ../../Makefile.Common
@@ -0,0 +1,28 @@
+# Websocket Processor
+
+The WebSocket processor, which can be positioned anywhere in a pipeline, allows
+data to pass through to the next component. Simultaneously, it makes a portion
+of the data accessible to WebSocket clients connecting on a configurable port.
+This functionality resembles that of the Unix `tee` command, which enables data
+to
+flow through while duplicating and redirecting it for inspection.
+
+To avoid overloading clients, the amount of telemetry transmitted via WebSockets
+is limited and can be adjusted according to user preferences.
+
+## Config
+
+The WebSocket processor has two configurable fields: `port` and `limit`:
+
+- `port`: The port on which the WebSocket processor listens. Optional. Defaults
+  to `12001`.
+- `limit`: The rate limit over the WebSocket in messages per second. Can be a
+  float or an integer. Optional. Defaults to `1`.
+
+Example configuration:
+
+```yaml
+websocket:
+  port: 12001
+  limit: 1 # rate limit 1 msg/sec
+```
diff --git a/processor/websocketprocessor/channelset.go b/processor/websocketprocessor/channelset.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package websocketprocessor
+package websocketprocessor // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/websocketprocessor"
 
 import (
 	"go.opentelemetry.io/collector/component"
@@ -26,7 +26,7 @@ type Config struct {
 	// Defaults to 12001.
 	Port int `mapstructure:"port"`
 	// Limit is a float that indicates the maximum number of messages repeated
-	// through the websocket by this processor in messages per second.
+	// through the websocket by this processor in messages per second. Defaults to 1.
 	Limit rate.Limit `mapstructure:"limit"`
 }
 

@@ -16,20 +16,12 @@ package websocketprocessor
 
 import (
 	"testing"
-	"time"
 
 	"github.com/stretchr/testify/assert"
 )
 
-func TestChannelset(t *testing.T) {
-	cs := newChannelSet()
-	ch := make(chan []byte)
-	key := cs.add(ch)
-	go func() {
-		cs.writeBytes([]byte("hello"))
-	}()
-	assert.Eventually(t, func() bool {
-		return assert.Equal(t, []byte("hello"), <-ch)
-	}, time.Second, time.Millisecond*10)
-	cs.closeAndRemove(key)
+func TestCreateDefaultConfig(t *testing.T) {
+	cfg := createDefaultConfig().(*Config)
+	assert.Equal(t, 12001, cfg.Port)
+	assert.EqualValues(t, 1, cfg.Limit)
 }
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package websocketprocessor
+package websocketprocessor // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/websocketprocessor"
 
 import (
 	"go.opentelemetry.io/collector/component"
@@ -28,6 +28,5 @@ func NewFactory() processor.Factory {
 	return processor.NewFactory(
 		typeStr,
 		createDefaultConfig,
-		processor.WithMetrics(newProdMetricsProcessor, stability),
 	)
 }