[exporter/faroexporter] Add Faro exporter logic to wireframe (#39016)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description

Faro exporter package enables exporting OpenTelemetry data to
Faro-compatible endpoints. This makes it work.

<!-- Issue number (e.g. #1234) or full URL to issue, if applicable. -->
#### Link to tracking issue

Fixes #35319

<!--Describe what testing was performed and which tests were added.-->
#### Testing

Added unit tests.

<!--Please delete paragraphs that you did not use before submitting.-->

Author: t00mas

.chloggen/faro-exporter.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# 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. filelogreceiver)
+component: faroexporter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Completes the implementation of the Faro exporter.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [35319]
+
+# (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:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: []

exporter/faroexporter/exporter.go

@@ -4,28 +4,46 @@
 package faroexporter // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/faroexporter"
 
 import (
+	"bytes"
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
+	"net/http"
 	"net/url"
 	"runtime"
+	"strconv"
+	"sync"
+	"time"
 
+	faro "github.com/grafana/faro/pkg/go"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/consumer/consumererror"
 	"go.opentelemetry.io/collector/exporter"
+	"go.opentelemetry.io/collector/exporter/exporterhelper"
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/pdata/ptrace"
+	"go.uber.org/multierr"
 	"go.uber.org/zap"
+
+	farotranslator "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/faro"
 )
 
 type faroExporter struct {
-	config *Config
-	// client    *http.Client
+	config    *Config
+	client    *http.Client
 	logger    *zap.Logger
 	settings  component.TelemetrySettings
 	userAgent string
 }
 
+const (
+	headerRetryAfter = "Retry-After"
+	jsonContentType  = "application/json"
+)
+
 func newExporter(cfg component.Config, set exporter.Settings) (*faroExporter, error) {
 	oCfg := cfg.(*Config)
 
@@ -47,16 +65,97 @@ func newExporter(cfg component.Config, set exporter.Settings) (*faroExporter, er
 	}, nil
 }
 
-func (fe *faroExporter) start(_ context.Context, _ component.Host) error {
+func (fe *faroExporter) start(ctx context.Context, host component.Host) error {
+	client, err := fe.config.ClientConfig.ToClient(ctx, host, fe.settings)
+	if err != nil {
+		return err
+	}
+	fe.client = client
 	return nil
 }
 
-func (fe *faroExporter) ConsumeTraces(_ context.Context, _ ptrace.Traces) error {
-	return nil
+func (fe *faroExporter) ConsumeTraces(ctx context.Context, td ptrace.Traces) error {
+	fp, err := farotranslator.TranslateFromTraces(ctx, td)
+	if err != nil {
+		return fmt.Errorf("failed to translate traces to faro payloads: %w", err)
+	}
+	return fe.consume(ctx, fp)
 }
 
-func (fe *faroExporter) ConsumeLogs(_ context.Context, _ plog.Logs) error {
-	return nil
+func (fe *faroExporter) ConsumeLogs(ctx context.Context, ld plog.Logs) error {
+	fp, err := farotranslator.TranslateFromLogs(ctx, ld)
+	if err != nil {
+		return fmt.Errorf("failed to translate logs to faro payloads: %w", err)
+	}
+	return fe.consume(ctx, fp)
+}
+
+func (fe *faroExporter) export(ctx context.Context, fp *faro.Payload) error {
+	fe.logger.Debug("Preparing to make HTTP request", zap.String("endpoint", fe.config.Endpoint))
+	request, err := json.Marshal(fp)
+	if err != nil {
+		return consumererror.NewPermanent(err)
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, fe.config.Endpoint, bytes.NewReader(request))
+	if err != nil {
+		return consumererror.NewPermanent(err)
+	}
+	req.Header.Set("Content-Type", jsonContentType)
+	req.Header.Set("User-Agent", fe.userAgent)
+
+	resp, err := fe.client.Do(req)
+	if err != nil {
+		return fmt.Errorf("failed to make an HTTP request: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode == http.StatusAccepted {
+		return nil
+	}
+
+	var errString string
+	var formattedErr error
+	bodyBytes, err := io.ReadAll(resp.Body)
+	bodyContent := "unknown response"
+	if err == nil {
+		bodyContent = string(bodyBytes)
+	}
+
+	errString = fmt.Sprintf(
+		"error exporting items, request to %s responded with HTTP Status Code %d, Message=%s",
+		fe.config.Endpoint, resp.StatusCode, bodyContent)
+	formattedErr = newStatusFromMsgAndHTTPCode(errString, resp.StatusCode).Err()
+
+	if isRetryableStatusCode(resp.StatusCode) {
+		retryAfter := 0
+		isThrottleError := resp.StatusCode == http.StatusTooManyRequests || resp.StatusCode == http.StatusServiceUnavailable
+		if val := resp.Header.Get(headerRetryAfter); isThrottleError && val != "" {
+			if seconds, err := strconv.Atoi(val); err == nil {
+				retryAfter = seconds
+			}
+		}
+		return exporterhelper.NewThrottleRetry(formattedErr, time.Duration(retryAfter)*time.Second)
+	}
+
+	return consumererror.NewPermanent(formattedErr)
+}
+
+func (fe *faroExporter) consume(ctx context.Context, fp []faro.Payload) error {
+	var errs error
+	var wg sync.WaitGroup
+	wg.Add(len(fp))
+	var mu sync.Mutex
+	for _, p := range fp {
+		go func() {
+			defer wg.Done()
+			err := fe.export(ctx, &p)
+			mu.Lock()
+			errs = multierr.Append(errs, err)
+			mu.Unlock()
+		}()
+	}
+	wg.Wait()
+	return errs
 }
 
 func (fe *faroExporter) Capabilities() consumer.Capabilities {