[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:https://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:https://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:https://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:https://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
+)