Enable SPM in Jaeger v2

.github/workflows/ci-e2e-spm.yml

@@ -18,6 +18,14 @@ permissions:
 jobs:
   spm:
     runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        mode:
+        - name: v1
+          binary: all-in-one
+        - name: v2
+          binary: jaeger
+
     steps:
     - name: Harden Runner
       uses: step-security/harden-runner@17d0e2bd7d51742c71671bd19fa12bdc9d40a3d6 # v2.8.1
@@ -38,11 +46,6 @@ jobs:
 
     - name: Setup Node.js version
       uses: ./.github/actions/setup-node.js
-
-    - name: Temporary - only run the build
-      run:
-        cd docker-compose/monitor && make build
-
+
     - name: Run SPM Test
-      run: ./scripts/spm-integration-test.sh
-
+      run:  bash scripts/spm-integration-test.sh  -b ${{ matrix.mode.binary }}

cmd/jaeger/internal/components.go

@@ -6,6 +6,7 @@ package internal
 import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/connector/spanmetricsconnector"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/kafkaexporter"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/prometheusexporter"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/jaegerreceiver"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kafkareceiver"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/zipkinreceiver"
@@ -90,6 +91,7 @@ func (b builders) build() (otelcol.Factories, error) {
 		// add-ons
 		storageexporter.NewFactory(), // generic exporter to Jaeger v1 spanstore.SpanWriter
 		kafkaexporter.NewFactory(),
+		prometheusexporter.NewFactory(),
 		// elasticsearch.NewFactory(),
 	)
 	if err != nil {

cmd/jaeger/internal/exporters/storageexporter/exporter_test.go

@@ -39,10 +39,13 @@ import (
 )
 
 type mockStorageExt struct {
-	name    string
-	factory *factoryMocks.Factory
+	name           string
+	factory        *factoryMocks.Factory
+	metricsFactory *factoryMocks.MetricsFactory
 }
 
+var _ jaegerstorage.Extension = (*mockStorageExt)(nil)
+
 func (*mockStorageExt) Start(context.Context, component.Host) error {
 	panic("not implemented")
 }
@@ -51,13 +54,20 @@ func (*mockStorageExt) Shutdown(context.Context) error {
 	panic("not implemented")
 }
 
-func (m *mockStorageExt) Factory(name string) (storage.Factory, bool) {
+func (m *mockStorageExt) TraceStorageFactory(name string) (storage.Factory, bool) {
 	if m.name == name {
 		return m.factory, true
 	}
 	return nil, false
 }
 
+func (m *mockStorageExt) MetricStorageFactory(name string) (storage.MetricsFactory, bool) {
+	if m.name == name {
+		return m.metricsFactory, true
+	}
+	return nil, false
+}
+
 func TestExporterConfigError(t *testing.T) {
 	config := createDefaultConfig().(*Config)
 	err := config.Validate()

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

@@ -18,14 +18,12 @@ var _ component.ConfigValidator = (*Config)(nil)
 // Config represents the configuration for jaeger-query,
 type Config struct {
 	queryApp.QueryOptionsBase `mapstructure:",squash"`
-
-	TraceStoragePrimary string `valid:"required" mapstructure:"trace_storage"`
-	TraceStorageArchive string `valid:"optional" mapstructure:"trace_storage_archive"`
-
-	HTTP confighttp.ServerConfig `mapstructure:",squash"`
-	GRPC configgrpc.ServerConfig `mapstructure:",squash"`
-
-	Tenancy tenancy.Options `mapstructure:"multi_tenancy"`
+	TraceStoragePrimary       string                  `valid:"required" mapstructure:"trace_storage"`
+	TraceStorageArchive       string                  `valid:"optional" mapstructure:"trace_storage_archive"`
+	MetricStorage             string                  `valid:"optional" mapstructure:"metric_storage"`
+	HTTP                      confighttp.ServerConfig `mapstructure:",squash"`
+	GRPC                      configgrpc.ServerConfig `mapstructure:",squash"`
+	Tenancy                   tenancy.Options         `mapstructure:"multi_tenancy"`
 }
 
 func (cfg *Config) Validate() error {

cmd/jaeger/internal/extension/jaegerquery/server.go

@@ -18,6 +18,7 @@ import (
 	"github.com/jaegertracing/jaeger/pkg/telemetery"
 	"github.com/jaegertracing/jaeger/pkg/tenancy"
 	"github.com/jaegertracing/jaeger/plugin/metrics/disabled"
+	"github.com/jaegertracing/jaeger/storage/metricsstore"
 )
 
 var (
@@ -67,7 +68,12 @@ func (s *server) Start(_ context.Context, host component.Host) error {
 		return err
 	}
 	qs := querysvc.NewQueryService(spanReader, depReader, opts)
-	metricsQueryService, _ := disabled.NewMetricsReader()
+
+	mqs, err := s.createMetricReader(host)
+	if err != nil {
+		return err
+	}
+
 	tm := tenancy.NewManager(&s.config.Tenancy)
 
 	// TODO OTel-collector does not initialize the tracer currently
@@ -89,7 +95,7 @@ func (s *server) Start(_ context.Context, host component.Host) error {
 	s.server, err = queryApp.NewServer(
 		// TODO propagate healthcheck updates up to the collector's runtime
 		qs,
-		metricsQueryService,
+		mqs,
 		s.makeQueryOptions(),
 		tm,
 		telset,
@@ -122,6 +128,24 @@ func (s *server) addArchiveStorage(opts *querysvc.QueryServiceOptions, host comp
 	return nil
 }
 
+func (s *server) createMetricReader(host component.Host) (metricsstore.Reader, error) {
+	if s.config.MetricStorage == "" {
+		s.telset.Logger.Info("Metric storage not configured")
+		return disabled.NewMetricsReader()
+	}
+
+	mf, err := jaegerstorage.GetMetricsFactory(s.config.MetricStorage, host)
+	if err != nil {
+		return nil, fmt.Errorf("cannot find metrics storage factory: %w", err)
+	}
+
+	metricsReader, err := mf.CreateMetricsReader()
+	if err != nil {
+		return nil, fmt.Errorf("cannot create metrics reader %w", err)
+	}
+	return metricsReader, err
+}
+
 func (s *server) makeQueryOptions() *queryApp.QueryOptions {
 	return &queryApp.QueryOptions{
 		QueryOptionsBase: s.config.QueryOptionsBase,

cmd/jaeger/internal/extension/jaegerquery/server_test.go

@@ -26,6 +26,8 @@ import (
 	"github.com/jaegertracing/jaeger/storage"
 	"github.com/jaegertracing/jaeger/storage/dependencystore"
 	depsmocks "github.com/jaegertracing/jaeger/storage/dependencystore/mocks"
+	"github.com/jaegertracing/jaeger/storage/metricsstore"
+	metricsstoremocks "github.com/jaegertracing/jaeger/storage/metricsstore/mocks"
 	"github.com/jaegertracing/jaeger/storage/spanstore"
 	spanstoremocks "github.com/jaegertracing/jaeger/storage/spanstore/mocks"
 )
@@ -62,17 +64,43 @@ func (ff fakeFactory) Initialize(metrics.Factory, *zap.Logger) error {
 	return nil
 }
 
+type fakeMetricsFactory struct {
+	name string
+}
+
+// Initialize implements storage.MetricsFactory.
+func (fmf fakeMetricsFactory) Initialize(*zap.Logger) error {
+	if fmf.name == "need-initialize-error" {
+		return fmt.Errorf("test-error")
+	}
+	return nil
+}
+
+func (fmf fakeMetricsFactory) CreateMetricsReader() (metricsstore.Reader, error) {
+	if fmf.name == "need-metrics-reader-error" {
+		return nil, fmt.Errorf("test-error")
+	}
+	return &metricsstoremocks.Reader{}, nil
+}
+
 type fakeStorageExt struct{}
 
 var _ jaegerstorage.Extension = (*fakeStorageExt)(nil)
 
-func (fakeStorageExt) Factory(name string) (storage.Factory, bool) {
+func (fakeStorageExt) TraceStorageFactory(name string) (storage.Factory, bool) {
 	if name == "need-factory-error" {
 		return nil, false
 	}
 	return fakeFactory{name: name}, true
 }
 
+func (fakeStorageExt) MetricStorageFactory(name string) (storage.MetricsFactory, bool) {
+	if name == "need-factory-error" {
+		return nil, false
+	}
+	return fakeMetricsFactory{name: name}, true
+}
+
 func (fakeStorageExt) Start(context.Context, component.Host) error {
 	return nil
 }
@@ -105,6 +133,7 @@ func TestServerStart(t *testing.T) {
 			config: &Config{
 				TraceStorageArchive: "jaeger_storage",
 				TraceStoragePrimary: "jaeger_storage",
+				MetricStorage:       "jaeger_metrics_storage",
 			},
 		},
 		{
@@ -136,6 +165,22 @@ func TestServerStart(t *testing.T) {
 			},
 			expectedErr: "cannot find archive storage factory",
 		},
+		{
+			name: "metrics storage error",
+			config: &Config{
+				MetricStorage:       "need-factory-error",
+				TraceStoragePrimary: "jaeger_storage",
+			},
+			expectedErr: "cannot find metrics storage factory",
+		},
+		{
+			name: " metrics reader error",
+			config: &Config{
+				MetricStorage:       "need-metrics-reader-error",
+				TraceStoragePrimary: "jaeger_storage",
+			},
+			expectedErr: "cannot create metrics reader",
+		},
 	}
 
 	for _, tt := range tests {
@@ -242,3 +287,51 @@ func TestServerAddArchiveStorage(t *testing.T) {
 		})
 	}
 }
+
+func TestServerAddMetricsStorage(t *testing.T) {
+	host := componenttest.NewNopHost()
+
+	tests := []struct {
+		name           string
+		config         *Config
+		extension      component.Component
+		expectedOutput string
+		expectedErr    string
+	}{
+		{
+			name:           "Metrics storage unset",
+			config:         &Config{},
+			expectedOutput: `{"level":"info","msg":"Metric storage not configured"}` + "\n",
+			expectedErr:    "",
+		},
+		{
+			name: "Metrics storage set",
+			config: &Config{
+				MetricStorage: "random-value",
+			},
+			expectedOutput: "",
+			expectedErr:    "cannot find metrics storage factory: cannot find extension",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			logger, buf := testutils.NewLogger()
+			telemetrySettings := component.TelemetrySettings{
+				Logger: logger,
+			}
+			server := newServer(tt.config, telemetrySettings)
+			if tt.extension != nil {
+				host = storagetest.NewStorageHost().WithExtension(jaegerstorage.ID, tt.extension)
+			}
+			_, err := server.createMetricReader(host)
+			if tt.expectedErr == "" {
+				require.NoError(t, err)
+			} else {
+				require.ErrorContains(t, err, tt.expectedErr)
+			}
+
+			assert.Contains(t, buf.String(), tt.expectedOutput)
+		})
+	}
+}

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

@@ -13,6 +13,8 @@ import (
 
 	casCfg "github.com/jaegertracing/jaeger/pkg/cassandra/config"
 	esCfg "github.com/jaegertracing/jaeger/pkg/es/config"
+	promCfg "github.com/jaegertracing/jaeger/pkg/prometheus/config"
+	"github.com/jaegertracing/jaeger/plugin/metrics/prometheus"
 	"github.com/jaegertracing/jaeger/plugin/storage/badger"
 	"github.com/jaegertracing/jaeger/plugin/storage/cassandra"
 	"github.com/jaegertracing/jaeger/plugin/storage/es"
@@ -23,6 +25,7 @@ import (
 var (
 	_ component.ConfigValidator = (*Config)(nil)
 	_ confmap.Unmarshaler       = (*Backend)(nil)
+	_ confmap.Unmarshaler       = (*MetricBackends)(nil)
 )
 
 // Config contains configuration(s) for jaeger trace storage.
@@ -31,7 +34,8 @@ var (
 // We tried to alias this type directly to a map, but conf did not populated it correctly.
 // Note also that the Backend struct has a custom unmarshaler.
 type Config struct {
-	Backends map[string]Backend `mapstructure:"backends"`
+	Backends       map[string]Backend        `mapstructure:"backends"`
+	MetricBackends map[string]MetricBackends `mapstructure:"metric_backends"`
 }
 
 type Backend struct {
@@ -43,6 +47,10 @@ type Backend struct {
 	Opensearch    *esCfg.Configuration    `mapstructure:"opensearch"`
 }
 
+type MetricBackends struct {
+	Prometheus *promCfg.Configuration `mapstructure:"prometheus"`
+}
+
 // Unmarshal implements confmap.Unmarshaler. This allows us to provide
 // defaults for different configs. It cannot be done in createDefaultConfig()
 // because at that time we don't know which backends the user wants to use.
@@ -98,3 +106,12 @@ func (cfg *Config) Validate() error {
 	}
 	return nil
 }
+
+func (cfg *MetricBackends) Unmarshal(conf *confmap.Conf) error {
+	// apply defaults
+	if conf.IsSet("prometheus") {
+		v := prometheus.DefaultConfig()
+		cfg.Prometheus = &v
+	}
+	return conf.Unmarshal(cfg)
+}

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

@@ -107,3 +107,14 @@ backends:
 	require.NoError(t, conf.Unmarshal(cfg))
 	assert.NotEmpty(t, cfg.Backends["some_storage"].Opensearch.Servers)
 }
+
+func TestConfigDefaultPrometheus(t *testing.T) {
+	conf := loadConf(t, `
+metric_backends:
+  some_metrics_storage:
+    prometheus:
+`)
+	cfg := createDefaultConfig().(*Config)
+	require.NoError(t, conf.Unmarshal(cfg))
+	assert.NotEmpty(t, cfg.MetricBackends["some_metrics_storage"].Prometheus.ServerURL)
+}