fix: obsy tests (#578)

* fix: obsy tests

* chore: option to add logging to the otel exporters

e2e/basic/observability_test.go

@@ -5,6 +5,8 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"regexp"
+	"strings"
 	"time"
 
 	"github.com/celestiaorg/knuu/pkg/sidecars/observability"
@@ -14,19 +16,26 @@ const (
 	prometheusPort   = observability.DefaultOtelMetricsPort
 	prometheusImage  = "prom/prometheus:latest"
 	prometheusConfig = "/etc/prometheus/prometheus.yml"
-	prometheusArgs   = "--config.file=/etc/prometheus/prometheus.yml"
+	prometheusArgs   = "--config.file=" + prometheusConfig
 
 	curlImage = "curlimages/curl:latest"
 	otlpPort  = observability.DefaultOtelOtlpPort
+
+	retryInterval = 1 * time.Second
+	retryTimeout  = 10 * time.Second
 )
 
 // TestObservabilityCollector is a test function that verifies the functionality of the otel collector setup
 func (s *Suite) TestObservabilityCollector() {
 	const (
-		namePrefix         = "observability"
-		targetStartCommand = "while true; do curl -X POST http://localhost:8888/v1/traces; sleep 5; done"
+		namePrefix             = "observability"
+		scrapeInterval         = "2s"
+		prometheusQueryTimeout = 30 * time.Second
+	)
+	var (
+		targetStartCommand = fmt.Sprintf("while true; do curl -X POST http://localhost:%d/v1/traces; sleep 2; done", otlpPort)
+		ctx                = context.Background()
 	)
-	ctx := context.Background()
 
 	// Setup Prometheus
 	prometheus, err := s.Knuu.NewInstance(namePrefix + "-prometheus")
@@ -44,12 +53,12 @@ func (s *Suite) TestObservabilityCollector() {
 	// Add Prometheus config file
 	prometheusConfigContent := fmt.Sprintf(`
 global:
-  scrape_interval: '10s'
+  scrape_interval: '%s'
+
 scrape_configs:
   - job_name: 'otel-collector'
     static_configs:
-      - targets: ['otel-collector:%d']
-`, otlpPort)
+      - targets: ['otel-collector:%d']`, scrapeInterval, otlpPort)
 	s.Require().NoError(prometheus.Storage().AddFileBytes([]byte(prometheusConfigContent), prometheusConfig, "0:0"))
 
 	s.Require().NoError(prometheus.Build().SetArgs(prometheusArgs))
@@ -59,12 +68,12 @@ scrape_configs:
 	observabilitySidecar := observability.New()
 
 	s.Require().NoError(observabilitySidecar.SetOtelEndpoint(4318))
-	s.Require().NoError(observabilitySidecar.SetPrometheusEndpoint(otlpPort, fmt.Sprintf("knuu-%s", s.Knuu.Scope), "10s"))
+	s.Require().NoError(observabilitySidecar.SetPrometheusEndpoint(otlpPort, fmt.Sprintf("knuu-%s", s.Knuu.Scope), scrapeInterval))
 	s.Require().NoError(observabilitySidecar.SetJaegerEndpoint(14250, 6831, 14268))
 	s.Require().NoError(observabilitySidecar.SetOtlpExporter("prometheus:9090", "", ""))
 
 	// Create and start a target pod and configure it to use the obsySidecar to push metrics
-	target, err := s.Knuu.NewInstance(namePrefix + "target")
+	target, err := s.Knuu.NewInstance(namePrefix + "-target")
 	s.Require().NoError(err)
 
 	s.Require().NoError(target.Build().SetImage(ctx, curlImage))
@@ -73,30 +82,87 @@ scrape_configs:
 	s.Require().NoError(err)
 
 	s.Require().NoError(target.Sidecars().Add(ctx, observabilitySidecar))
+
 	s.Require().NoError(target.Build().Commit(ctx))
 	s.Require().NoError(target.Execution().Start(ctx))
 
-	// Wait for the target pod to push data to the otel collector
-	s.T().Log("Waiting one minute for the target pod to push data to the otel collector")
-	time.Sleep(1 * time.Minute)
-
 	// Verify that data has been pushed to Prometheus
+	s.Require().Eventually(func() bool {
+		url := fmt.Sprintf("%s/api/v1/query?query=up", prometheusEndpoint)
+		ctx, cancel := context.WithTimeout(context.Background(), prometheusQueryTimeout)
+		defer cancel()
+
+		req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
+		if err != nil {
+			s.T().Logf("Error creating request: %v", err)
+			return false
+		}
+
+		resp, err := http.DefaultClient.Do(req)
+		if err != nil {
+			s.T().Logf("Error sending request: %v", err)
+			return false
+		}
+		if resp.StatusCode != http.StatusOK {
+			s.T().Logf("Prometheus API returned status code: %d", resp.StatusCode)
+			return false
+		}
+
+		defer resp.Body.Close()
+		body, err := io.ReadAll(resp.Body)
+		if err != nil {
+			s.T().Logf("Error reading response body: %v", err)
+			return false
+		}
+		return strings.Contains(string(body), "otel-collector")
+
+	}, retryTimeout, retryInterval, "otel-collector data source not found in Prometheus")
+}
 
-	prometheusURL := fmt.Sprintf("%s/api/v1/query?query=up", prometheusEndpoint)
-	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
-	defer cancel()
+func (s *Suite) TestObservabilityCollectorWithLogging() {
+	const (
+		namePrefix         = "observability"
+		targetStartCommand = "while true; do curl -X POST http://localhost:8888/v1/traces; sleep 2; done"
+	)
+	ctx := context.Background()
 
-	req, err := http.NewRequestWithContext(ctx, "GET", prometheusURL, nil)
-	s.Require().NoError(err)
+	// Setup obsySidecar collector
+	obsySidecar := observability.New()
+
+	s.Require().NoError(obsySidecar.SetOtelEndpoint(4318))
+	s.Require().NoError(obsySidecar.SetLoggingExporter("debug"))
 
-	resp, err := http.DefaultClient.Do(req)
+	// Create and start a target pod and configure it to use the obsySidecar to push metrics
+	target, err := s.Knuu.NewInstance(namePrefix + "target")
 	s.Require().NoError(err)
-	s.Require().Equal(http.StatusOK, resp.StatusCode, "Prometheus API is not accessible")
 
-	defer resp.Body.Close()
-	body, err := io.ReadAll(resp.Body)
+	s.Require().NoError(target.Build().SetImage(ctx, curlImage))
+
+	err = target.Build().SetStartCommand("sh", "-c", targetStartCommand)
 	s.Require().NoError(err)
-	s.Require().Contains(string(body), "otel-collector", "otel-collector data source not found in Prometheus")
 
-	s.T().Log("otel-collector data source is available in Prometheus")
+	s.Require().NoError(target.Sidecars().Add(ctx, obsySidecar))
+	s.Require().NoError(target.Build().Commit(ctx))
+	s.Require().NoError(target.Execution().Start(ctx))
+
+	// Verify that data has been pushed to the logging exporter
+	s.Require().Eventually(func() bool {
+		logsReader, err := obsySidecar.Instance().Monitoring().Logs(ctx)
+		if err != nil {
+			s.T().Logf("Error getting logs: %v", err)
+			return false
+		}
+		logsOutput, err := io.ReadAll(logsReader)
+		if err != nil {
+			s.T().Logf("Error reading logs: %v", err)
+			return false
+		}
+
+		loggingExporterPattern := regexp.MustCompile(`"kind": "exporter", "data_type": "metrics", "name": "logging"`)
+		if !loggingExporterPattern.Match(logsOutput) {
+			s.T().Logf("Logging exporter not found in the logs: `%s`", string(logsOutput))
+			return false
+		}
+		return true
+	}, retryTimeout, retryInterval, "Logging exporter not found in the logs")
 }

e2e/system/build_image_test.go

@@ -100,7 +100,7 @@ func (s *Suite) TestBuildWithBuildArgs() {
 		// This file is created by the dockerfile in the repo
 		// ref: https://github.com/celestiaorg/knuu/blob/test/build-from-git/Dockerfile
 		filePath     = "/test.txt"
-		expectedData = "Hello, build arg!"
+		expectedData = "Hello, World!"
 	)
 
 	s.T().Log("Creating new instance")

pkg/sidecars/observability/helpers.go

@@ -86,6 +86,19 @@ func (o *Obsy) SetPrometheusRemoteWriteExporter(endpoint string) error {
 	return nil
 }
 
+// SetLoggingExporter sets the logging exporter for the instance
+func (o *Obsy) SetLoggingExporter(logLevel string) error {
+	if err := o.validateStateForObsy("Logging exporter"); err != nil {
+		return err
+	}
+
+	if logLevel == "" {
+		logLevel = defaultLoggingExporterLogLevel
+	}
+	o.obsyConfig.loggingExporterLogLevel = logLevel
+	return nil
+}
+
 func (o *Obsy) validateStateForObsy(endpoint string) error {
 	if o.instance != nil && !o.instance.IsInState(instance.StateNone) {
 		return ErrSettingNotAllowed.WithParams(endpoint, o.instance.State().String())

pkg/sidecars/observability/obsy.go

@@ -75,6 +75,9 @@ type ObsyConfig struct {
 
 	// prometheusRemoteWriteExporterEndpoint is the endpoint of the prometheus remote write
 	prometheusRemoteWriteExporterEndpoint string
+
+	// loggingExporterLogLevel is the log level for the logging exporter
+	loggingExporterLogLevel string
 }
 
 func New() *Obsy {

pkg/sidecars/observability/otel.go

@@ -22,6 +22,8 @@ const (
 	jaegerExporterName                = "jaeger"
 	prometheusExporterName            = "prometheus"
 	prometheusRemoteWriteExporterName = "prometheusremotewrite"
+	loggingExporterName               = "logging"
+	defaultLoggingExporterLogLevel    = "debug"
 	attributesProcessorName           = "attributes"
 
 	scopeAttributeKey = "scope"
@@ -112,6 +114,7 @@ type Exporters struct {
 	Jaeger                JaegerExporter                `yaml:"jaeger,omitempty"`
 	Prometheus            PrometheusExporter            `yaml:"prometheus,omitempty"`
 	PrometheusRemoteWrite PrometheusRemoteWriteExporter `yaml:"prometheusremotewrite,omitempty"`
+	Logging               LoggingExporter               `yaml:"logging,omitempty"`
 }
 
 type OTLPHTTPExporter struct {
@@ -137,6 +140,10 @@ type PrometheusRemoteWriteExporter struct {
 	TLS      TLS    `yaml:"tls,omitempty"`
 }
 
+type LoggingExporter struct {
+	LogLevel string `yaml:"loglevel,omitempty"`
+}
+
 type TLS struct {
 	Insecure bool `yaml:"insecure,omitempty"`
 }
@@ -319,6 +326,12 @@ func (o *Obsy) createPrometheusRemoteWriteExporter() PrometheusRemoteWriteExport
 	}
 }
 
+func (o *Obsy) createLoggingExporter() LoggingExporter {
+	return LoggingExporter{
+		LogLevel: o.obsyConfig.loggingExporterLogLevel,
+	}
+}
+
 func (o *Obsy) createExporters() Exporters {
 	exporters := Exporters{}
 
@@ -338,6 +351,10 @@ func (o *Obsy) createExporters() Exporters {
 		exporters.PrometheusRemoteWrite = o.createPrometheusRemoteWriteExporter()
 	}
 
+	if o.obsyConfig.loggingExporterLogLevel != "" {
+		exporters.Logging = o.createLoggingExporter()
+	}
+
 	return exporters
 }
 
@@ -358,7 +375,16 @@ func (o *Obsy) prepareMetricsForServicePipeline() Metrics {
 	if o.obsyConfig.prometheusRemoteWriteExporterEndpoint != "" {
 		metrics.Exporters = append(metrics.Exporters, prometheusRemoteWriteExporterName)
 	}
+	if o.obsyConfig.loggingExporterLogLevel != "" {
+		metrics.Exporters = append(metrics.Exporters, loggingExporterName)
+	}
 	metrics.Processors = []string{attributesProcessorName}
+
+	// if no metrics receiver or exporter is added, remove any metrics pipeline
+	if len(metrics.Receivers) == 0 || len(metrics.Exporters) == 0 {
+		metrics = Metrics{}
+	}
+
 	return metrics
 }
 
@@ -376,7 +402,16 @@ func (o *Obsy) prepareTracesForServicePipeline() Traces {
 	if o.obsyConfig.jaegerEndpoint != "" {
 		traces.Exporters = append(traces.Exporters, jaegerExporterName)
 	}
+	if o.obsyConfig.loggingExporterLogLevel != "" {
+		traces.Exporters = append(traces.Exporters, loggingExporterName)
+	}
 	traces.Processors = []string{attributesProcessorName}
+
+	// if no trace receiver or exporter is added, remove any trace pipeline
+	if len(traces.Receivers) == 0 || len(traces.Exporters) == 0 {
+		traces = Traces{}
+	}
+
 	return traces
 }