Add Tempo-gated e2e test for OTel tracing
Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
424
test/e2e/tracing_test.go
Normal file
424
test/e2e/tracing_test.go
Normal file
@@ -0,0 +1,424 @@
|
||||
//go:build e2e
|
||||
// +build e2e
|
||||
|
||||
/*
|
||||
Copyright 2026.
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
|
||||
package e2e
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
|
||||
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/test/utils"
|
||||
)
|
||||
|
||||
// The tracing e2e proves the full pipeline against a real Tempo: operator in
|
||||
// kind → OTLP export → Tempo ingest → traces queryable with the documented
|
||||
// span topology. It is gated on TEMPO_URL (Tempo query API, e.g.
|
||||
// http://192.168.0.30:3200) and OTLP_ENDPOINT (OTLP HTTP ingest, e.g.
|
||||
// http://192.168.0.30:4318) and skips when either is unset, so the rest of
|
||||
// the suite runs anywhere.
|
||||
//
|
||||
// Every span of a run carries the resource attribute test.run.id=<runID>
|
||||
// (injected via OTEL_RESOURCE_ATTRIBUTES — no code changes), so one TraceQL
|
||||
// query finds exactly this run's traces, in the test and in Grafana alike.
|
||||
var _ = Describe("OTel tracing", Ordered, func() {
|
||||
// Proxy CRs live in default, not the operator namespace: the squid pods
|
||||
// the kubernetes provider creates carry no securityContext and would be
|
||||
// rejected by the operator namespace's restricted PSS label.
|
||||
const proxyNS = "default"
|
||||
const deploymentName = "egress-proxies-operator-controller-manager"
|
||||
const squidImage = "ubuntu/squid:6.6-24.04_edge"
|
||||
|
||||
proxyNames := []string{"proxy-tracing-e2e-1", "proxy-tracing-e2e-2"}
|
||||
|
||||
var (
|
||||
tempoURL string
|
||||
otlpEndpoint string
|
||||
runID string
|
||||
suiteStart time.Time
|
||||
)
|
||||
|
||||
BeforeAll(func() {
|
||||
tempoURL = os.Getenv("TEMPO_URL")
|
||||
otlpEndpoint = os.Getenv("OTLP_ENDPOINT")
|
||||
if tempoURL == "" || otlpEndpoint == "" {
|
||||
Skip("TEMPO_URL / OTLP_ENDPOINT not set — skipping the Tempo-backed tracing e2e")
|
||||
}
|
||||
suiteStart = time.Now()
|
||||
runID = "e2e-" + strconv.FormatInt(suiteStart.UnixNano(), 10)
|
||||
_, _ = fmt.Fprintf(GinkgoWriter,
|
||||
"tracing e2e run id: %s — find this run in Grafana with TraceQL {resource.test.run.id=%q}\n",
|
||||
runID, runID)
|
||||
|
||||
By("creating manager namespace")
|
||||
cmd := exec.Command("kubectl", "create", "ns", namespace)
|
||||
_, err := utils.Run(cmd)
|
||||
Expect(err).NotTo(HaveOccurred(), "Failed to create namespace")
|
||||
|
||||
By("labeling the namespace to enforce the restricted security policy")
|
||||
cmd = exec.Command("kubectl", "label", "--overwrite", "ns", namespace,
|
||||
"pod-security.kubernetes.io/enforce=restricted")
|
||||
_, err = utils.Run(cmd)
|
||||
Expect(err).NotTo(HaveOccurred(), "Failed to label namespace with restricted policy")
|
||||
|
||||
By("installing CRDs")
|
||||
cmd = exec.Command("make", "install")
|
||||
_, err = utils.Run(cmd)
|
||||
Expect(err).NotTo(HaveOccurred(), "Failed to install CRDs")
|
||||
|
||||
By("deploying the controller-manager")
|
||||
cmd = exec.Command("make", "deploy", fmt.Sprintf("IMG=%s", managerImage))
|
||||
_, err = utils.Run(cmd)
|
||||
Expect(err).NotTo(HaveOccurred(), "Failed to deploy the controller-manager")
|
||||
|
||||
By("pre-pulling the squid image into kind (best effort, kills the biggest flake source)")
|
||||
if _, err := utils.Run(exec.Command("docker", "pull", squidImage)); err == nil {
|
||||
if err := utils.LoadImageToKindClusterWithName(squidImage); err != nil {
|
||||
_, _ = fmt.Fprintf(GinkgoWriter, "kind load of %s failed (continuing): %v\n", squidImage, err)
|
||||
}
|
||||
} else {
|
||||
_, _ = fmt.Fprintf(GinkgoWriter, "docker pull %s failed (continuing): %v\n", squidImage, err)
|
||||
}
|
||||
|
||||
By("preflighting the OTLP endpoint from inside the cluster")
|
||||
// Host-reachability of the OTLP endpoint does not prove
|
||||
// pod-reachability from inside kind, and the operator logs export
|
||||
// failures only at V(1) — without this, a broken path is a slow,
|
||||
// opaque search timeout instead of a clear failure.
|
||||
preflightOTLP(otlpEndpoint)
|
||||
|
||||
By("pointing the operator at the OTLP endpoint and tagging the test run")
|
||||
// OTEL_RESOURCE_ATTRIBUTES is replaced in place, which keeps it
|
||||
// listed after the downward-API POD_NAME/POD_NAMESPACE vars —
|
||||
// $(VAR) expansion only sees earlier-listed vars. exec.Command
|
||||
// passes $(...) through without shell mangling.
|
||||
cmd = exec.Command("kubectl", "set", "env",
|
||||
"deployment/"+deploymentName, "-n", namespace,
|
||||
"OTEL_EXPORTER_OTLP_ENDPOINT="+otlpEndpoint,
|
||||
fmt.Sprintf(
|
||||
"OTEL_RESOURCE_ATTRIBUTES=k8s.pod.name=$(POD_NAME),k8s.namespace.name=$(POD_NAMESPACE),test.run.id=%s",
|
||||
runID))
|
||||
_, err = utils.Run(cmd)
|
||||
Expect(err).NotTo(HaveOccurred(), "Failed to set OTel env on the deployment")
|
||||
|
||||
cmd = exec.Command("kubectl", "rollout", "status",
|
||||
"deployment/"+deploymentName, "-n", namespace, "--timeout=3m")
|
||||
_, err = utils.Run(cmd)
|
||||
Expect(err).NotTo(HaveOccurred(), "Rollout after set env did not finish")
|
||||
|
||||
By("verifying the operator reports tracing enabled")
|
||||
Eventually(func(g Gomega) {
|
||||
pod := newestControllerPod(g)
|
||||
out, err := utils.Run(exec.Command("kubectl", "logs", pod, "-n", namespace))
|
||||
g.Expect(err).NotTo(HaveOccurred())
|
||||
g.Expect(out).To(ContainSubstring("tracing enabled"),
|
||||
"operator did not log 'tracing enabled' after rollout")
|
||||
}, 2*time.Minute).Should(Succeed())
|
||||
})
|
||||
|
||||
AfterAll(func() {
|
||||
if tempoURL == "" || otlpEndpoint == "" {
|
||||
return // spec was skipped; nothing was deployed
|
||||
}
|
||||
By("cleaning up test proxies")
|
||||
args := append([]string{"delete", "proxy", "-n", proxyNS, "--ignore-not-found"}, proxyNames...)
|
||||
_, _ = utils.Run(exec.Command("kubectl", args...))
|
||||
|
||||
By("cleaning up the OTLP preflight pod")
|
||||
_, _ = utils.Run(exec.Command("kubectl", "delete", "pod", otlpProbePodName, "-n", namespace,
|
||||
"--ignore-not-found"))
|
||||
|
||||
By("undeploying the controller-manager")
|
||||
_, _ = utils.Run(exec.Command("make", "undeploy"))
|
||||
|
||||
By("uninstalling CRDs")
|
||||
_, _ = utils.Run(exec.Command("make", "uninstall"))
|
||||
|
||||
By("removing manager namespace")
|
||||
_, _ = utils.Run(exec.Command("kubectl", "delete", "ns", namespace))
|
||||
})
|
||||
|
||||
It("creates proxies and reports reconcile traces to Tempo", func() {
|
||||
By("applying two kubernetes-provider proxies")
|
||||
// Absolute paths on purpose: utils.Run chdirs the whole process.
|
||||
dir := GinkgoT().TempDir()
|
||||
for _, name := range proxyNames {
|
||||
manifest := fmt.Sprintf(`apiVersion: crawl.example.com/v1alpha1
|
||||
kind: Proxy
|
||||
metadata:
|
||||
name: %s
|
||||
namespace: %s
|
||||
spec:
|
||||
mode: Managed
|
||||
provider: kubernetes
|
||||
attributes:
|
||||
purpose: tracing-e2e
|
||||
`, name, proxyNS)
|
||||
path := filepath.Join(dir, name+".yaml")
|
||||
Expect(os.WriteFile(path, []byte(manifest), 0o644)).To(Succeed())
|
||||
_, err := utils.Run(exec.Command("kubectl", "apply", "-f", path))
|
||||
Expect(err).NotTo(HaveOccurred(), "Failed to apply %s", name)
|
||||
}
|
||||
|
||||
By("waiting for the proxies to become Ready")
|
||||
Eventually(func(g Gomega) {
|
||||
for _, name := range proxyNames {
|
||||
out, err := utils.Run(exec.Command("kubectl", "get", "proxy", name,
|
||||
"-n", proxyNS, "-o", "jsonpath={.status.phase}"))
|
||||
g.Expect(err).NotTo(HaveOccurred())
|
||||
g.Expect(out).To(Equal("Ready"), "proxy %s not Ready", name)
|
||||
}
|
||||
}, 5*time.Minute).Should(Succeed())
|
||||
|
||||
By("finding a provider.create trace for this run in Tempo")
|
||||
// Anchored on provider.create, not the root span name: a
|
||||
// "Reconcile Proxy" hit could be the finalizer-add or a drift
|
||||
// reconcile, which contain no provider call.
|
||||
traceID := eventuallyFindTrace(tempoURL, runID, "provider.create", suiteStart)
|
||||
|
||||
By("asserting the reconcile trace structure")
|
||||
spanNames, resAttrs, err := tempoTrace(tempoURL, traceID)
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
Expect(spanNames).To(ContainElements(
|
||||
"Reconcile Proxy", "reconcile.managed", "provider.create", "status.patch"),
|
||||
"trace %s is missing expected spans; got: %v", traceID, spanNames)
|
||||
Expect(resAttrs["service.name"]).To(Equal("egress-proxies-operator"))
|
||||
Expect(resAttrs["test.run.id"]).To(Equal(runID))
|
||||
})
|
||||
|
||||
It("traces proxy deletion", func() {
|
||||
By("deleting the proxies")
|
||||
args := append([]string{"delete", "proxy", "-n", proxyNS, "--wait=false"}, proxyNames...)
|
||||
_, err := utils.Run(exec.Command("kubectl", args...))
|
||||
Expect(err).NotTo(HaveOccurred(), "Failed to delete proxies")
|
||||
|
||||
By("waiting for the proxies to be gone (finalizer ran provider.delete)")
|
||||
Eventually(func(g Gomega) {
|
||||
out, err := utils.Run(exec.Command("kubectl", "get", "proxy", "-n", proxyNS, "-o", "name"))
|
||||
g.Expect(err).NotTo(HaveOccurred())
|
||||
g.Expect(out).NotTo(ContainSubstring("proxy-tracing-e2e"))
|
||||
}, 3*time.Minute).Should(Succeed())
|
||||
|
||||
By("finding a provider.delete trace for this run in Tempo")
|
||||
traceID := eventuallyFindTrace(tempoURL, runID, "provider.delete", suiteStart)
|
||||
|
||||
By("asserting the deletion trace structure")
|
||||
spanNames, _, err := tempoTrace(tempoURL, traceID)
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
Expect(spanNames).To(ContainElements("Reconcile Proxy", "reconcile.delete", "provider.delete"),
|
||||
"trace %s is missing expected spans; got: %v", traceID, spanNames)
|
||||
})
|
||||
})
|
||||
|
||||
// preflightOTLP runs a one-shot curl pod inside the cluster POSTing to the
|
||||
// OTLP HTTP ingest, and fails with a clear message when it is unreachable.
|
||||
// The pod runs in the restricted-PSS operator namespace, hence the full
|
||||
// securityContext (same shape as the curl-metrics pod).
|
||||
func preflightOTLP(otlpEndpoint string) {
|
||||
script := fmt.Sprintf(
|
||||
"for i in $(seq 1 10); do "+
|
||||
"code=$(curl -sS -o /dev/null -w '%%{http_code}' -X POST "+
|
||||
"-H 'Content-Type: application/json' -d '{}' %s/v1/traces); "+
|
||||
"echo \"attempt $i: HTTP $code\"; "+
|
||||
"[ \"$code\" = \"200\" ] && echo OTLP_OK && exit 0; sleep 2; "+
|
||||
"done; echo OTLP_UNREACHABLE; exit 1",
|
||||
otlpEndpoint)
|
||||
cmd := exec.Command("kubectl", "run", otlpProbePodName, "--restart=Never",
|
||||
"--namespace", namespace,
|
||||
"--image=curlimages/curl:latest",
|
||||
"--overrides",
|
||||
fmt.Sprintf(`{
|
||||
"spec": {
|
||||
"containers": [{
|
||||
"name": "curl",
|
||||
"image": "curlimages/curl:latest",
|
||||
"command": ["/bin/sh", "-c"],
|
||||
"args": [%q],
|
||||
"securityContext": {
|
||||
"readOnlyRootFilesystem": true,
|
||||
"allowPrivilegeEscalation": false,
|
||||
"capabilities": {
|
||||
"drop": ["ALL"]
|
||||
},
|
||||
"runAsNonRoot": true,
|
||||
"runAsUser": 1000,
|
||||
"seccompProfile": {
|
||||
"type": "RuntimeDefault"
|
||||
}
|
||||
}
|
||||
}]
|
||||
}
|
||||
}`, script))
|
||||
_, err := utils.Run(cmd)
|
||||
Expect(err).NotTo(HaveOccurred(), "Failed to create the OTLP preflight pod")
|
||||
|
||||
Eventually(func(g Gomega) {
|
||||
out, err := utils.Run(exec.Command("kubectl", "get", "pod", otlpProbePodName,
|
||||
"-n", namespace, "-o", "jsonpath={.status.phase}"))
|
||||
g.Expect(err).NotTo(HaveOccurred())
|
||||
g.Expect(out).To(BeElementOf("Succeeded", "Failed"), "preflight pod still running")
|
||||
}, 2*time.Minute).Should(Succeed())
|
||||
|
||||
logs, _ := utils.Run(exec.Command("kubectl", "logs", otlpProbePodName, "-n", namespace))
|
||||
Expect(logs).To(ContainSubstring("OTLP_OK"),
|
||||
"OTLP endpoint %s is not reachable from inside the kind cluster; curl output:\n%s",
|
||||
otlpEndpoint, logs)
|
||||
}
|
||||
|
||||
// otlpProbePodName mirrors the Describe-local constant for the helpers below.
|
||||
const otlpProbePodName = "curl-otlp"
|
||||
|
||||
// newestControllerPod returns the most recently created controller pod —
|
||||
// right after a rollout, an unsorted lookup may pick the terminating one.
|
||||
func newestControllerPod(g Gomega) string {
|
||||
out, err := utils.Run(exec.Command("kubectl", "get", "pods",
|
||||
"-l", "control-plane=controller-manager", "-n", namespace,
|
||||
"--sort-by=.metadata.creationTimestamp", "-o", "name"))
|
||||
g.Expect(err).NotTo(HaveOccurred())
|
||||
lines := utils.GetNonEmptyLines(out)
|
||||
g.Expect(lines).NotTo(BeEmpty(), "no controller pods found")
|
||||
return strings.TrimPrefix(lines[len(lines)-1], "pod/")
|
||||
}
|
||||
|
||||
// eventuallyFindTrace polls Tempo until a trace containing a span with the
|
||||
// given name exists for this run, and returns its trace ID. The batch span
|
||||
// processor flushes every ~5s, so a couple of polls is normal.
|
||||
func eventuallyFindTrace(tempoURL, runID, spanName string, since time.Time) string {
|
||||
var traceID string
|
||||
query := fmt.Sprintf(`{resource.test.run.id=%q && name=%q}`, runID, spanName)
|
||||
Eventually(func(g Gomega) {
|
||||
ids, err := tempoSearch(tempoURL, query, since)
|
||||
g.Expect(err).NotTo(HaveOccurred())
|
||||
g.Expect(ids).NotTo(BeEmpty(), "no trace for %s yet (query: %s)", spanName, query)
|
||||
traceID = ids[0]
|
||||
}, 2*time.Minute).Should(Succeed())
|
||||
return traceID
|
||||
}
|
||||
|
||||
// tempoSearch runs a TraceQL query against Tempo's search API and returns
|
||||
// the matching trace IDs. start/end are Unix seconds.
|
||||
func tempoSearch(tempoURL, traceql string, since time.Time) ([]string, error) {
|
||||
u, err := url.Parse(tempoURL + "/api/search")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
q := u.Query()
|
||||
q.Set("q", traceql)
|
||||
q.Set("start", strconv.FormatInt(since.Add(-5*time.Minute).Unix(), 10))
|
||||
q.Set("end", strconv.FormatInt(time.Now().Add(time.Minute).Unix(), 10))
|
||||
q.Set("limit", "20")
|
||||
u.RawQuery = q.Encode()
|
||||
|
||||
body, err := tempoGet(u.String())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var result struct {
|
||||
Traces []struct {
|
||||
TraceID string `json:"traceID"`
|
||||
} `json:"traces"`
|
||||
}
|
||||
if err := json.Unmarshal(body, &result); err != nil {
|
||||
return nil, fmt.Errorf("decoding Tempo search response: %w", err)
|
||||
}
|
||||
ids := make([]string, 0, len(result.Traces))
|
||||
for _, t := range result.Traces {
|
||||
ids = append(ids, t.TraceID)
|
||||
}
|
||||
return ids, nil
|
||||
}
|
||||
|
||||
// tempoTrace fetches one trace and flattens it to a span-name list plus the
|
||||
// resource attributes. Tempo returns OTLP-JSON (batches → scopeSpans →
|
||||
// spans), not Jaeger's shape.
|
||||
func tempoTrace(tempoURL, traceID string) ([]string, map[string]string, error) {
|
||||
body, err := tempoGet(tempoURL + "/api/traces/" + traceID)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
var trace struct {
|
||||
Batches []struct {
|
||||
Resource struct {
|
||||
Attributes []struct {
|
||||
Key string `json:"key"`
|
||||
Value struct {
|
||||
StringValue string `json:"stringValue"`
|
||||
} `json:"value"`
|
||||
} `json:"attributes"`
|
||||
} `json:"resource"`
|
||||
ScopeSpans []struct {
|
||||
Spans []struct {
|
||||
Name string `json:"name"`
|
||||
} `json:"spans"`
|
||||
} `json:"scopeSpans"`
|
||||
} `json:"batches"`
|
||||
}
|
||||
if err := json.Unmarshal(body, &trace); err != nil {
|
||||
return nil, nil, fmt.Errorf("decoding Tempo trace %s: %w", traceID, err)
|
||||
}
|
||||
var spanNames []string
|
||||
resAttrs := map[string]string{}
|
||||
for _, b := range trace.Batches {
|
||||
for _, a := range b.Resource.Attributes {
|
||||
if a.Value.StringValue != "" {
|
||||
resAttrs[a.Key] = a.Value.StringValue
|
||||
}
|
||||
}
|
||||
for _, ss := range b.ScopeSpans {
|
||||
for _, s := range ss.Spans {
|
||||
spanNames = append(spanNames, s.Name)
|
||||
}
|
||||
}
|
||||
}
|
||||
return spanNames, resAttrs, nil
|
||||
}
|
||||
|
||||
func tempoGet(rawURL string) ([]byte, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
|
||||
defer cancel()
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, rawURL, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
resp, err := http.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("querying Tempo: %w", err)
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
body, err := io.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return nil, fmt.Errorf("tempo returned %d for %s: %s", resp.StatusCode, rawURL, string(body))
|
||||
}
|
||||
return body, nil
|
||||
}
|
||||
Reference in New Issue
Block a user