Replace APIv2 with APIv3 client in e2e tests

Makefile.IntegrationTests.mk

@@ -14,8 +14,8 @@ all-in-one-integration-test:
 .PHONY: jaeger-v2-storage-integration-test
 jaeger-v2-storage-integration-test:
 	(cd cmd/jaeger/ && go build .)
-	# Expire tests results for jaeger storage integration tests since the environment might change
-	# even though the code remains the same.
+	# Expire tests results for jaeger storage integration tests since the environment
+	# might have changed even though the code remains the same.
 	go clean -testcache
 	bash -c "set -e; set -o pipefail; $(GOTEST) -coverpkg=./... -coverprofile $(COVEROUT) $(JAEGER_V2_STORAGE_PKGS) $(COLORIZE)"
 

cmd/jaeger/config-kafka-ingester.yaml

@@ -12,7 +12,7 @@ service:
       level: detailed
       address: 0.0.0.0:8888
     logs:
-      level: debug
+      level: info
 
 extensions:
   healthcheckv2:

cmd/jaeger/internal/integration/binary.go

@@ -0,0 +1,145 @@
+// Copyright (c) 2024 The Jaeger Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package integration
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/jaegertracing/jaeger/ports"
+)
+
+// Binary is a wrapper around exec.Cmd to help running binaries in tests.
+type Binary struct {
+	Name            string
+	HealthCheckPort int // overridable for Kafka tests which run two binaries and need different ports
+	exec.Cmd
+}
+
+func (b *Binary) Start(t *testing.T) {
+	outFile, err := os.OpenFile(
+		filepath.Join(t.TempDir(), "output_logs.txt"),
+		os.O_CREATE|os.O_WRONLY,
+		os.ModePerm,
+	)
+	require.NoError(t, err)
+	t.Logf("Writing the %s output logs into %s", b.Name, outFile.Name())
+
+	errFile, err := os.OpenFile(
+		filepath.Join(t.TempDir(), "error_logs.txt"),
+		os.O_CREATE|os.O_WRONLY,
+		os.ModePerm,
+	)
+	require.NoError(t, err)
+	t.Logf("Writing the %s error logs into %s", b.Name, errFile.Name())
+
+	b.Stdout = outFile
+	b.Stderr = errFile
+
+	t.Logf("Running command: %v", b.Args)
+	require.NoError(t, b.Cmd.Start())
+
+	t.Cleanup(func() {
+		if err := b.Process.Kill(); err != nil {
+			t.Errorf("Failed to kill %s process: %v", b.Name, err)
+		}
+		t.Logf("Waiting for %s to exit", b.Name)
+		b.Process.Wait()
+		t.Logf("%s exited", b.Name)
+		if t.Failed() {
+			b.dumpLogs(t, outFile, errFile)
+		}
+	})
+
+	// Wait for the binary to start and become ready to serve requests.
+	require.Eventually(t, func() bool { return b.doHealthCheck(t) },
+		60*time.Second, 3*time.Second, "%s did not start", b.Name)
+	t.Logf("%s is ready", b.Name)
+}
+
+func (b *Binary) doHealthCheck(t *testing.T) bool {
+	healthCheckPort := b.HealthCheckPort
+	if healthCheckPort == 0 {
+		healthCheckPort = ports.CollectorV2HealthChecks
+	}
+	healthCheckEndpoint := fmt.Sprintf("http://localhost:%d/status", healthCheckPort)
+	t.Logf("Checking if %s is available on %s", b.Name, healthCheckEndpoint)
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, healthCheckEndpoint, nil)
+	if err != nil {
+		t.Logf("HTTP request creation failed: %v", err)
+		return false
+	}
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Logf("HTTP request failed: %v", err)
+		return false
+	}
+	defer resp.Body.Close()
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		t.Logf("Failed to read HTTP response body: %v", err)
+		return false
+	}
+	if resp.StatusCode != http.StatusOK {
+		t.Logf("HTTP response not OK: %v", string(body))
+		return false
+	}
+
+	var healthResponse struct {
+		Status string `json:"status"`
+	}
+	if err := json.NewDecoder(bytes.NewReader(body)).Decode(&healthResponse); err != nil {
+		t.Logf("Failed to decode JSON response '%s': %v", string(body), err)
+		return false
+	}
+
+	// Check if the status field in the JSON is "StatusOK"
+	if healthResponse.Status != "StatusOK" {
+		t.Logf("Received non-OK status %s: %s", healthResponse.Status, string(body))
+		return false
+	}
+	return true
+}
+
+// Special Github markers to create a foldable section in the GitHub Actions runner output.
+// https://docs.github.com/en/actions/using-workflows/workflow-commands-for-github-actions#grouping-log-lines
+const (
+	githubBeginGroup = "::group::"
+	githubEndGroup   = "::endgroup::"
+)
+
+func (b *Binary) dumpLogs(t *testing.T, outFile, errFile *os.File) {
+	fmt.Printf(githubBeginGroup+" 🚧 🚧 🚧 %s binary logs\n", b.Name)
+	outLogs, err := os.ReadFile(outFile.Name())
+	if err != nil {
+		t.Errorf("Failed to read output logs: %v", err)
+	} else {
+		fmt.Printf("⏩⏩⏩ Start %s output logs:\n", b.Name)
+		fmt.Printf("%s\n", outLogs)
+		fmt.Printf("⏹️⏹️⏹️ End %s output logs.\n", b.Name)
+	}
+
+	errLogs, err := os.ReadFile(errFile.Name())
+	if err != nil {
+		t.Errorf("Failed to read error logs: %v", err)
+	} else {
+		fmt.Printf("⏩⏩⏩ Start %s error logs:\n", b.Name)
+		fmt.Printf("%s\n", errLogs)
+		fmt.Printf("⏹️⏹️⏹️ End %s error logs.\n", b.Name)
+	}
+	fmt.Println(githubEndGroup)
+}

cmd/jaeger/internal/integration/e2e_integration.go

@@ -4,17 +4,15 @@
 package integration
 
 import (
-	"bytes"
 	"context"
-	"encoding/json"
 	"fmt"
 	"io"
 	"net/http"
 	"os"
 	"os/exec"
 	"path/filepath"
+	"strings"
 	"testing"
-	"time"
 
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
@@ -24,7 +22,6 @@ import (
 	"github.com/jaegertracing/jaeger/cmd/jaeger/internal/integration/storagecleaner"
 	"github.com/jaegertracing/jaeger/plugin/storage/integration"
 	"github.com/jaegertracing/jaeger/ports"
-	"github.com/jaegertracing/jaeger/storage_v2/v1adapter"
 )
 
 const otlpPort = 4317
@@ -54,69 +51,10 @@ type E2EStorageIntegration struct {
 	EnvVarOverrides map[string]string
 }
 
-// Binary is a wrapper around exec.Cmd to help running binaries in tests.
-type Binary struct {
-	Name string
-	exec.Cmd
-}
-
-func (b *Binary) Start(t *testing.T) {
-	outFile, err := os.OpenFile(
-		filepath.Join(t.TempDir(), "output_logs.txt"),
-		os.O_CREATE|os.O_WRONLY,
-		os.ModePerm,
-	)
-	require.NoError(t, err)
-	t.Logf("Writing the %s output logs into %s", b.Name, outFile.Name())
-
-	errFile, err := os.OpenFile(
-		filepath.Join(t.TempDir(), "error_logs.txt"),
-		os.O_CREATE|os.O_WRONLY,
-		os.ModePerm,
-	)
-	require.NoError(t, err)
-	t.Logf("Writing the %s error logs into %s", b.Name, errFile.Name())
-
-	b.Stdout = outFile
-	b.Stderr = errFile
-
-	t.Logf("Running command: %v", b.Args)
-	require.NoError(t, b.Cmd.Start())
-	t.Cleanup(func() {
-		if err := b.Process.Kill(); err != nil {
-			t.Errorf("Failed to kill %s process: %v", b.Name, err)
-		}
-		if t.Failed() {
-			// A special annotation to create a foldable section in the GitHub Actions runner output.
-			// https://docs.github.com/en/actions/using-workflows/workflow-commands-for-github-actions#grouping-log-lines
-			fmt.Printf("::group::🚧 🚧 🚧 %s binary logs\n", b.Name)
-			outLogs, err := os.ReadFile(outFile.Name())
-			if err != nil {
-				t.Errorf("Failed to read output logs: %v", err)
-			} else {
-				fmt.Printf("🚧 🚧 🚧 %s output logs:\n%s", b.Name, outLogs)
-			}
-
-			errLogs, err := os.ReadFile(errFile.Name())
-			if err != nil {
-				t.Errorf("Failed to read error logs: %v", err)
-			} else {
-				fmt.Printf("🚧 🚧 🚧 %s error logs:\n%s", b.Name, errLogs)
-			}
-			// End of Github Actions foldable section annotation.
-			fmt.Println("::endgroup::")
-		}
-	})
-}
-
 // e2eInitialize starts the Jaeger-v2 collector with the provided config file,
 // it also initialize the SpanWriter and SpanReader below.
 // This function should be called before any of the tests start.
 func (s *E2EStorageIntegration) e2eInitialize(t *testing.T, storage string) {
-	// set environment variable overrides
-	for key, value := range s.EnvVarOverrides {
-		t.Setenv(key, value)
-	}
 	logger := zaptest.NewLogger(t, zaptest.WrapOptions(zap.AddCaller()))
 	if s.BinaryName == "" {
 		s.BinaryName = "jaeger-v2"
@@ -130,91 +68,41 @@ func (s *E2EStorageIntegration) e2eInitialize(t *testing.T, storage string) {
 	require.FileExists(t, configFile, "Config file does not exist at the resolved path")
 
 	t.Logf("Starting %s in the background with config file %s", s.BinaryName, configFile)
+	cfgBytes, err := os.ReadFile(configFile)
+	require.NoError(t, err)
+	t.Logf("Config file content:\n%s", string(cfgBytes))
+
+	envVars := []string{"OTEL_TRACES_SAMPLER=always_off"}
+	for key, value := range s.EnvVarOverrides {
+		envVars = append(envVars, fmt.Sprintf("%s=%s", key, value))
+	}
+
 	cmd := Binary{
-		Name: s.BinaryName,
+		Name:            s.BinaryName,
+		HealthCheckPort: s.HealthCheckPort,
 		Cmd: exec.Cmd{
 			Path: "./cmd/jaeger/jaeger",
 			Args: []string{"jaeger", "--config", configFile},
 			// Change the working directory to the root of this project
 			// since the binary config file jaeger_query's ui.config_file points to
 			// "./cmd/jaeger/config-ui.json"
 			Dir: "../../../..",
+			Env: envVars,
 		},
 	}
 	cmd.Start(t)
 
-	// Wait for the binary to start and become ready to serve requests.
-	require.Eventually(t, func() bool { return s.doHealthCheck(t) },
-		60*time.Second, 3*time.Second, "%s did not start", s.BinaryName)
-	t.Logf("%s is ready", s.BinaryName)
-
 	s.SpanWriter, err = createSpanWriter(logger, otlpPort)
 	require.NoError(t, err)
-	spanReader, err := createSpanReader(logger, ports.QueryGRPC)
+	s.TraceReader, err = createTraceReader(logger, ports.QueryGRPC)
 	require.NoError(t, err)
-	s.TraceReader = v1adapter.NewTraceReader(spanReader)
 
 	t.Cleanup(func() {
-		// Call e2eCleanUp to close the SpanReader and SpanWriter gRPC connection.
-		s.e2eCleanUp(t)
+		require.NoError(t, s.TraceReader.(io.Closer).Close())
+		require.NoError(t, s.SpanWriter.(io.Closer).Close())
 	})
 }
 
-func (s *E2EStorageIntegration) doHealthCheck(t *testing.T) bool {
-	healthCheckPort := s.HealthCheckPort
-	if healthCheckPort == 0 {
-		healthCheckPort = ports.CollectorV2HealthChecks
-	}
-	healthCheckEndpoint := fmt.Sprintf("http://localhost:%d/status", healthCheckPort)
-	t.Logf("Checking if %s is available on %s", s.BinaryName, healthCheckEndpoint)
-	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
-	defer cancel()
-	req, err := http.NewRequestWithContext(ctx, http.MethodGet, healthCheckEndpoint, nil)
-	if err != nil {
-		t.Logf("HTTP request creation failed: %v", err)
-		return false
-	}
-	resp, err := http.DefaultClient.Do(req)
-	if err != nil {
-		t.Logf("HTTP request failed: %v", err)
-		return false
-	}
-	defer resp.Body.Close()
-	body, err := io.ReadAll(resp.Body)
-	if err != nil {
-		t.Logf("Failed to read HTTP response body: %v", err)
-		return false
-	}
-	if resp.StatusCode != http.StatusOK {
-		t.Logf("HTTP response not OK: %v", string(body))
-		return false
-	}
-
-	var healthResponse struct {
-		Status string `json:"status"`
-	}
-	if err := json.NewDecoder(bytes.NewReader(body)).Decode(&healthResponse); err != nil {
-		t.Logf("Failed to decode JSON response '%s': %v", string(body), err)
-		return false
-	}
-
-	// Check if the status field in the JSON is "StatusOK"
-	if healthResponse.Status != "StatusOK" {
-		t.Logf("Received non-OK status %s: %s", healthResponse.Status, string(body))
-		return false
-	}
-	return true
-}
-
-// e2eCleanUp closes the SpanReader and SpanWriter gRPC connection.
-// This function should be called after all the tests are finished.
-func (s *E2EStorageIntegration) e2eCleanUp(t *testing.T) {
-	spanReader, err := v1adapter.GetV1Reader(s.TraceReader)
-	require.NoError(t, err)
-	require.NoError(t, spanReader.(io.Closer).Close())
-	require.NoError(t, s.SpanWriter.(io.Closer).Close())
-}
-
 func createStorageCleanerConfig(t *testing.T, configFile string, storage string) string {
 	data, err := os.ReadFile(configFile)
 	require.NoError(t, err)
@@ -261,7 +149,9 @@ func createStorageCleanerConfig(t *testing.T, configFile string, storage string)
 
 	newData, err := yaml.Marshal(config)
 	require.NoError(t, err)
-	tempFile := filepath.Join(t.TempDir(), "storageCleaner_config.yaml")
+	fileExt := filepath.Ext(filepath.Base(configFile))
+	fileName := strings.TrimSuffix(filepath.Base(configFile), fileExt)
+	tempFile := filepath.Join(t.TempDir(), fileName+"_with_storageCleaner"+fileExt)
 	err = os.WriteFile(tempFile, newData, 0o600)
 	require.NoError(t, err)