Add support for Opensearch client span

Signed-off-by: Giuseppe Ognibene <[email protected]>

Author: pinoOgni

internal/test/integration/components/elasticsearch/main.py

@@ -1,4 +1,4 @@
-from fastapi import FastAPI, HTTPException
+from fastapi import FastAPI, HTTPException, Request
 import os
 import uvicorn
 import requests
@@ -8,34 +8,44 @@
 
 app = FastAPI()
 HEADERS = {'Content-Type': 'application/json'}
-ELASTICSEARCH_HOST = "http://elasticsearchserver:9200"
 
 @app.get("/health")
-async def health():
-    HEALTH_URL = ELASTICSEARCH_HOST + "/_cluster/health"
-    
+async def health(request: Request):
+    host_url = request.query_params.get("host_url")
+    if host_url is None:
+        raise HTTPException(
+            status_code=400, 
+            detail="The 'host_url' query parameter is required."
+        )
+    SERVER_URL = host_url + "/_cluster/health"
     try:
-        response = requests.get(HEALTH_URL, timeout=5)
+        response = requests.get(SERVER_URL, timeout=5)
         response.raise_for_status() 
         status = response.json().get("status", "red")
 
         if status in ("red","yellow"):
             raise HTTPException(
                 status_code=503, 
-                detail={"status": "red","message": "Elasticsearch cluster unhealthy"})
-        return {"status": status, "message": "Elasticsearch cluster healthy"}
+                detail={"status": "red","message": "Cluster unhealthy"})
+        return {"status": status, "message": "Cluster healthy"}
 
     except requests.RequestException as e:
         raise HTTPException(
             status_code=503, 
-            detail={"status": "error","message": f"Cannot reach Elasticsearch cluster: {str(e)}"})
+            detail={"status": "error","message": f"Cannot reach Cluster: {str(e)}"})
 
 @app.get("/doc")
-async def doc():
-    ELASTICSEARCH_URL = ELASTICSEARCH_HOST + "/test_index/_doc/1"
+async def doc(request: Request):
+    host_url = request.query_params.get("host_url")
+    if host_url is None:
+        raise HTTPException(
+            status_code=400, 
+            detail="The 'host_url' query parameter is required."
+        )
+    SERVER_URL = host_url + "/test_index/_doc/1"
 
     try:
-        response = requests.get(ELASTICSEARCH_URL, headers=HEADERS)
+        response = requests.get(SERVER_URL, headers=HEADERS)
 
     except Exception as e:
         print(json.dumps({"error": str(e)}))
@@ -44,8 +54,15 @@ async def doc():
 
 
 @app.get("/search")
-async def search():
-    ELASTICSEARCH_URL = ELASTICSEARCH_HOST + "/test_index/_search"
+async def search(request: Request):
+    host_url = request.query_params.get("host_url")
+    if host_url is None:
+        raise HTTPException(
+            status_code=400, 
+            detail="The 'host_url' query parameter is required."
+        )
+    SERVER_URL = host_url + "/test_index/_search"
+    print(SERVER_URL)
     query_body = {
         "query": {
             "match": {
@@ -54,16 +71,22 @@ async def search():
             }
         }
     try:
-        response = requests.post(ELASTICSEARCH_URL, json=query_body, headers=HEADERS)
+        response = requests.post(SERVER_URL, json=query_body, headers=HEADERS)
 
     except Exception as e:
         print(json.dumps({"error": str(e)}))
         sys.exit(1)
     return {"status": "OK"}
 
 @app.get("/msearch")
-async def msearch():
-    ELASTICSEARCH_URL = ELASTICSEARCH_HOST + "/_msearch"
+async def msearch(request: Request):
+    host_url = request.query_params.get("host_url")
+    if host_url is None:
+        raise HTTPException(
+            status_code=400, 
+            detail="The 'host_url' query parameter is required."
+        )
+    SERVER_URL = host_url + "/_msearch"
     searches = [
         {},
         {
@@ -83,7 +106,7 @@ async def msearch():
         }
     ]
     try:
-        response = requests.post(ELASTICSEARCH_URL, json=searches, headers=HEADERS)
+        response = requests.post(SERVER_URL, json=searches, headers=HEADERS)
 
     except Exception as e:
         print(json.dumps({"error": str(e)}))
@@ -92,8 +115,14 @@ async def msearch():
 
 
 @app.get("/bulk")
-async def bulk():
-    ELASTICSEARCH_URL = ELASTICSEARCH_HOST + "/_bulk"
+async def bulk(request: Request):
+    host_url = request.query_params.get("host_url")
+    if host_url is None:
+        raise HTTPException(
+            status_code=400, 
+            detail="The 'host_url' query parameter is required."
+        )
+    SERVER_URL = host_url + "/_bulk"
     actions=[
         {
             "index": {
@@ -132,7 +161,7 @@ async def bulk():
         }
     ]
     try:
-        response = requests.post(ELASTICSEARCH_URL, json=actions, headers=HEADERS)
+        response = requests.post(SERVER_URL, json=actions, headers=HEADERS)
 
     except Exception as e:
         print(json.dumps({"error": str(e)}))

internal/test/integration/docker-compose-opensearch.yml

@@ -0,0 +1,109 @@
+services:
+  opensearchserver:
+    image: opensearchproject/opensearch:3.3.1@sha256:773b05b026c01e7d520f6b857f486b7312d9290e14f2aebe4630eb80787e9eb3
+    container_name: opensearch-server
+    ports:
+      - "9200:9200"
+    environment:
+      - discovery.type=single-node
+      - DISABLE_SECURITY_PLUGIN=true
+      - DISABLE_INSTALL_DEMO_CONFIG=true
+  # we can reuse the elasticseearch server test
+  # no need to duplicate code
+  testserver:
+    build:
+      context: ../../test/integration/components/elasticsearch/
+      dockerfile: Dockerfile
+    image: hatest-testserver-python-opensearch
+    ports:
+      - "${TEST_SERVICE_PORTS}"
+    depends_on:
+      otelcol:
+        condition: service_started
+      opensearchserver:
+        condition: service_started
+
+  obi:
+    build:
+      context: ../../..
+      dockerfile: ./internal/test/integration/components/ebpf-instrument/Dockerfile
+    volumes:
+      - ./configs/:/configs
+      - ./system/sys/kernel/security:/sys/kernel/security
+      - ../../../testoutput:/coverage
+      - ../../../testoutput/run-python-opensearch:/var/run/beyla
+    image: hatest-obi
+    container_name: hatest-obi
+    privileged: true # in some environments (not GH Pull Requests) you can set it to false and then cap_add: [ SYS_ADMIN ]
+    network_mode: "service:testserver"
+    pid: "service:testserver"
+    environment:
+      OTEL_EBPF_CONFIG_PATH: "/configs/obi-config.yml"
+      GOCOVERDIR: "/coverage"
+      OTEL_EBPF_TRACE_PRINTER: "json_indent"
+      OTEL_EBPF_OPEN_PORT: "${OTEL_EBPF_OPEN_PORT}"
+      OTEL_EBPF_DISCOVERY_POLL_INTERVAL: 500ms
+      OTEL_EBPF_EXECUTABLE_PATH: "${OTEL_EBPF_EXECUTABLE_PATH}"
+      OTEL_EBPF_SERVICE_NAMESPACE: "integration-test"
+      OTEL_EBPF_METRICS_INTERVAL: "10ms"
+      OTEL_EBPF_BPF_BATCH_TIMEOUT: "10ms"
+      OTEL_EBPF_OTLP_TRACES_BATCH_TIMEOUT: "1ms"
+      OTEL_EBPF_LOG_LEVEL: "DEBUG"
+      OTEL_EBPF_LOG_CONFIG: "yaml"
+      OTEL_EBPF_BPF_DEBUG: "TRUE"
+      OTEL_EBPF_HOSTNAME: "beyla"
+      OTEL_EBPF_BPF_HTTP_REQUEST_TIMEOUT: "5s"
+      OTEL_EBPF_PROCESSES_INTERVAL: "100ms"
+      OTEL_EBPF_METRICS_FEATURES: "application"
+      OTEL_EBPF_BPF_BUFFER_SIZE_HTTP: 1024
+      OTEL_EBPF_HTTP_ELASTICSEARCH_ENABLED: true
+      OTEL_EBPF_PROTOCOL_DEBUG_PRINT: true
+    depends_on:
+      testserver:
+        condition: service_started
+
+  # OpenTelemetry Collector
+  otelcol:
+    image: otel/opentelemetry-collector-contrib:0.104.0@sha256:e07e325e303e86f4a87a617491e921b579a92da6d404007394757ac910bf9587
+    container_name: otel-col
+    deploy:
+      resources:
+        limits:
+          memory: 125M
+    restart: unless-stopped
+    command: ["--config=/etc/otelcol-config/otelcol-config.yml"]
+    volumes:
+      - ./configs/:/etc/otelcol-config
+    ports:
+      - "4317" # OTLP over gRPC receiver
+      - "4318:4318" # OTLP over HTTP receiver
+      - "9464" # Prometheus exporter
+      - "8888" # metrics endpoint
+    depends_on:
+      prometheus:
+        condition: service_started
+
+  # Prometheus
+  prometheus:
+    image: quay.io/prometheus/prometheus:v2.55.1@sha256:2659f4c2ebb718e7695cb9b25ffa7d6be64db013daba13e05c875451cf51b0d3
+    container_name: prometheus
+    command:
+      - --config.file=/etc/prometheus/prometheus-config.yml
+      - --web.enable-lifecycle
+      - --web.route-prefix=/
+    volumes:
+      - ./configs/:/etc/prometheus
+    ports:
+      - "9090:9090"
+
+  jaeger:
+    image: jaegertracing/all-in-one:1.60@sha256:4fd2d70fa347d6a47e79fcb06b1c177e6079f92cba88b083153d56263082135e
+    ports:
+      - "16686:16686" # Query frontend
+      - "4317" # OTEL GRPC traces collector
+      - "4318" # OTEL HTTP traces collector
+    environment:
+      - COLLECTOR_OTLP_ENABLED=true
+      - LOG_LEVEL=debug
+# curl http://localhost:16686/api/services
+# curl http://localhost:16686/api/traces?service=testserver

internal/test/integration/suites_test.go

@@ -522,6 +522,16 @@ func TestSuite_PythonElasticsearch(t *testing.T) {
 	require.NoError(t, compose.Close())
 }
 
+func TestSuite_PythonOpensearch(t *testing.T) {
+	compose, err := docker.ComposeSuite("docker-compose-opensearch.yml", path.Join(pathOutput, "test-suite-opensearch.log"))
+	require.NoError(t, err)
+
+	compose.Env = append(compose.Env, `OTEL_EBPF_OPEN_PORT=8080`, `OTEL_EBPF_EXECUTABLE_PATH=`, `TEST_SERVICE_PORTS=8381:8080`)
+	require.NoError(t, compose.Up())
+	t.Run("Python opensearch", testPythonOpensearch)
+	require.NoError(t, compose.Close())
+}
+
 func TestSuite_PythonAWSS3(t *testing.T) {
 	compose, err := docker.ComposeSuite("docker-compose-python-aws.yml", path.Join(pathOutput, "test-suite-python-aws-s3.log"))
 	require.NoError(t, err)

internal/test/integration/test_python_elasticsearchclient.go

@@ -22,23 +22,24 @@ import (
 )
 
 func testPythonElasticsearch(t *testing.T) {
-	url := "http://localhost:8381"
+	testServerURL := "http://localhost:8381"
+	elasticsearchURL := "http://elasticsearchserver:9200"
 	comm := "python3.12"
 	index := "test_index"
-
-	waitForTestComponentsRoute(t, url, "/health")
-	testElasticsearchSearch(t, comm, url, index)
+	queryParam := "?host_url=" + elasticsearchURL
+	waitForTestComponentsNoMetrics(t, testServerURL+"/health"+queryParam)
+	testElasticsearchSearch(t, comm, testServerURL, queryParam, index)
 	// populate the server is optional, the elasticsearch request will fail
 	// but we will have the span
-	testElasticsearchMsearch(t, comm, url)
-	testElasticsearchBulk(t, comm, url)
-	testElasticsearchDoc(t, comm, url, index)
+	testElasticsearchMsearch(t, comm, testServerURL, queryParam)
+	testElasticsearchBulk(t, comm, testServerURL, queryParam)
+	testElasticsearchDoc(t, comm, testServerURL, queryParam, index)
 }
 
-func testElasticsearchSearch(t *testing.T, comm, url, index string) {
+func testElasticsearchSearch(t *testing.T, comm, testServerURL, queryParam, index string) {
 	queryText := "{\"query\": {\"match\": {\"name\": \"OBI\"}}}"
 	urlPath := "/search"
-	ti.DoHTTPGet(t, url+urlPath, 200)
+	ti.DoHTTPGet(t, testServerURL+urlPath+queryParam, 200)
 	assertElasticsearchOperation(t, comm, "search", queryText, index)
 }
 
@@ -93,23 +94,23 @@ func assertElasticsearchOperation(t *testing.T, comm, op, queryText, index strin
 	}, test.Interval(100*time.Millisecond))
 }
 
-func testElasticsearchMsearch(t *testing.T, comm, url string) {
+func testElasticsearchMsearch(t *testing.T, comm, testServerURL, queryParam string) {
 	queryText := "[{}, {\"query\": {\"match\": {\"message\": \"this is a test\"}}}, {\"index\": \"my-index-000002\"}, {\"query\": {\"match_all\": {}}}]"
 	urlPath := "/msearch"
-	ti.DoHTTPGet(t, url+urlPath, 200)
+	ti.DoHTTPGet(t, testServerURL+urlPath+queryParam, 200)
 	assertElasticsearchOperation(t, comm, "msearch", queryText, "")
 }
 
-func testElasticsearchBulk(t *testing.T, comm, url string) {
+func testElasticsearchBulk(t *testing.T, comm, testServerURL, queryParam string) {
 	queryText := "[{\"index\": {\"_index\": \"test\", \"_id\": \"1\"}}, {\"field1\": \"value1\"}, {\"delete\": {\"_index\": \"test\", \"_id\": \"2\"}}, {\"create\": {\"_index\": \"test\", \"_id\": \"3\"}}, {\"field1\": \"value3\"}, {\"update\": {\"_id\": \"1\", \"_index\": \"test\"}}, {\"doc\": {\"field2\": \"value2\"}}]"
 	urlPath := "/bulk"
-	ti.DoHTTPGet(t, url+urlPath, 200)
+	ti.DoHTTPGet(t, testServerURL+urlPath+queryParam, 200)
 	assertElasticsearchOperation(t, comm, "bulk", queryText, "")
 }
 
-func testElasticsearchDoc(t *testing.T, comm, url, index string) {
+func testElasticsearchDoc(t *testing.T, comm, testServerURL, queryParam, index string) {
 	queryText := ""
 	urlPath := "/doc"
-	ti.DoHTTPGet(t, url+urlPath, 200)
+	ti.DoHTTPGet(t, testServerURL+urlPath+queryParam, 200)
 	assertElasticsearchOperation(t, comm, "doc", queryText, index)
 }