[processor/websocketprocessor] add websocket processor skeleton (#20883)

This change adds just the config, factory, and initial README for the WebSocket processor.

.chloggen/websocket-processor-1.yaml

@@ -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:

.github/CODEOWNERS

@@ -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

.github/ISSUE_TEMPLATE/bug_report.yaml

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

.github/ISSUE_TEMPLATE/feature_request.yaml

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

.github/ISSUE_TEMPLATE/other.yaml

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

.github/dependabot.yml

@@ -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:

processor/websocketprocessor/Makefile

@@ -0,0 +1 @@
+include ../../Makefile.Common

processor/websocketprocessor/README.md

@@ -0,0 +1,27 @@
+# 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 duplicated over 
+any open WebSockets is rate limited by an adjustable amount.
+
+## 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
+```

processor/websocketprocessor/config.go

@@ -0,0 +1,38 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package websocketprocessor // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/websocketprocessor"
+
+import (
+	"go.opentelemetry.io/collector/component"
+	"golang.org/x/time/rate"
+)
+
+const defaultPort = 12001
+
+type Config struct {
+	// Port indicates the port used by the web socket listener started by this processor.
+	// 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. Defaults to 1.
+	Limit rate.Limit `mapstructure:"limit"`
+}
+
+func createDefaultConfig() component.Config {
+	return &Config{
+		Port:  defaultPort,
+		Limit: 1,
+	}
+}

processor/websocketprocessor/config_test.go

@@ -0,0 +1,27 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package websocketprocessor
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestCreateDefaultConfig(t *testing.T) {
+	cfg := createDefaultConfig().(*Config)
+	assert.Equal(t, 12001, cfg.Port)
+	assert.EqualValues(t, 1, cfg.Limit)
+}

processor/websocketprocessor/factory.go

@@ -0,0 +1,32 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package websocketprocessor // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/websocketprocessor"
+
+import (
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/processor"
+)
+
+const (
+	typeStr   = "websocket"
+	stability = component.StabilityLevelDevelopment
+)
+
+func NewFactory() processor.Factory {
+	return processor.NewFactory(
+		typeStr,
+		createDefaultConfig,
+	)
+}

processor/websocketprocessor/factory_test.go

@@ -0,0 +1,28 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package websocketprocessor
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNewFactory(t *testing.T) {
+	factory := NewFactory()
+	assert.EqualValues(t, "websocket", factory.Type())
+	config := factory.CreateDefaultConfig()
+	assert.NotNil(t, config)
+}

processor/websocketprocessor/go.mod

@@ -0,0 +1,41 @@
+module github.com/open-telemetry/opentelemetry-collector-contrib/processor/websocketprocessor
+
+go 1.19
+
+require (
+	github.com/stretchr/testify v1.8.2
+	go.opentelemetry.io/collector v0.77.0
+	go.opentelemetry.io/collector/component v0.77.0
+	golang.org/x/time v0.0.0-20190308202827-9d24e82272b4
+)
+
+require (
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/golang/protobuf v1.5.3 // indirect
+	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/knadh/koanf v1.5.0 // indirect
+	github.com/mitchellh/copystructure v1.2.0 // indirect
+	github.com/mitchellh/mapstructure v1.5.0 // indirect
+	github.com/mitchellh/reflectwalk v1.0.2 // indirect
+	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
+	github.com/modern-go/reflect2 v1.0.2 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
+	go.opentelemetry.io/collector/confmap v0.77.0 // indirect
+	go.opentelemetry.io/collector/consumer v0.77.0 // indirect
+	go.opentelemetry.io/collector/featuregate v0.77.0 // indirect
+	go.opentelemetry.io/collector/pdata v1.0.0-rcv0011 // indirect
+	go.opentelemetry.io/otel v1.15.1 // indirect
+	go.opentelemetry.io/otel/metric v0.38.1 // indirect
+	go.opentelemetry.io/otel/trace v1.15.1 // indirect
+	go.uber.org/atomic v1.10.0 // indirect
+	go.uber.org/multierr v1.11.0 // indirect
+	go.uber.org/zap v1.24.0 // indirect
+	golang.org/x/net v0.9.0 // indirect
+	golang.org/x/sys v0.7.0 // indirect
+	golang.org/x/text v0.9.0 // indirect
+	google.golang.org/genproto v0.0.0-20230110181048-76db0878b65f // indirect
+	google.golang.org/grpc v1.54.0 // indirect
+	google.golang.org/protobuf v1.30.0 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
+)