[exporter/elasticsearch] support sending otlp tracing to Elasticsearch

exporter/elasticsearchexporter/README.md

@@ -1,10 +1,10 @@
 # Elasticsearch Exporter
 
-| Status                   |           |
-| ------------------------ |-----------|
-| Stability                | [beta]    |
-| Supported pipeline types | logs      |
-| Distributions            | [contrib] |
+| Status                   |             |
+| ------------------------ |-------------|
+| Stability                | [beta]      |
+| Supported pipeline types | logs,traces |
+| Distributions            | [contrib]   |
 
 This exporter supports sending OpenTelemetry logs to [Elasticsearch](https://www.elastic.co/elasticsearch).
 
@@ -20,7 +20,15 @@ This exporter supports sending OpenTelemetry logs to [Elasticsearch](https://www
 - `index`: The
   [index](https://www.elastic.co/guide/en/elasticsearch/reference/current/indices.html)
   or [datastream](https://www.elastic.co/guide/en/elasticsearch/reference/current/data-streams.html)
-  name to publish events to. The default value is `logs-generic-default`.
+  name to publish events to. The default value is `logs-generic-default`. Note: To better differentiate between log indexes and traces indexes, `index` option are deprecated and replaced with below `logs_index`
+- `logs_index`: The
+  [index](https://www.elastic.co/guide/en/elasticsearch/reference/current/indices.html)
+  or [datastream](https://www.elastic.co/guide/en/elasticsearch/reference/current/data-streams.html)
+  name to publish events to. The default value is `logs-generic-default`
+- `traces_index`: The
+  [index](https://www.elastic.co/guide/en/elasticsearch/reference/current/indices.html)
+  or [datastream](https://www.elastic.co/guide/en/elasticsearch/reference/current/data-streams.html)
+  name to publish traces to. The default value is `traces-generic-default`.
 - `pipeline` (optional): Optional [Ingest Node](https://www.elastic.co/guide/en/elasticsearch/reference/current/ingest.html)
   pipeline ID used for processing documents published by the exporter.
 - `flush`: Event bulk buffer flush settings
@@ -85,9 +93,23 @@ nodes will automatically be used for load balancing.
 
 ```yaml
 exporters:
-  elasticsearch:
-    endpoints:
-    - "https://localhost:9200"
+  elasticsearch/trace:
+    endpoints: [https://elastic.example.com:9200]
+    traces_index: trace_index
+  elasticsearch/log:
+    endpoints: [http://localhost:9200]
+    logs_index: my_log_index
+······
+service:
+  pipelines:
+    logs:
+      receivers: [otlp]
+      processors: [batch]
+      exporters: [elasticsearch/log]
+    traces:
+      receivers: [otlp]
+      exporters: [elasticsearch/trace]
+      processors: [batch]
 ```
 [beta]:https://github.com/open-telemetry/opentelemetry-collector#beta
 [contrib]:https://github.com/open-telemetry/opentelemetry-collector-releases/tree/main/distributions/otelcol-contrib

exporter/elasticsearchexporter/config.go

@@ -49,9 +49,15 @@ type Config struct {
 	// https://www.elastic.co/guide/en/elasticsearch/reference/current/indices.html
 	// https://www.elastic.co/guide/en/elasticsearch/reference/current/data-streams.html
 	//
-	// This setting is required.
+	// Deprecated: `index` is deprecated and replaced with `logs_index`.
 	Index string `mapstructure:"index"`
 
+	// This setting is required when logging pipelines used.
+	LogsIndex string `mapstructure:"logs_index"`
+
+	// This setting is required when traces pipelines used.
+	TracesIndex string `mapstructure:"traces_index"`
+
 	// Pipeline configures the ingest node pipeline name that should be used to process the
 	// events.
 	//
@@ -171,7 +177,6 @@ const (
 var (
 	errConfigNoEndpoint    = errors.New("endpoints or cloudid must be specified")
 	errConfigEmptyEndpoint = errors.New("endpoints must not include empty entries")
-	errConfigNoIndex       = errors.New("index must be specified")
 )
 
 func (m MappingMode) String() string {
@@ -217,10 +222,6 @@ func (cfg *Config) Validate() error {
 		}
 	}
 
-	if cfg.Index == "" {
-		return errConfigNoIndex
-	}
-
 	if _, ok := mappingModes[cfg.Mapping.Mode]; !ok {
 		return fmt.Errorf("unknown mapping mode %v", cfg.Mapping.Mode)
 	}

exporter/elasticsearchexporter/config_test.go

@@ -26,6 +26,56 @@ import (
 	"go.opentelemetry.io/collector/service/servicetest"
 )
 
+func TestLoad_DeprecatedIndexConfigOption(t *testing.T) {
+	factories, err := componenttest.NopFactories()
+	require.NoError(t, err)
+
+	factory := NewFactory()
+	factories.Exporters[typeStr] = factory
+	cfg, err := servicetest.LoadConfigAndValidate(filepath.Join("testdata", "config-use-deprecated-index_option.yaml"), factories)
+	require.NoError(t, err)
+	require.NotNil(t, cfg)
+
+	r1 := cfg.Exporters[config.NewComponentIDWithName(typeStr, "log")].(*Config)
+	assert.Equal(t, r1, &Config{
+		ExporterSettings: config.NewExporterSettings(config.NewComponentIDWithName(typeStr, "log")),
+		Endpoints:        []string{"http://localhost:9200"},
+		CloudID:          "TRNMxjXlNJEt",
+		Index:            "my_log_index",
+		LogsIndex:        "logs-generic-default",
+		TracesIndex:      "traces-generic-default",
+		Pipeline:         "mypipeline",
+		HTTPClientSettings: HTTPClientSettings{
+			Authentication: AuthenticationSettings{
+				User:     "elastic",
+				Password: "search",
+				APIKey:   "AvFsEiPs==",
+			},
+			Timeout: 2 * time.Minute,
+			Headers: map[string]string{
+				"myheader": "test",
+			},
+		},
+		Discovery: DiscoverySettings{
+			OnStart: true,
+		},
+		Flush: FlushSettings{
+			Bytes: 10485760,
+		},
+		Retry: RetrySettings{
+			Enabled:         true,
+			MaxRequests:     5,
+			InitialInterval: 100 * time.Millisecond,
+			MaxInterval:     1 * time.Minute,
+		},
+		Mapping: MappingsSettings{
+			Mode:  "ecs",
+			Dedup: true,
+			Dedot: true,
+		},
+	})
+}
+
 func TestLoadConfig(t *testing.T) {
 	factories, err := componenttest.NopFactories()
 	require.NoError(t, err)
@@ -36,19 +86,21 @@ func TestLoadConfig(t *testing.T) {
 	require.NoError(t, err)
 	require.NotNil(t, cfg)
 
-	assert.Equal(t, len(cfg.Exporters), 2)
+	assert.Equal(t, len(cfg.Exporters), 3)
 
 	defaultCfg := factory.CreateDefaultConfig()
 	defaultCfg.(*Config).Endpoints = []string{"https://elastic.example.com:9200"}
 	r0 := cfg.Exporters[config.NewComponentID(typeStr)]
 	assert.Equal(t, r0, defaultCfg)
 
-	r1 := cfg.Exporters[config.NewComponentIDWithName(typeStr, "customname")].(*Config)
+	r1 := cfg.Exporters[config.NewComponentIDWithName(typeStr, "trace")].(*Config)
 	assert.Equal(t, r1, &Config{
-		ExporterSettings: config.NewExporterSettings(config.NewComponentIDWithName(typeStr, "customname")),
+		ExporterSettings: config.NewExporterSettings(config.NewComponentIDWithName(typeStr, "trace")),
 		Endpoints:        []string{"https://elastic.example.com:9200"},
 		CloudID:          "TRNMxjXlNJEt",
-		Index:            "myindex",
+		Index:            "",
+		LogsIndex:        "logs-generic-default",
+		TracesIndex:      "trace_index",
 		Pipeline:         "mypipeline",
 		HTTPClientSettings: HTTPClientSettings{
 			Authentication: AuthenticationSettings{
@@ -79,6 +131,46 @@ func TestLoadConfig(t *testing.T) {
 			Dedot: true,
 		},
 	})
+
+	r2 := cfg.Exporters[config.NewComponentIDWithName(typeStr, "log")].(*Config)
+	assert.Equal(t, r2, &Config{
+		ExporterSettings: config.NewExporterSettings(config.NewComponentIDWithName(typeStr, "log")),
+		Endpoints:        []string{"http://localhost:9200"},
+		CloudID:          "TRNMxjXlNJEt",
+		Index:            "",
+		LogsIndex:        "my_log_index",
+		TracesIndex:      "traces-generic-default",
+		Pipeline:         "mypipeline",
+		HTTPClientSettings: HTTPClientSettings{
+			Authentication: AuthenticationSettings{
+				User:     "elastic",
+				Password: "search",
+				APIKey:   "AvFsEiPs==",
+			},
+			Timeout: 2 * time.Minute,
+			Headers: map[string]string{
+				"myheader": "test",
+			},
+		},
+		Discovery: DiscoverySettings{
+			OnStart: true,
+		},
+		Flush: FlushSettings{
+			Bytes: 10485760,
+		},
+		Retry: RetrySettings{
+			Enabled:         true,
+			MaxRequests:     5,
+			InitialInterval: 100 * time.Millisecond,
+			MaxInterval:     1 * time.Minute,
+		},
+		Mapping: MappingsSettings{
+			Mode:  "ecs",
+			Dedup: true,
+			Dedot: true,
+		},
+	})
+
 }
 
 func withDefaultConfig(fns ...func(*Config)) *Config {