Merge branch 'main' into OTEL_Config/Telemetry

.github/workflows/ci-e2e-memory.yaml

@@ -0,0 +1,41 @@
+name: CIT Memory
+
+on:
+  push:
+    branches: [main]
+
+  pull_request:
+    branches: [main]
+
+concurrency:
+  group: ${{ github.workflow }}-${{ (github.event.pull_request && github.event.pull_request.number) || github.ref || github.run_id }}
+  cancel-in-progress: true
+
+# See https://github.com/ossf/scorecard/blob/main/docs/checks.md#token-permissions
+permissions:  # added using https://github.com/step-security/secure-workflows
+  contents: read
+
+jobs:
+  memory-v2:
+    runs-on: ubuntu-latest
+    steps:
+    - name: Harden Runner
+      uses: step-security/harden-runner@17d0e2bd7d51742c71671bd19fa12bdc9d40a3d6 # v2.8.1
+      with:
+        egress-policy: audit # TODO: change to 'egress-policy: block' after couple of runs
+
+    - uses: actions/checkout@692973e3d937129bcbf40652eb9f2f61becf3332 # v4.1.7
+
+    - uses: actions/setup-go@0a12ed9d6a96ab950c8f026ed9f722fe0da7ef32 # v5.0.2
+      with:
+        go-version: 1.22.x
+
+    - name: Run Memory storage integration tests
+      run: |
+        STORAGE=memory_v2 make jaeger-v2-storage-integration-test
+
+    - name: Upload coverage to codecov
+      uses: ./.github/actions/upload-codecov
+      with:
+        files: cover.out
+        flags: memory_v2

cmd/jaeger/internal/integration/e2e_integration.go

@@ -177,6 +177,7 @@ func createStorageCleanerConfig(t *testing.T, configFile string, storage string)
 
 func purge(t *testing.T) {
 	addr := fmt.Sprintf("http://0.0.0.0:%s%s", storagecleaner.Port, storagecleaner.URL)
+	t.Logf("Purging storage via %s", addr)
 	r, err := http.NewRequestWithContext(context.Background(), http.MethodPost, addr, nil)
 	require.NoError(t, err)
 
@@ -185,6 +186,8 @@ func purge(t *testing.T) {
 	resp, err := client.Do(r)
 	require.NoError(t, err)
 	defer resp.Body.Close()
+	body, err := io.ReadAll(resp.Body)
+	require.NoError(t, err)
 
-	require.Equal(t, http.StatusOK, resp.StatusCode)
+	require.Equal(t, http.StatusOK, resp.StatusCode, "body: %s", string(body))
 }

cmd/jaeger/internal/integration/es_test.go → cmd/jaeger/internal/integration/elasticsearch_test.go

@@ -9,7 +9,7 @@ import (
 	"github.com/jaegertracing/jaeger/plugin/storage/integration"
 )
 
-func TestESStorage(t *testing.T) {
+func TestElasticsearchStorage(t *testing.T) {
 	integration.SkipUnlessEnv(t, "elasticsearch")
 
 	s := &E2EStorageIntegration{

cmd/jaeger/internal/integration/memory_test.go

@@ -0,0 +1,27 @@
+// Copyright (c) 2024 The Jaeger Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package integration
+
+import (
+	"testing"
+
+	"github.com/jaegertracing/jaeger/plugin/storage/integration"
+)
+
+func TestMemoryStorage(t *testing.T) {
+	integration.SkipUnlessEnv(t, "memory_v2")
+
+	s := &E2EStorageIntegration{
+		ConfigFile: "../../config.yaml",
+		StorageIntegration: integration.StorageIntegration{
+			SkipArchiveTest: true,
+			CleanUp:         purge,
+		},
+	}
+	s.e2eInitialize(t, "memory")
+	t.Cleanup(func() {
+		s.e2eCleanUp(t)
+	})
+	s.RunAll(t)
+}

cmd/jaeger/internal/integration/os_test.go → cmd/jaeger/internal/integration/opensearch_test.go

@@ -9,7 +9,7 @@ import (
 	"github.com/jaegertracing/jaeger/plugin/storage/integration"
 )
 
-func TestOSStorage(t *testing.T) {
+func TestOpenSearchStorage(t *testing.T) {
 	integration.SkipUnlessEnv(t, "opensearch")
 	s := &E2EStorageIntegration{
 		ConfigFile: "../../config-opensearch.yaml",

cmd/jaeger/internal/integration/storagecleaner/extension.go

@@ -13,6 +13,7 @@ import (
 	"github.com/gorilla/mux"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/extension"
+	"go.uber.org/zap"
 
 	"github.com/jaegertracing/jaeger/cmd/jaeger/internal/extension/jaegerstorage"
 	"github.com/jaegertracing/jaeger/storage"
@@ -29,15 +30,15 @@ const (
 )
 
 type storageCleaner struct {
-	config   *Config
-	server   *http.Server
-	settings component.TelemetrySettings
+	config *Config
+	server *http.Server
+	telset component.TelemetrySettings
 }
 
-func newStorageCleaner(config *Config, telemetrySettings component.TelemetrySettings) *storageCleaner {
+func newStorageCleaner(config *Config, telset component.TelemetrySettings) *storageCleaner {
 	return &storageCleaner{
-		config:   config,
-		settings: telemetrySettings,
+		config: config,
+		telset: telset,
 	}
 }
 
@@ -74,10 +75,11 @@ func (c *storageCleaner) Start(_ context.Context, host component.Host) error {
 		Handler:           r,
 		ReadHeaderTimeout: 3 * time.Second,
 	}
+	c.telset.Logger.Info("Starting storage cleaner server", zap.String("addr", c.server.Addr))
 	go func() {
 		if err := c.server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
 			err = fmt.Errorf("error starting cleaner server: %w", err)
-			c.settings.ReportStatus(component.NewFatalErrorEvent(err))
+			c.telset.ReportStatus(component.NewFatalErrorEvent(err))
 		}
 	}()
 

cmd/jaeger/internal/integration/storagecleaner/extension_test.go

@@ -15,6 +15,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/component"
+	"go.uber.org/zap/zaptest"
 
 	"github.com/jaegertracing/jaeger/cmd/jaeger/internal/extension/jaegerstorage"
 	"github.com/jaegertracing/jaeger/storage"
@@ -84,7 +85,9 @@ func TestStorageCleanerExtension(t *testing.T) {
 				TraceStorage: "storage",
 				Port:         Port,
 			}
-			s := newStorageCleaner(config, component.TelemetrySettings{})
+			s := newStorageCleaner(config, component.TelemetrySettings{
+				Logger: zaptest.NewLogger(t),
+			})
 			require.NotEmpty(t, s.Dependencies())
 			host := storagetest.NewStorageHost()
 			host.WithExtension(jaegerstorage.ID, &mockStorageExt{
@@ -110,7 +113,9 @@ func TestStorageCleanerExtension(t *testing.T) {
 
 func TestGetStorageFactoryError(t *testing.T) {
 	config := &Config{}
-	s := newStorageCleaner(config, component.TelemetrySettings{})
+	s := newStorageCleaner(config, component.TelemetrySettings{
+		Logger: zaptest.NewLogger(t),
+	})
 	host := storagetest.NewStorageHost()
 	host.WithExtension(jaegerstorage.ID, &mockStorageExt{
 		name:    "storage",
@@ -128,6 +133,7 @@ func TestStorageExtensionStartError(t *testing.T) {
 	}
 	var startStatus atomic.Pointer[component.StatusEvent]
 	s := newStorageCleaner(config, component.TelemetrySettings{
+		Logger: zaptest.NewLogger(t),
 		ReportStatus: func(status *component.StatusEvent) {
 			startStatus.Store(status)
 		},

plugin/storage/memory/factory.go

@@ -16,6 +16,7 @@
 package memory
 
 import (
+	"context"
 	"flag"
 
 	"github.com/spf13/viper"
@@ -36,6 +37,7 @@ var ( // interface comformance checks
 	_ storage.ArchiveFactory       = (*Factory)(nil)
 	_ storage.SamplingStoreFactory = (*Factory)(nil)
 	_ plugin.Configurable          = (*Factory)(nil)
+	_ storage.Purger               = (*Factory)(nil)
 )
 
 // Factory implements storage.Factory and creates storage components backed by memory store.
@@ -126,3 +128,10 @@ func (*Factory) CreateLock() (distributedlock.Lock, error) {
 func (f *Factory) publishOpts() {
 	safeexpvar.SetInt("jaeger_storage_memory_max_traces", int64(f.options.Configuration.MaxTraces))
 }
+
+// Purge removes all data from the Factory's underlying Memory store.
+// This function is intended for testing purposes only and should not be used in production environments.
+func (f *Factory) Purge(ctx context.Context) error {
+	f.store.purge(ctx)
+	return nil
+}

plugin/storage/memory/memory.go

@@ -337,3 +337,10 @@ func flattenTags(span *model.Span) model.KeyValues {
 	}
 	return retMe
 }
+
+// purge supports Purger interface.
+func (st *Store) purge(context.Context) {
+	st.Lock()
+	st.perTenant = make(map[string]*Tenant)
+	st.Unlock()
+}