Prototype "connectors", a new type of pipeline component

.chloggen/connectors-prototype-v2.yaml

@@ -0,0 +1,22 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: 'enhancement'
+
+# The name of the component, or a single word describing the area of concern, (e.g. otlpreceiver)
+component: connectors
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add "connectors", a new type of pipeline component
+
+# One or more tracking issues or pull requests related to the change
+issues: [2336]
+
+# (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: |
+  - Connectors connect pipelines. A connector acts as exporter and a receiver working together.
+  For example, a `nopconnector`` may be used to replicate signals by exporting data from one
+  pipeline and receiving the data on two other pipelines, which can each process a copy of the data
+  in different ways.
+  Connectors can also derive one signal from another. For example, a `countconnector` can be used as
+  an exporter on a logs pipeline and emit metrics, describing the number of logs, onto a metrics pipeline.

.github/dependabot.yml

@@ -26,6 +26,14 @@ updates:
     directory: "/consumer"
     schedule:
       interval: "weekly"
+  - package-ecosystem: "gomod"
+    directory: "/connector/countconnector"
+    schedule:
+      interval: "weekly"
+  - package-ecosystem: "gomod"
+    directory: "/connector/nopconnector"
+    schedule:
+      interval: "weekly"
   - package-ecosystem: "gomod"
     directory: "/exporter/loggingexporter"
     schedule:

.gitignore

@@ -1,5 +1,6 @@
 bin/
 dist/
+local/
 
 # GoLand IDEA
 /.idea/

cmd/builder/internal/builder/config.go

@@ -42,6 +42,7 @@ type Config struct {
 	Extensions   []Module     `mapstructure:"extensions"`
 	Receivers    []Module     `mapstructure:"receivers"`
 	Processors   []Module     `mapstructure:"processors"`
+	Connectors   []Module     `mapstructure:"connectors"`
 	Replaces     []string     `mapstructure:"replaces"`
 	Excludes     []string     `mapstructure:"excludes"`
 }
@@ -90,7 +91,13 @@ func NewDefaultConfig() Config {
 
 // Validate checks whether the current configuration is valid
 func (c *Config) Validate() error {
-	return multierr.Combine(validateModules(c.Extensions), validateModules(c.Receivers), validateModules(c.Exporters), validateModules(c.Processors))
+	return multierr.Combine(
+		validateModules(c.Extensions),
+		validateModules(c.Receivers),
+		validateModules(c.Exporters),
+		validateModules(c.Processors),
+		validateModules(c.Connectors),
+	)
 }
 
 // SetGoPath sets go path
@@ -133,6 +140,11 @@ func (c *Config) ParseModules() error {
 		return err
 	}
 
+	c.Connectors, err = parseModules(c.Connectors)
+	if err != nil {
+		return err
+	}
+
 	return nil
 }
 

cmd/builder/internal/builder/main_test.go

@@ -15,6 +15,8 @@
 package builder
 
 import (
+	"fmt"
+	"path/filepath"
 	"runtime"
 	"testing"
 	"time"
@@ -49,8 +51,11 @@ func TestGenerateAndCompileDefault(t *testing.T) {
 	cfg := NewDefaultConfig()
 	cfg.Distribution.OutputPath = t.TempDir()
 
-	// we override this version, otherwise this would break during releases
-	cfg.Distribution.OtelColVersion = "0.52.0"
+	_, thisFile, _, _ := runtime.Caller(0)
+	workspaceDir := filepath.Dir(filepath.Dir(filepath.Dir(filepath.Dir(filepath.Dir(thisFile)))))
+
+	cfg.Replaces = append(cfg.Replaces, fmt.Sprintf("go.opentelemetry.io/collector => %s", workspaceDir))
+	cfg.Replaces = append(cfg.Replaces, fmt.Sprintf("go.opentelemetry.io/collector/component => %s/component", workspaceDir))
 
 	assert.NoError(t, cfg.Validate())
 	assert.NoError(t, cfg.SetGoPath())

cmd/builder/internal/builder/templates/components.go.tmpl

@@ -4,6 +4,9 @@ package main
 
 import (
 	"go.opentelemetry.io/collector/component"
+	{{- range .Connectors}}
+	{{.Name}} "{{.Import}}"
+	{{- end}}
 	{{- range .Exporters}}
 	{{.Name}} "{{.Import}}"
 	{{- end}}
@@ -31,6 +34,15 @@ func components() (component.Factories, error) {
 		return component.Factories{}, err
 	}
 
+	factories.Connectors, err = component.MakeConnectorFactoryMap(
+		{{- range .Connectors}}
+		{{.Name}}.NewFactory(),
+		{{- end}}
+	)
+	if err != nil {
+		return component.Factories{}, err
+	}
+
 	factories.Receivers, err = component.MakeReceiverFactoryMap(
 		{{- range .Receivers}}
 		{{.Name}}.NewFactory(),

cmd/builder/internal/builder/templates/components_test.go.tmpl

@@ -26,4 +26,7 @@ func TestValidateConfigs(t *testing.T) {
 	for _, factory := range factories.Extensions {
 		assert.NoError(t, componenttest.CheckConfigStruct(factory.CreateDefaultConfig()))
 	}
+	for _, factory := range factories.Connectors {
+		assert.NoError(t, componenttest.CheckConfigStruct(factory.CreateDefaultConfig()))
+	}
 }

cmd/builder/internal/builder/templates/go.mod.tmpl

@@ -17,6 +17,9 @@ require (
 	{{- range .Processors}}
 	{{if .GoMod}}{{.GoMod}}{{end}}
 	{{- end}}
+	{{- range .Connectors}}
+	{{if .GoMod}}{{.GoMod}}{{end}}
+	{{- end}}
 	go.opentelemetry.io/collector v{{.Distribution.OtelColVersion}}
 )
 
@@ -32,6 +35,10 @@ require (
 {{- range .Processors}}
 {{if ne .Path ""}}replace {{.GoMod}} => {{.Path}}{{end}}
 {{- end}}
+{{- range .Connectors}}
+{{if ne .Path ""}}replace {{.GoMod}} => {{.Path}}{{end}}
+{{- end}}
+
 {{- range .Replaces}}
 replace {{.}}
 {{- end}}

cmd/builder/internal/command.go

@@ -166,6 +166,7 @@ func applyCfgFromFile(flags *flag.FlagSet, cfgFromFile builder.Config) {
 	cfg.Extensions = cfgFromFile.Extensions
 	cfg.Receivers = cfgFromFile.Receivers
 	cfg.Processors = cfgFromFile.Processors
+	cfg.Connectors = cfgFromFile.Connectors
 	cfg.Replaces = cfgFromFile.Replaces
 	cfg.Excludes = cfgFromFile.Excludes
 

cmd/builder/internal/config/default.yaml

@@ -17,4 +17,9 @@ extensions:
 processors:
   - gomod: go.opentelemetry.io/collector/processor/batchprocessor v0.65.0
   - gomod: go.opentelemetry.io/collector/processor/memorylimiterprocessor v0.65.0
+connectors:
+  - import: go.opentelemetry.io/collector/connector/countconnector
+    gomod: go.opentelemetry.io/collector/connector/countconnector v0.0.0
+  - import: go.opentelemetry.io/collector/connector/nopconnector
+    gomod: go.opentelemetry.io/collector/connector/nopconnector v0.0.0
 

cmd/builder/test/core.builder.yaml

@@ -1,15 +1,22 @@
 dist:
   module: go.opentelemetry.io/collector/builder/test/core
-  otelcol_version: 0.44.0
+  otelcol_version: 0.64.1
 
 extensions:
   - import: go.opentelemetry.io/collector/extension/zpagesextension
-    gomod: go.opentelemetry.io/collector v0.44.0
+    gomod: go.opentelemetry.io/collector v0.64.1
+    path: ${WORKSPACE_DIR}
 
 receivers:
   - import: go.opentelemetry.io/collector/receiver/otlpreceiver
-    gomod: go.opentelemetry.io/collector v0.44.0
+    gomod: go.opentelemetry.io/collector v0.64.1
+    path: ${WORKSPACE_DIR}
 
 exporters:
   - import: go.opentelemetry.io/collector/exporter/loggingexporter
-    gomod: go.opentelemetry.io/collector v0.44.0
+    gomod: go.opentelemetry.io/collector v0.64.1
+    path: ${WORKSPACE_DIR}
+
+replaces:
+  - go.opentelemetry.io/collector => ${WORKSPACE_DIR}
+  - go.opentelemetry.io/collector/component => ${WORKSPACE_DIR}/component

cmd/builder/test/test.sh

@@ -1,4 +1,6 @@
 #!/bin/bash
+SCRIPT_DIR=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd )
+export WORKSPACE_DIR=$( cd -- "$( dirname $(dirname $(dirname -- "${SCRIPT_DIR}")) )" &> /dev/null && pwd )
 
 GOBIN=$(go env GOBIN)
 if [[ "$GO" == "" ]]; then
@@ -25,7 +27,9 @@ test_build_config() {
 
     echo "Starting test '${test}' at `date`" >> "${out}/test.log"
 
-    go run . --go "${GOBIN}" --config "$build_config" --output-path "${out}" --name otelcol-built-test > "${out}/builder.log" 2>&1
+    final_build_config=$(basename ${build_config})
+    envsubst < "$build_config" > "${out}/${final_build_config}"
+    go run . --go "${GOBIN}" --config "${out}/${final_build_config}" --output-path "${out}" --name otelcol-built-test > "${out}/builder.log" 2>&1
 
     if [ $? != 0 ]; then
         echo "❌ FAIL ${test}. Failed to compile the test ${test}. Build logs:"

cmd/otelcorecol/builder-config.yaml

@@ -17,11 +17,18 @@ extensions:
 processors:
   - gomod: go.opentelemetry.io/collector/processor/batchprocessor v0.65.0
   - gomod: go.opentelemetry.io/collector/processor/memorylimiterprocessor v0.65.0
+connectors:
+  - import: go.opentelemetry.io/collector/connector/countconnector
+    gomod: go.opentelemetry.io/collector/connector/countconnector v0.0.0
+  - import: go.opentelemetry.io/collector/connector/nopconnector
+    gomod: go.opentelemetry.io/collector/connector/nopconnector v0.0.0
 
 replaces:
   - go.opentelemetry.io/collector => ../../
   - go.opentelemetry.io/collector/component => ../../component
   - go.opentelemetry.io/collector/consumer => ../../consumer
+  - go.opentelemetry.io/collector/connector/countconnector => ../../connector/countconnector
+  - go.opentelemetry.io/collector/connector/nopconnector => ../../connector/nopconnector
   - go.opentelemetry.io/collector/exporter/loggingexporter => ../../exporter/loggingexporter
   - go.opentelemetry.io/collector/exporter/otlpexporter => ../../exporter/otlpexporter
   - go.opentelemetry.io/collector/exporter/otlphttpexporter => ../../exporter/otlphttpexporter

cmd/otelcorecol/go.mod

@@ -8,6 +8,8 @@ require (
 	github.com/stretchr/testify v1.8.1
 	go.opentelemetry.io/collector v0.65.0
 	go.opentelemetry.io/collector/component v0.65.0
+	go.opentelemetry.io/collector/connector/countconnector v0.0.0
+	go.opentelemetry.io/collector/connector/nopconnector v0.0.0
 	go.opentelemetry.io/collector/exporter/loggingexporter v0.65.0
 	go.opentelemetry.io/collector/exporter/otlpexporter v0.65.0
 	go.opentelemetry.io/collector/exporter/otlphttpexporter v0.65.0
@@ -82,6 +84,7 @@ require (
 	go.uber.org/zap v1.23.0 // indirect
 	golang.org/x/net v0.0.0-20221014081412-f15817d10f9b // indirect
 	golang.org/x/text v0.4.0 // indirect
+	gonum.org/v1/gonum v0.12.0 // indirect
 	google.golang.org/genproto v0.0.0-20221027153422-115e99e71e1c // indirect
 	google.golang.org/grpc v1.51.0 // indirect
 	google.golang.org/protobuf v1.28.1 // indirect
@@ -95,6 +98,10 @@ replace go.opentelemetry.io/collector/component => ../../component
 
 replace go.opentelemetry.io/collector/consumer => ../../consumer
 
+replace go.opentelemetry.io/collector/connector/countconnector => ../../connector/countconnector
+
+replace go.opentelemetry.io/collector/connector/nopconnector => ../../connector/nopconnector
+
 replace go.opentelemetry.io/collector/exporter/loggingexporter => ../../exporter/loggingexporter
 
 replace go.opentelemetry.io/collector/exporter/otlpexporter => ../../exporter/otlpexporter

cmd/otelcorecol/go.sum

@@ -467,6 +467,7 @@ golang.org/x/exp v0.0.0-20191129062945-2f5052295587/go.mod h1:2RIsYlXP63K8oxa1u0
 golang.org/x/exp v0.0.0-20191227195350-da58074b4299/go.mod h1:2RIsYlXP63K8oxa1u096TMicItID8zy7Y6sNkU49FU4=
 golang.org/x/exp v0.0.0-20200119233911-0405dc783f0a/go.mod h1:2RIsYlXP63K8oxa1u096TMicItID8zy7Y6sNkU49FU4=
 golang.org/x/exp v0.0.0-20200207192155-f17229e696bd/go.mod h1:J/WKrq2StrnmMY6+EHIKF9dgMWnmCNThgcyBT1FY9mM=
+golang.org/x/exp v0.0.0-20200224162631-6cc2880d07d6 h1:QE6XYQK6naiK1EPAe1g/ILLxN5RBoH5xkJk3CqlMI/Y=
 golang.org/x/exp v0.0.0-20200224162631-6cc2880d07d6/go.mod h1:3jZMyOhIsHpP37uCMkUooju7aAi5cS1Q23tOzKc+0MU=
 golang.org/x/image v0.0.0-20190227222117-0694c2d4d067/go.mod h1:kZ7UVZpmo3dzQBMxlp+ypCbDeSB+sBbTgSJuh5dn5js=
 golang.org/x/image v0.0.0-20190802002840-cff245a6509b/go.mod h1:FeLwcggjj3mMvU+oOTbSwawSJRM1uh48EjtB4UJZlP0=
@@ -674,6 +675,8 @@ golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8T
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+gonum.org/v1/gonum v0.12.0 h1:xKuo6hzt+gMav00meVPUlXwSdoEJP46BR+wdxQEFK2o=
+gonum.org/v1/gonum v0.12.0/go.mod h1:73TDxJfAAHeA8Mk9mf8NlIppyhQNo5GLTcYeqgo2lvY=
 google.golang.org/api v0.4.0/go.mod h1:8k5glujaEP+g9n7WNsDg8QP6cUVNI86fCNMcbazEtwE=
 google.golang.org/api v0.7.0/go.mod h1:WtwebWUNSVBH/HAw79HIFXZNqEvBhG+Ra+ax0hx3E3M=
 google.golang.org/api v0.8.0/go.mod h1:o4eAsZoiT+ibD93RtjEohWalFOjRDx6CVaqeizhEnKg=

component/component.go

@@ -104,6 +104,7 @@ const (
 	KindProcessor
 	KindExporter
 	KindExtension
+	KindConnector
 )
 
 // StabilityLevel represents the stability level of the component created by the factory.