[jaeger-v2] Add support for Elasticsearch

cmd/jaeger/Dockerfile.debug

@@ -0,0 +1,11 @@
+FROM alpine:latest
+
+WORKDIR /app
+
+COPY jaeger-linux-amd64 /app/
+
+RUN chmod +x /app/jaeger-linux-amd64
+
+EXPOSE 5775/udp 6831/udp 6832/udp 5778 16686 14268 14250 9200
+
+ENTRYPOINT ["/app/jaeger-linux-amd64"]

cmd/jaeger/docker-compose.yaml

@@ -0,0 +1,45 @@
+version: '3.7'
+
+services:
+  elasticsearch:
+    image: elasticsearch:8.12.0
+    container_name: elasticsearch
+    environment:
+      - discovery.type=single-node
+      - xpack.security.enabled=false
+    ports:
+      - "9200:9200"
+    healthcheck:
+      test: ["CMD-SHELL", "curl -s http://localhost:9200/_cluster/health | grep -vq '\"status\":\"red\"'"]
+      interval: 10s
+      timeout: 5s
+      retries: 5
+    networks:
+      - esnet
+
+  jaeger:
+    build:
+      context: ./
+      dockerfile: Dockerfile.debug
+    container_name: jaeger
+    ports:
+      - "16686:16686"
+      - "6831:6831/udp"
+      - "6832:6832/udp"
+      - "5778:5778"
+      - "14268:14268"
+      - "14250:14250"
+    depends_on:
+      elasticsearch:
+        condition: service_healthy
+    networks:
+      - esnet
+    volumes:
+      - ./es_config.yaml:/app/es_config.yaml
+      - ./config-ui.json:/app/cmd/jaeger/config-ui.json
+    command: ["--config", "/app/es_config.yaml"]
+    restart: always
+
+networks:
+  esnet:
+    driver: bridge

cmd/jaeger/es_config.yaml

@@ -0,0 +1,37 @@
+service:
+  extensions: [jaeger_storage, jaeger_query]
+  pipelines:
+    traces:
+      receivers: [otlp]
+      processors: [batch]
+      exporters: [jaeger_storage_exporter]
+
+extensions:
+  jaeger_query:
+    trace_storage: es_main
+    trace_storage_archive: es_archive
+    ui_config: ./cmd/jaeger/config-ui.json
+
+  jaeger_storage:
+    elasticsearch:
+      es_main:
+        server_urls: http://elasticsearch:9200
+        log_level: "error"
+        num_shards: 5
+      es_archive:
+        server_urls: http://elasticsearch:9200
+        log_level: "error"
+        index_prefix: "jaeger-archive"
+        num_shards: 5
+receivers:
+  otlp:
+    protocols:
+      grpc:
+      http:
+
+processors:
+  batch:
+
+exporters:
+  jaeger_storage_exporter:
+    trace_storage: es_main

cmd/jaeger/internal/extension/jaegerstorage/config.go

@@ -7,14 +7,16 @@ import (
 	"fmt"
 	"reflect"
 
+	esCfg "github.com/jaegertracing/jaeger/pkg/es/config"
 	memoryCfg "github.com/jaegertracing/jaeger/pkg/memory/config"
 	badgerCfg "github.com/jaegertracing/jaeger/plugin/storage/badger"
 )
 
 // Config has the configuration for jaeger-query,
 type Config struct {
-	Memory map[string]memoryCfg.Configuration   `mapstructure:"memory"`
-	Badger map[string]badgerCfg.NamespaceConfig `mapstructure:"badger"`
+	Memory        map[string]memoryCfg.Configuration   `mapstructure:"memory"`
+	Badger        map[string]badgerCfg.NamespaceConfig `mapstructure:"badger"`
+	Elasticsearch map[string]esCfg.Configuration       `mapstructure:"elasticsearch"`
 	// TODO add other storage types here
 	// TODO how will this work with 3rd party storage implementations?
 	//      Option: instead of looking for specific name, check interface.

cmd/jaeger/internal/extension/jaegerstorage/extension.go

@@ -13,10 +13,12 @@ import (
 	"go.opentelemetry.io/collector/extension"
 	"go.uber.org/zap"
 
+	esCfg "github.com/jaegertracing/jaeger/pkg/es/config"
 	memoryCfg "github.com/jaegertracing/jaeger/pkg/memory/config"
 	"github.com/jaegertracing/jaeger/pkg/metrics"
 	"github.com/jaegertracing/jaeger/plugin/storage/badger"
 	badgerCfg "github.com/jaegertracing/jaeger/plugin/storage/badger"
+	"github.com/jaegertracing/jaeger/plugin/storage/es"
 	"github.com/jaegertracing/jaeger/plugin/storage/memory"
 	"github.com/jaegertracing/jaeger/storage"
 )
@@ -107,10 +109,17 @@ func (s *storageExt) Start(ctx context.Context, host component.Host) error {
 		cfg:         s.config.Badger,
 		builder:     badger.NewFactoryWithConfig,
 	}
+	esStarter := &starter[esCfg.Configuration, *es.Factory]{
+		ext:         s,
+		storageKind: "elasticsearch",
+		cfg:         s.config.Elasticsearch,
+		builder:     es.NewFactoryWithConfig,
+	}
 
 	builders := []func(ctx context.Context, host component.Host) error{
 		memStarter.build,
 		badgerStarter.build,
+		esStarter.build,
 		// TODO add support for other backends
 	}
 	for _, builder := range builders {

cmd/jaeger/internal/extension/jaegerstorage/extension_test.go

@@ -6,6 +6,8 @@ package jaegerstorage
 import (
 	"context"
 	"fmt"
+	"net/http"
+	"net/http/httptest"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -16,6 +18,7 @@ import (
 	nooptrace "go.opentelemetry.io/otel/trace/noop"
 	"go.uber.org/zap"
 
+	esCfg "github.com/jaegertracing/jaeger/pkg/es/config"
 	memoryCfg "github.com/jaegertracing/jaeger/pkg/memory/config"
 	"github.com/jaegertracing/jaeger/pkg/metrics"
 	badgerCfg "github.com/jaegertracing/jaeger/plugin/storage/badger"
@@ -151,6 +154,46 @@ func TestBadgerStorageExtensionError(t *testing.T) {
 	require.ErrorContains(t, err, "/bad/path")
 }
 
+func TestESStorageExtension(t *testing.T) {
+	mockEsServerResponse := []byte(`
+	{
+		"Version": {
+			"Number": "6"
+		}
+	}
+	`)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Write(mockEsServerResponse)
+	}))
+	defer server.Close()
+	storageExtension := makeStorageExtenion(t, &Config{
+		Elasticsearch: map[string]esCfg.Configuration{
+			"foo": {
+				Servers:  []string{server.URL},
+				LogLevel: "error",
+			},
+		},
+	})
+	ctx := context.Background()
+	err := storageExtension.Start(ctx, componenttest.NewNopHost())
+	require.NoError(t, err)
+	require.NoError(t, storageExtension.Shutdown(ctx))
+}
+
+func TestESStorageExtensionError(t *testing.T) {
+	ext := makeStorageExtenion(t, &Config{
+		Elasticsearch: map[string]esCfg.Configuration{
+			"foo": {
+				Servers:  []string{"badurl"},
+				LogLevel: "error",
+			},
+		},
+	})
+	err := ext.Start(context.Background(), componenttest.NewNopHost())
+	require.ErrorContains(t, err, "failed to initialize elasticsearch storage")
+	require.ErrorContains(t, err, "no Elasticsearch node available")
+}
+
 func noopTelemetrySettings() component.TelemetrySettings {
 	return component.TelemetrySettings{
 		Logger:         zap.L(),

pkg/es/config/config.go

@@ -45,7 +45,7 @@ import (
 
 // Configuration describes the configuration properties needed to connect to an ElasticSearch cluster
 type Configuration struct {
-	Servers                        []string       `mapstructure:"server_urls"`
+	Servers                        []string       `mapstructure:"server_urls" validate:"required,min=1,dive,url"`
 	RemoteReadClusters             []string       `mapstructure:"remote_read_clusters"`
 	Username                       string         `mapstructure:"username"`
 	Password                       string         `mapstructure:"password" json:"-"`
@@ -54,13 +54,13 @@ type Configuration struct {
 	AllowTokenFromContext          bool           `mapstructure:"-"`
 	Sniffer                        bool           `mapstructure:"sniffer"` // https://github.com/olivere/elastic/wiki/Sniffing
 	SnifferTLSEnabled              bool           `mapstructure:"sniffer_tls_enabled"`
-	MaxDocCount                    int            `mapstructure:"-"`                     // Defines maximum number of results to fetch from storage per query
-	MaxSpanAge                     time.Duration  `yaml:"max_span_age" mapstructure:"-"` // configures the maximum lookback on span reads
-	NumShards                      int64          `yaml:"shards" mapstructure:"num_shards"`
-	NumReplicas                    int64          `yaml:"replicas" mapstructure:"num_replicas"`
-	PrioritySpanTemplate           int64          `yaml:"priority_span_template" mapstructure:"priority_span_template"`
-	PriorityServiceTemplate        int64          `yaml:"priority_service_template" mapstructure:"priority_service_template"`
-	PriorityDependenciesTemplate   int64          `yaml:"priority_dependencies_template" mapstructure:"priority_dependencies_template"`
+	MaxDocCount                    int            `mapstructure:"-"`
+	MaxSpanAge                     time.Duration  `mapstructure:"-"`
+	NumShards                      int64          `mapstructure:"num_shards"`
+	NumReplicas                    int64          `mapstructure:"num_replicas"`
+	PrioritySpanTemplate           int64          `mapstructure:"priority_span_template"`
+	PriorityServiceTemplate        int64          `mapstructure:"priority_service_template"`
+	PriorityDependenciesTemplate   int64          `mapstructure:"priority_dependencies_template"`
 	Timeout                        time.Duration  `validate:"min=500" mapstructure:"-"`
 	BulkSize                       int            `mapstructure:"-"`
 	BulkWorkers                    int            `mapstructure:"-"`

plugin/storage/es/factory.go

@@ -83,6 +83,24 @@ func NewFactory() *Factory {
 	}
 }
 
+func NewFactoryWithConfig(
+	cfg config.Configuration,
+	metricsFactory metrics.Factory,
+	logger *zap.Logger,
+) (*Factory, error) {
+	f := NewFactory()
+	cfg.MaxDocCount = defaultMaxDocCount
+	f.InitFromOptions(Options{
+		Primary: namespaceConfig{Configuration: cfg},
+		others:  make(map[string]*namespaceConfig),
+	})
+	err := f.Initialize(metricsFactory, logger)
+	if err != nil {
+		return nil, err
+	}
+	return f, nil
+}
+
 // AddFlags implements plugin.Configurable
 func (f *Factory) AddFlags(flagSet *flag.FlagSet) {
 	f.Options.AddFlags(flagSet)

plugin/storage/es/factory_test.go

@@ -263,6 +263,26 @@ func TestInitFromOptions(t *testing.T) {
 	assert.Equal(t, o.Get(archiveNamespace), f.archiveConfig)
 }
 
+func TestESStorageFactoryWithConfig(t *testing.T) {
+	cfg := escfg.Configuration{}
+	_, err := NewFactoryWithConfig(cfg, metrics.NullFactory, zap.NewNop())
+	require.Error(t, err)
+	require.ErrorContains(t, err, "failed to create primary Elasticsearch client")
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Write(mockEsServerResponse)
+	}))
+	defer server.Close()
+
+	cfg = escfg.Configuration{
+		Servers:  []string{server.URL},
+		LogLevel: "error",
+	}
+	factory, err := NewFactoryWithConfig(cfg, metrics.NullFactory, zap.NewNop())
+	require.NoError(t, err)
+	defer factory.Close()
+}
+
 func TestPasswordFromFile(t *testing.T) {
 	defer testutils.VerifyGoLeaksOnce(t)
 	t.Run("primary client", func(t *testing.T) {