Faro receiver component

pkg/receiver/faro/config.go

@@ -0,0 +1,71 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package faroreceiver // import "github.com/grafana/faro/pkg/receiver/faro"
+
+import (
+	"errors"
+	"fmt"
+	"net/url"
+	"path"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config/confighttp"
+	"go.opentelemetry.io/collector/confmap"
+)
+
+const protoHTTP = "protocols::http"
+
+type HTTPConfig struct {
+	*confighttp.ServerConfig `mapstructure:",squash"`
+
+	FaroURLPath string `mapstructure:"faro_url_path,omitempty"`
+}
+
+type Protocols struct {
+	HTTP *HTTPConfig `mapstructure:"http"`
+}
+
+type Config struct {
+	Protocols `mapstructure:"protocols,squash"`
+}
+
+var (
+	_ component.Config    = (*Config)(nil)
+	_ confmap.Unmarshaler = (*Config)(nil)
+)
+
+func (cfg *Config) Validate() error {
+	if cfg.HTTP == nil {
+		return errors.New("must specify HTTP protocol")
+	}
+	return nil
+}
+
+func (cfg *Config) Unmarshal(conf *confmap.Conf) error {
+	err := conf.Unmarshal(cfg)
+	if err != nil {
+		return err
+	}
+
+	if !conf.IsSet(protoHTTP) {
+		cfg.HTTP = nil
+	} else {
+		if cfg.HTTP.FaroURLPath, err = sanitizeURLPath(cfg.HTTP.FaroURLPath); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func sanitizeURLPath(urlPath string) (string, error) {
+	u, err := url.Parse(urlPath)
+	if err != nil {
+		return "", fmt.Errorf("invalid HTTP URL path set for signal: %w", err)
+	}
+
+	if !path.IsAbs(u.Path) {
+		u.Path = "/" + u.Path
+	}
+	return u.Path, nil
+}

pkg/receiver/faro/encoder.go

@@ -0,0 +1,92 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package faroreceiver // import "github.com/grafana/faro/pkg/receiver/faro"
+
+import (
+	"bytes"
+
+	"github.com/gogo/protobuf/jsonpb"
+	"go.opentelemetry.io/collector/pdata/plog/plogotlp"
+	"go.opentelemetry.io/collector/pdata/pmetric/pmetricotlp"
+	"go.opentelemetry.io/collector/pdata/pprofile/pprofileotlp"
+	"go.opentelemetry.io/collector/pdata/ptrace/ptraceotlp"
+	spb "google.golang.org/genproto/googleapis/rpc/status"
+)
+
+const (
+	pbContentType   = "application/x-protobuf"
+	jsonContentType = "application/json"
+)
+
+var (
+	jsEncoder       = &jsonEncoder{}
+	jsonPbMarshaler = &jsonpb.Marshaler{}
+)
+
+type encoder interface {
+	unmarshalTracesRequest(buf []byte) (ptraceotlp.ExportRequest, error)
+	unmarshalMetricsRequest(buf []byte) (pmetricotlp.ExportRequest, error)
+	unmarshalLogsRequest(buf []byte) (plogotlp.ExportRequest, error)
+	unmarshalProfilesRequest(buf []byte) (pprofileotlp.ExportRequest, error)
+
+	marshalTracesResponse(ptraceotlp.ExportResponse) ([]byte, error)
+	marshalMetricsResponse(pmetricotlp.ExportResponse) ([]byte, error)
+	marshalLogsResponse(plogotlp.ExportResponse) ([]byte, error)
+	marshalProfilesResponse(pprofileotlp.ExportResponse) ([]byte, error)
+
+	marshalStatus(rsp *spb.Status) ([]byte, error)
+
+	contentType() string
+}
+
+type jsonEncoder struct{}
+
+func (jsonEncoder) unmarshalTracesRequest(buf []byte) (ptraceotlp.ExportRequest, error) {
+	req := ptraceotlp.NewExportRequest()
+	err := req.UnmarshalJSON(buf)
+	return req, err
+}
+
+func (jsonEncoder) unmarshalMetricsRequest(buf []byte) (pmetricotlp.ExportRequest, error) {
+	req := pmetricotlp.NewExportRequest()
+	err := req.UnmarshalJSON(buf)
+	return req, err
+}
+
+func (jsonEncoder) unmarshalLogsRequest(buf []byte) (plogotlp.ExportRequest, error) {
+	req := plogotlp.NewExportRequest()
+	err := req.UnmarshalJSON(buf)
+	return req, err
+}
+
+func (jsonEncoder) unmarshalProfilesRequest(buf []byte) (pprofileotlp.ExportRequest, error) {
+	req := pprofileotlp.NewExportRequest()
+	err := req.UnmarshalJSON(buf)
+	return req, err
+}
+
+func (jsonEncoder) marshalTracesResponse(resp ptraceotlp.ExportResponse) ([]byte, error) {
+	return resp.MarshalJSON()
+}
+
+func (jsonEncoder) marshalMetricsResponse(resp pmetricotlp.ExportResponse) ([]byte, error) {
+	return resp.MarshalJSON()
+}
+
+func (jsonEncoder) marshalLogsResponse(resp plogotlp.ExportResponse) ([]byte, error) {
+	return resp.MarshalJSON()
+}
+
+func (jsonEncoder) marshalProfilesResponse(resp pprofileotlp.ExportResponse) ([]byte, error) {
+	return resp.MarshalJSON()
+}
+
+func (jsonEncoder) marshalStatus(resp *spb.Status) ([]byte, error) {
+	buf := new(bytes.Buffer)
+	err := jsonPbMarshaler.Marshal(buf, resp)
+	return buf.Bytes(), err
+}
+
+func (jsonEncoder) contentType() string {
+	return jsonContentType
+}

pkg/receiver/faro/go.mod

@@ -0,0 +1,70 @@
+module github.com/grafana/faro/pkg/receiver/faro
+
+go 1.23.3
+
+require go.opentelemetry.io/collector/config/confighttp v0.118.0
+
+require (
+	github.com/apapsch/go-jsonmerge/v2 v2.0.0 // indirect
+	github.com/go-logfmt/logfmt v0.6.0 // indirect
+	github.com/go-viper/mapstructure/v2 v2.2.1 // indirect
+	github.com/google/uuid v1.6.0 // indirect
+	github.com/grafana/faro/pkg/go v0.0.0-20250124090506-63d620100884 // indirect
+	github.com/grafana/faro/pkg/translator/faro v0.0.0-20250124090506-63d620100884 // indirect
+	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/klauspost/cpuid/v2 v2.2.9 // indirect
+	github.com/knadh/koanf/maps v0.1.1 // indirect
+	github.com/knadh/koanf/providers/confmap v0.1.0 // indirect
+	github.com/knadh/koanf/v2 v2.1.2 // indirect
+	github.com/mitchellh/copystructure v1.2.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/oapi-codegen/runtime v1.1.1 // indirect
+	github.com/wk8/go-ordered-map v1.0.0 // indirect
+	github.com/zeebo/xxh3 v1.0.2 // indirect
+	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
+	go.opentelemetry.io/collector/consumer/consumererror v0.118.0 // indirect
+	go.opentelemetry.io/collector/pdata/pprofile v0.118.0 // indirect
+	go.opentelemetry.io/collector/pipeline v0.118.0 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
+)
+
+require (
+	github.com/felixge/httpsnoop v1.0.4 // indirect
+	github.com/fsnotify/fsnotify v1.8.0 // indirect
+	github.com/go-logr/logr v1.4.2 // indirect
+	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/golang/snappy v0.0.4 // indirect
+	github.com/klauspost/compress v1.17.11 // indirect
+	github.com/pierrec/lz4/v4 v4.1.22 // indirect
+	github.com/rs/cors v1.11.1 // indirect
+	go.opentelemetry.io/collector/client v1.24.0 // indirect
+	go.opentelemetry.io/collector/component v0.118.0 // indirect
+	go.opentelemetry.io/collector/config/configauth v0.118.0 // indirect
+	go.opentelemetry.io/collector/config/configcompression v1.24.0 // indirect
+	go.opentelemetry.io/collector/config/configopaque v1.24.0 // indirect
+	go.opentelemetry.io/collector/config/configtelemetry v0.118.0 // indirect
+	go.opentelemetry.io/collector/config/configtls v1.24.0 // indirect
+	go.opentelemetry.io/collector/confmap v1.24.0
+	go.opentelemetry.io/collector/consumer v1.24.0
+	go.opentelemetry.io/collector/consumer/xconsumer v0.118.0
+	go.opentelemetry.io/collector/extension v0.118.0 // indirect
+	go.opentelemetry.io/collector/extension/auth v0.118.0 // indirect
+	go.opentelemetry.io/collector/pdata v1.24.0 // indirect
+	go.opentelemetry.io/collector/receiver v0.118.0
+	go.opentelemetry.io/collector/receiver/otlpreceiver v0.118.0
+	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.56.0 // indirect
+	go.opentelemetry.io/otel v1.33.0 // indirect
+	go.opentelemetry.io/otel/metric v1.33.0 // indirect
+	go.opentelemetry.io/otel/trace v1.33.0 // indirect
+	go.uber.org/multierr v1.11.0 // indirect
+	go.uber.org/zap v1.27.0 // indirect
+	golang.org/x/net v0.34.0 // indirect
+	golang.org/x/sys v0.29.0 // indirect
+	golang.org/x/text v0.21.0 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20250115164207-1a7da9e5054f // indirect
+	google.golang.org/grpc v1.69.4 // indirect
+	google.golang.org/protobuf v1.36.3 // indirect
+)