Integ/open search exporter setup

cmd/otelcontribcol/builder-config.yaml

@@ -66,7 +66,6 @@ exporters:
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/exporter/lokiexporter v0.80.0
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/exporter/mezmoexporter v0.80.0
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/exporter/opencensusexporter v0.80.0
-  - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/exporter/opensearchexporter v0.80.0
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/exporter/parquetexporter v0.80.0
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/exporter/prometheusexporter v0.80.0
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/exporter/prometheusremotewriteexporter v0.80.0
@@ -356,7 +355,6 @@ replaces:
   - github.com/open-telemetry/opentelemetry-collector-contrib/exporter/fileexporter => ../../exporter/fileexporter
   - github.com/open-telemetry/opentelemetry-collector-contrib/pkg/resourcetotelemetry => ../../pkg/resourcetotelemetry
   - github.com/open-telemetry/opentelemetry-collector-contrib/exporter/opencensusexporter => ../../exporter/opencensusexporter
-  - github.com/open-telemetry/opentelemetry-collector-contrib/exporter/opensearchexporter => ../../exporter/opensearchexporter
   - github.com/open-telemetry/opentelemetry-collector-contrib/internal/metadataproviders => ../../internal/metadataproviders
   - github.com/open-telemetry/opentelemetry-collector-contrib/internal/aws/proxy => ../../internal/aws/proxy
   - github.com/open-telemetry/opentelemetry-collector-contrib/receiver/snmpreceiver => ../../receiver/snmpreceiver

exporter/opensearchexporter/README.md

@@ -0,0 +1,35 @@
+# OpenSearch Exporter
+
+| Status                   |               |
+| ------------------------ |---------------|
+| Stability                | [development] |
+| Supported pipeline types | traces        |
+| Distributions            | [contrib]     |
+
+This exporter supports sending OpenTelemetry signals as documents to [OpenSearch](https://www.opensearch.org).
+
+The documents are sent using [observability catalog](https://github.com/opensearch-project/opensearch-catalog/tree/main/schema/observability) schema.
+
+## Getting Started
+
+The settings are:
+
+- `endpoint` (required, default = 0.0.0.0:3500 for grpc protocol, 0.0.0.0:3600 http protocol): host:port to which the receiver is going to receive data.
+- `use_incoming_timestamp` (optional, default = false) if set `true` the timestamp from Loki log entry is used
+
+Example:
+```yaml
+exporters:
+  opensearch/trace:
+    endpoints: [https://opensearch.example.com:9200]
+# ······
+service:
+  pipelines:
+    traces:
+      receivers: [otlp]
+      exporters: [opensearch/trace]
+      processors: [batch]
+```
+
+[development]:https://github.com/open-telemetry/opentelemetry-collector#development
+[contrib]:https://github.com/open-telemetry/opentelemetry-collector-releases/tree/main/distributions/otelcol-contrib

exporter/opensearchexporter/config.go

@@ -3,5 +3,120 @@
 
 package opensearchexporter // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/opensearchexporter"
 
+import (
+	"errors"
+	"os"
+	"time"
+
+	"go.opentelemetry.io/collector/config/configopaque"
+	"go.opentelemetry.io/collector/config/configtls"
+)
+
+const (
+	defaultOpenSearchEnvName = "OPENSEARCH_URL"
+)
+
+// Config defines configuration for OpenSearch exporter.
 type Config struct {
+
+	// Endpoints holds the OpenSearch URLs the exporter should send events to.
+	//
+	// OPENSEARCH_URL environment variable is not set.
+	Endpoints []string `mapstructure:"endpoints"`
+
+	// NumWorkers configures the number of workers publishing bulk requests.
+	NumWorkers int `mapstructure:"num_workers"`
+
+	HTTPClientSettings `mapstructure:",squash"`
+	Retry              RetrySettings `mapstructure:"retry"`
+	Flush              FlushSettings `mapstructure:"flush"`
+	Namespace          string        `mapstructure:"namespace"`
+	Dataset            string        `mapstructure:"dataset"`
+}
+
+type HTTPClientSettings struct {
+	Authentication AuthenticationSettings `mapstructure:",squash"`
+
+	// ReadBufferSize for HTTP client. See http.Transport.ReadBufferSize.
+	ReadBufferSize int `mapstructure:"read_buffer_size"`
+
+	// WriteBufferSize for HTTP client. See http.Transport.WriteBufferSize.
+	WriteBufferSize int `mapstructure:"write_buffer_size"`
+
+	// Timeout configures the HTTP request timeout.
+	Timeout time.Duration `mapstructure:"timeout"`
+
+	// Headers allows users to configure optional HTTP headers that
+	// will be sent with each HTTP request.
+	Headers map[string]string `mapstructure:"headers,omitempty"`
+
+	configtls.TLSClientSetting `mapstructure:"tls,omitempty"`
+}
+
+// AuthenticationSettings defines user authentication related settings.
+type AuthenticationSettings struct {
+	// User is used to configure HTTP Basic Authentication.
+	User string `mapstructure:"user"`
+
+	// Password is used to configure HTTP Basic Authentication.
+	Password configopaque.String `mapstructure:"password"`
+}
+
+// FlushSettings  defines settings for configuring the write buffer flushing
+// policy in the OpenSearch exporter. The exporter sends a bulk request with
+// all events already serialized into the send-buffer.
+type FlushSettings struct {
+	// Bytes sets the send buffer flushing limit.
+	Bytes int `mapstructure:"bytes"`
+
+	// Interval configures the max age of a document in the send buffer.
+	Interval time.Duration `mapstructure:"interval"`
+}
+
+// RetrySettings defines settings for the HTTP request retries in the OpenSearch exporter.
+// Failed sends are retried with exponential backoff.
+type RetrySettings struct {
+	// Enabled allows users to disable retry without having to comment out all settings.
+	Enabled bool `mapstructure:"enabled"`
+
+	// MaxRequests configures how often an HTTP request is retried before it is assumed to be failed.
+	MaxRequests int `mapstructure:"max_requests"`
+
+	// InitialInterval configures the initial waiting time if a request failed.
+	InitialInterval time.Duration `mapstructure:"initial_interval"`
+
+	// MaxInterval configures the max waiting time if consecutive requests failed.
+	MaxInterval time.Duration `mapstructure:"max_interval"`
+}
+
+type MappingsSettings struct {
+	// Mode configures the field mappings.
+	Mode string `mapstructure:"mode"`
+
+	// Try to find and remove duplicate fields
+	Dedup bool `mapstructure:"dedup"`
+
+	Dedot bool `mapstructure:"dedot"`
+}
+
+var (
+	errConfigNoEndpoint    = errors.New("endpoints must be specified")
+	errConfigEmptyEndpoint = errors.New("endpoints must not include empty entries")
+)
+
+// Validate validates the opensearch server configuration.
+func (cfg *Config) Validate() error {
+	if len(cfg.Endpoints) == 0 {
+		if os.Getenv(defaultOpenSearchEnvName) == "" {
+			return errConfigNoEndpoint
+		}
+	}
+
+	for _, endpoint := range cfg.Endpoints {
+		if endpoint == "" {
+			return errConfigEmptyEndpoint
+		}
+	}
+
+	return nil
 }

exporter/opensearchexporter/factory.go

@@ -4,10 +4,55 @@
 package opensearchexporter // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/opensearchexporter"
 
 import (
+	"context"
+	"time"
+
+	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/exporter"
+	"go.opentelemetry.io/collector/exporter/exporterhelper"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+)
+
+const (
+	// The value of "type" key in configuration.
+	typeStr = "opensearch"
+	// The stability level of the exporter.
+	stability = component.StabilityLevelDevelopment
 )
 
 // NewFactory creates a factory for OpenSearch exporter.
 func NewFactory() exporter.Factory {
-	return nil
+	return exporter.NewFactory(
+		typeStr,
+		createDefaultConfig,
+		exporter.WithTraces(createTracesExporter, stability),
+	)
+}
+
+func createDefaultConfig() component.Config {
+	return &Config{
+		HTTPClientSettings: HTTPClientSettings{
+			Timeout: 90 * time.Second,
+		},
+		Namespace: "",
+		Dataset:   "",
+		Retry: RetrySettings{
+			Enabled:         true,
+			MaxRequests:     3,
+			InitialInterval: 100 * time.Millisecond,
+			MaxInterval:     1 * time.Minute,
+		},
+	}
+}
+
+func createTracesExporter(ctx context.Context,
+	set exporter.CreateSettings,
+	cfg component.Config) (exporter.Traces, error) {
+
+	return exporterhelper.NewTracesExporter(ctx, set, cfg, func(ctx context.Context, ld ptrace.Traces) error {
+		return nil
+	},
+		exporterhelper.WithShutdown(func(ctx context.Context) error {
+			return nil
+		}))
 }

exporter/opensearchexporter/go.mod

@@ -3,12 +3,20 @@ module github.com/open-telemetry/opentelemetry-collector-contrib/exporter/opense
 go 1.19
 
 require (
+	github.com/cenkalti/backoff/v4 v4.2.1
+	github.com/open-telemetry/opentelemetry-collector-contrib/internal/common v0.80.0
+	github.com/opensearch-project/opensearch-go/v2 v2.3.0
 	go.opentelemetry.io/collector/component v0.80.0
+	go.opentelemetry.io/collector/config/configopaque v0.81.0
+	go.opentelemetry.io/collector/config/configtls v0.80.0
 	go.opentelemetry.io/collector/exporter v0.80.0
+	go.opentelemetry.io/collector/pdata v1.0.0-rcv0013
+	go.uber.org/multierr v1.11.0
+	go.uber.org/zap v1.24.0
 )
 
 require (
-	github.com/cenkalti/backoff/v4 v4.2.1 // indirect
+	github.com/fsnotify/fsnotify v1.6.0 // 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
@@ -25,15 +33,12 @@ require (
 	go.opentelemetry.io/collector/consumer v0.80.0 // indirect
 	go.opentelemetry.io/collector/extension v0.80.0 // indirect
 	go.opentelemetry.io/collector/featuregate v1.0.0-rcv0013 // indirect
-	go.opentelemetry.io/collector/pdata v1.0.0-rcv0013 // indirect
 	go.opentelemetry.io/collector/processor v0.80.0 // indirect
 	go.opentelemetry.io/collector/receiver v0.80.0 // indirect
 	go.opentelemetry.io/otel v1.16.0 // indirect
 	go.opentelemetry.io/otel/metric v1.16.0 // indirect
 	go.opentelemetry.io/otel/trace v1.16.0 // 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.11.0 // indirect
 	golang.org/x/sys v0.9.0 // indirect
 	golang.org/x/text v0.10.0 // indirect