[receiver/faro] add initial implementation of faroreceiver

receiver/faroreceiver/Makefile

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

receiver/faroreceiver/README.md

@@ -0,0 +1,20 @@
+# Faro Receiver
+
+| Status                   |                     |
+| ------------------------ |---------------------|
+| Stability                | [development]       |
+| Supported pipeline types | traces, logs        |
+| Distributions            | []                  |
+
+Receives data via HTTP from the [faro-web-sdk](https://github.com/grafana/faro-web-sdk) Faro transport.
+
+## Getting Started
+
+```yaml
+receivers:
+  faro:
+```
+
+## Advanced Configuration
+
+TODO

receiver/faroreceiver/config.go

@@ -0,0 +1,23 @@
+// 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 faroreceiver
+
+import (
+	"go.opentelemetry.io/collector/config/confighttp"
+)
+
+type Config struct {
+	HTTP *confighttp.HTTPServerSettings `mapstructure:"http"`
+}

receiver/faroreceiver/factory.go

@@ -0,0 +1,94 @@
+// 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 faroreceiver
+
+import (
+	"context"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config/confighttp"
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/receiver"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/sharedcomponent"
+)
+
+const (
+	typeStr   = "faro"
+	stability = component.StabilityLevelDevelopment
+
+	defaultHTTPEndpoint = "0.0.0.0:8886"
+)
+
+func NewFactory() receiver.Factory {
+	return receiver.NewFactory(
+		typeStr,
+		createDefaultConfig,
+		receiver.WithLogs(createLogReceiver, stability),
+		receiver.WithTraces(createTraceReceiver, stability),
+	)
+}
+
+func createDefaultConfig() component.Config {
+	return &Config{
+		HTTP: &confighttp.HTTPServerSettings{
+			Endpoint: defaultHTTPEndpoint,
+		},
+	}
+}
+
+// createLogReceiver creates a log receiver based on provided config.
+func createLogReceiver(
+	_ context.Context,
+	set receiver.CreateSettings,
+	cfg component.Config,
+	next consumer.Logs,
+) (receiver.Logs, error) {
+	r := receivers.GetOrAdd(cfg, func() component.Component {
+		c := cfg.(*Config)
+		return newFaroReceiver(*c, set)
+	})
+
+	if err := r.Unwrap().(*faroReceiver).registerLogConsumer(set.Logger, next); err != nil {
+		return nil, err
+	}
+
+	return r, nil
+}
+
+// createTraceReceiver creates a trace receiver based on provided config.
+func createTraceReceiver(
+	_ context.Context,
+	set receiver.CreateSettings,
+	cfg component.Config,
+	next consumer.Traces,
+) (receiver.Traces, error) {
+	r := receivers.GetOrAdd(cfg, func() component.Component {
+		c := cfg.(*Config)
+		return newFaroReceiver(*c, set)
+	})
+
+	if err := r.Unwrap().(*faroReceiver).registerTraceConsumer(set.Logger, next); err != nil {
+		return nil, err
+	}
+
+	return r, nil
+}
+
+// This is the map of already created Faro receivers for particular configurations.
+// We maintain this map because the Factory is asked trace and log receivers separately
+// when it gets createTraceReceiver() and createLogReceiver() but they must not
+// create separate objects, they must use one faroReceiver object per configuration.
+var receivers = sharedcomponent.NewSharedComponents()

receiver/faroreceiver/faro.go

@@ -0,0 +1,128 @@
+// 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 faroreceiver // import "github.com/grafana/opentelemetry-collector-components/components/receiver/faroreceiver"
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"sync"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config/confighttp"
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/receiver"
+	"go.uber.org/zap"
+)
+
+type faroReceiver struct {
+	cfg *Config
+	set receiver.CreateSettings
+
+	server       *http.Server
+	startOnce    sync.Once
+	shutdownOnce sync.Once
+	shutdownWG   sync.WaitGroup
+
+	logConsumer   consumer.Logs
+	traceConsumer consumer.Traces
+
+	logLogger   *zap.Logger
+	traceLogger *zap.Logger
+
+	// TODO
+	// obsrepHTTP *obsreport.Receiver
+}
+
+// newFaroReceiver TODO...
+func newFaroReceiver(cfg Config, set receiver.CreateSettings) *faroReceiver {
+	return &faroReceiver{
+		cfg: &cfg,
+		set: set,
+	}
+}
+
+func (r *faroReceiver) registerTraceConsumer(log *zap.Logger, tc consumer.Traces) error {
+	if tc == nil {
+		return component.ErrNilNextConsumer
+	}
+	r.traceLogger = log
+	r.traceConsumer = tc
+	return nil
+}
+
+func (r *faroReceiver) registerLogConsumer(log *zap.Logger, lc consumer.Logs) error {
+	if lc == nil {
+		return component.ErrNilNextConsumer
+	}
+	r.logLogger = log
+	r.logConsumer = lc
+	return nil
+}
+
+// Start will start the receiver http server
+func (r *faroReceiver) Start(ctx context.Context, host component.Host) error {
+	var err error
+	r.startOnce.Do(func() {
+		if r.cfg.HTTP != nil {
+			if r.server, err = r.cfg.HTTP.ToServer(
+				host,
+				r.set.TelemetrySettings,
+				&Handler{
+					logLogger:     r.logLogger,
+					traceLogger:   r.traceLogger,
+					logConsumer:   r.logConsumer,
+					traceConsumer: r.traceConsumer,
+				},
+			); err != nil {
+				return
+			}
+
+			err = r.startHTTPServer(r.cfg.HTTP, host)
+		}
+	})
+	return err
+}
+
+// Shutdown stops the receiver
+func (r *faroReceiver) Shutdown(ctx context.Context) error {
+	var err error
+	r.shutdownOnce.Do(func() {
+		r.set.Logger.Info("Stopping HTTP server")
+		if r.server == nil {
+			return
+		}
+		err = r.server.Shutdown(ctx)
+		r.shutdownWG.Wait()
+	})
+	return err
+}
+
+func (r *faroReceiver) startHTTPServer(cfg *confighttp.HTTPServerSettings, host component.Host) error {
+	r.set.Logger.Info("Starting HTTP server", zap.String("endpoint", cfg.Endpoint))
+	hln, err := cfg.ToListener()
+	if err != nil {
+		return err
+	}
+	r.shutdownWG.Add(1)
+	go func() {
+		defer r.shutdownWG.Done()
+
+		if errHTTP := r.server.Serve(hln); errHTTP != nil && !errors.Is(errHTTP, http.ErrServerClosed) {
+			host.ReportFatalError(errHTTP)
+		}
+	}()
+	return err
+}

receiver/faroreceiver/go.mod

@@ -0,0 +1,46 @@
+module github.com/open-telemetry/opentelemetry-collector-contrib/receiver/faroreceiver
+
+go 1.19
+
+require (
+	github.com/go-logfmt/logfmt v0.6.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/internal/sharedcomponent v0.72.0
+	github.com/wk8/go-ordered-map v1.0.0
+	go.opentelemetry.io/collector v0.72.0
+	go.opentelemetry.io/collector/component v0.72.0
+	go.opentelemetry.io/collector/consumer v0.72.0
+	go.opentelemetry.io/collector/pdata v1.0.0-rc6
+	go.uber.org/zap v1.24.0
+)
+
+require (
+	github.com/felixge/httpsnoop v1.0.3 // indirect
+	github.com/go-logr/logr v1.2.3 // indirect
+	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/golang/protobuf v1.5.2 // indirect
+	github.com/golang/snappy v0.0.4 // indirect
+	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/klauspost/compress v1.15.15 // 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/rs/cors v1.8.3 // indirect
+	go.opentelemetry.io/collector/confmap v0.72.0 // indirect
+	go.opentelemetry.io/collector/featuregate v0.72.0 // indirect
+	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.39.0 // indirect
+	go.opentelemetry.io/otel v1.13.0 // indirect
+	go.opentelemetry.io/otel/metric v0.36.0 // indirect
+	go.opentelemetry.io/otel/trace v1.13.0 // indirect
+	go.uber.org/atomic v1.10.0 // indirect
+	go.uber.org/multierr v1.9.0 // indirect
+	golang.org/x/net v0.7.0 // indirect
+	golang.org/x/sys v0.5.0 // indirect
+	golang.org/x/text v0.7.0 // indirect
+	google.golang.org/genproto v0.0.0-20230110181048-76db0878b65f // indirect
+	google.golang.org/grpc v1.53.0 // indirect
+	google.golang.org/protobuf v1.28.1 // indirect
+)