contrib: add validation tests using test-agent

.github/workflows/unit-integration-tests.yml

@@ -109,6 +109,18 @@ jobs:
  ports:
  - 8125:8125/udp
  - 8126:8126
+ testagent:
+ image: williamconti549/dd-apm-test-agent:service-naming
+ ports:
+ - 9126:9126
+ env:
+ LOG_LEVEL: DEBUG
+ TRACE_LANGUAGE: golang
+ DISABLED_CHECKS: trace_content_length,meta_tracer_version_header
+ PORT: 9126
+ DD_SUPPRESS_TRACE_PARSE_ERRORS: true
+ DD_POOL_TRACE_CHECK_FAILURES: true
+ DD_DISABLE_ERROR_RESPONSES: true
  cassandra:
  image: cassandra:3.7
  env:
@@ -241,6 +253,11 @@ jobs:
  shell: bash
  run: bash <(curl -s https://codecov.io/bash)
 
+ - name: Get Datadog APM Test Agent Logs
+ if: always()
+ shell: bash
+ run: docker logs ${{ job.services.testagent.id }}
+
  - name: Testing outlier google.golang.org/api
  run: |
  go get google.golang.org/[email protected] # version used to generate code

contrib/internal/validationtest/integrations/gomemcache/memcache/integration.go

@@ -0,0 +1,39 @@
+package memcache
+
+import (
+ "testing"
+
+ memcachetrace "gopkg.in/DataDog/dd-trace-go.v1/contrib/bradfitz/gomemcache/memcache"
+
+ "github.com/bradfitz/gomemcache/memcache"
+ "github.com/stretchr/testify/require"
+)
+
+type Integration struct {
+ client *memcachetrace.Client
+ numSpans int
+}
+
+func New() *Integration {
+ return &Integration{}
+}
+
+func (i *Integration) Name() string {
+ return "contrib/bradfitz/gomemcache/memcache"
+}
+
+func (i *Integration) Init(_ *testing.T) func() {
+ i.client = memcachetrace.WrapClient(memcache.New("127.0.0.1:11211"))
+ return func() {}
+}
+
+func (i *Integration) GenSpans(t *testing.T) {
+ t.Helper()
+ err := i.client.Set(&memcache.Item{Key: "myKey", Value: []byte("myValue")})
+ require.NoError(t, err)
+ i.numSpans++
+}
+
+func (i *Integration) NumSpans() int {
+ return i.numSpans
+}

contrib/internal/validationtest/integrations/miekg/dns/integration.go

@@ -0,0 +1,118 @@
+package memcache
+
+import (
+ "context"
+ "net"
+ "testing"
+ "time"
+
+ dnstrace "gopkg.in/DataDog/dd-trace-go.v1/contrib/miekg/dns"
+
+ "github.com/miekg/dns"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+type Integration struct {
+ msg *dns.Msg
+ mux *dns.ServeMux
+
+ addr string
+
+ numSpans int
+}
+
+func New() *Integration {
+ return &Integration{}
+}
+
+func (i *Integration) Name() string {
+ return "contrib/miekg/dns"
+}
+
+func (i *Integration) Init(t *testing.T) func() {
+ t.Helper()
+ i.addr = getFreeAddr(t).String()
+ server := &dns.Server{
+ Addr: i.addr,
+ Net: "udp",
+ Handler: dnstrace.WrapHandler(&handler{t: t, ig: i}),
+ }
+
+ // start the traced server
+ go func() {
+ require.NoError(t, server.ListenAndServe())
+ }()
+
+ // wait for the server to be ready
+ waitServerReady(t, server.Addr)
+
+ cleanup := func() {
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ assert.NoError(t, server.ShutdownContext(ctx))
+ }
+ return cleanup
+}
+
+func (i *Integration) GenSpans(t *testing.T) {
+ t.Helper()
+ msg := newMessage()
+
+ _, err := dnstrace.Exchange(msg, i.addr)
+ require.NoError(t, err)
+ i.numSpans++
+}
+
+func (i *Integration) NumSpans() int {
+ return i.numSpans
+}
+
+func newMessage() *dns.Msg {
+ m := new(dns.Msg)
+ m.SetQuestion("miek.nl.", dns.TypeMX)
+ return m
+}
+
+type handler struct {
+ t *testing.T
+ ig *Integration
+}
+
+func (h *handler) ServeDNS(w dns.ResponseWriter, r *dns.Msg) {
+ m := new(dns.Msg)
+ m.SetReply(r)
+ assert.NoError(h.t, w.WriteMsg(m))
+ h.ig.numSpans++
+}
+
+func getFreeAddr(t *testing.T) net.Addr {
+ li, err := net.Listen("tcp", "127.0.0.1:0")
+ require.NoError(t, err)
+ addr := li.Addr()
+ require.NoError(t, li.Close())
+ return addr
+}
+
+func waitServerReady(t *testing.T, addr string) {
+ ticker := time.NewTicker(100 * time.Millisecond)
+ defer ticker.Stop()
+ timeoutChan := time.After(5 * time.Second)
+
+ for {
+ m := new(dns.Msg)
+ m.SetQuestion("miek.nl.", dns.TypeMX)
+ _, err := dns.Exchange(m, addr)
+ if err == nil {
+ break
+ }
+
+ select {
+ case <-ticker.C:
+ continue
+
+ case <-timeoutChan:
+ t.Fatal("timeout waiting for DNS server to be ready")
+ }
+ }
+}

contrib/internal/validationtest/validation_test.go

@@ -0,0 +1,134 @@
+package validationtest
+
+import (
+ "encoding/json"
+ "fmt"
+ "io"
+ "net/http"
+ "testing"
+ "time"
+
+ memcachetest "gopkg.in/DataDog/dd-trace-go.v1/contrib/internal/validationtest/integrations/gomemcache/memcache"
+ dnstest "gopkg.in/DataDog/dd-trace-go.v1/contrib/internal/validationtest/integrations/miekg/dns"
+ "gopkg.in/DataDog/dd-trace-go.v1/ddtrace/tracer"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+// Integration is an interface that should be implemented by integrations (packages under the contrib/ folder) in
+// order to be tested.
+type Integration interface {
+ // Name returns name of the integration (usually the import path starting from /contrib).
+ Name() string
+
+ // Init initializes the integration (start a server in the background, initialize the client, etc.).
+ // It should return a cleanup function that will be executed after the test finishes.
+ // It should also call t.Helper() before making any assertions.
+ Init(t *testing.T) func()
+
+ // GenSpans performs any operation(s) from the integration that generate spans.
+ // It should call t.Helper() before making any assertions.
+ GenSpans(t *testing.T)
+
+ // NumSpans returns the number of spans that should have been generated during the test.
+ NumSpans() int
+}
+
+func TestIntegrations(t *testing.T) {
+ integrations := []Integration{
+ memcachetest.New(),
+ dnstest.New(),
+ }
+ for _, ig := range integrations {
+ name := ig.Name()
+ t.Run(name, func(t *testing.T) {
+ sessionToken := fmt.Sprintf("%s-%d", name, time.Now().Unix())
+ t.Setenv("DD_SERVICE", "Datadog-Test-Agent-Trace-Checks")
+ t.Setenv("CI_TEST_AGENT_SESSION_TOKEN", sessionToken)
+ t.Setenv("DD_TRACE_SPAN_ATTRIBUTE_SCHEMA", "v1")
+
+ tracer.Start(tracer.WithAgentAddr("localhost:9126"))
+ defer tracer.Stop()
+
+ cleanup := ig.Init(t)
+ defer cleanup()
+
+ ig.GenSpans(t)
+
+ tracer.Flush()
+
+ assertNumSpans(t, sessionToken, ig.NumSpans())
+ checkFailures(t, sessionToken)
+ })
+ }
+}
+
+func assertNumSpans(t *testing.T, sessionToken string, wantSpans int) {
+ t.Helper()
+ run := func() bool {
+ req, err := http.NewRequest("GET", "http://localhost:9126/test/session/traces", nil)
+ require.NoError(t, err)
+ req.Header.Set("X-Datadog-Test-Session-Token", sessionToken)
+
+ resp, err := http.DefaultClient.Do(req)
+ require.NoError(t, err)
+
+ defer resp.Body.Close()
+ require.Equal(t, http.StatusOK, resp.StatusCode)
+
+ body, err := io.ReadAll(resp.Body)
+ require.NoError(t, err)
+
+ var traces [][]map[string]interface{}
+ require.NoError(t, json.Unmarshal(body, &traces))
+
+ receivedSpans := 0
+ for _, traceSpans := range traces {
+ receivedSpans += len(traceSpans)
+ }
+ if receivedSpans > wantSpans {
+ t.Fatalf("received more spans than expected (wantSpans: %d, receivedSpans: %d)", wantSpans, receivedSpans)
+ }
+ return receivedSpans == wantSpans
+ }
+
+ ticker := time.NewTicker(1 * time.Second)
+ defer ticker.Stop()
+ timeoutChan := time.After(5 * time.Second)
+
+ for {
+ if done := run(); done {
+ return
+ }
+ select {
+ case <-ticker.C:
+ continue
+
+ case <-timeoutChan:
+ t.Fatal("timeout waiting for spans")
+ }
+ }
+}
+
+func checkFailures(t *testing.T, sessionToken string) {
+ t.Helper()
+ req, err := http.NewRequest("GET", "http://localhost:9126/test/trace_check/failures", nil)
+ require.NoError(t, err)
+ req.Header.Set("X-Datadog-Test-Session-Token", sessionToken)
+
+ resp, err := http.DefaultClient.Do(req)
+ require.NoError(t, err)
+
+ defer resp.Body.Close()
+
+ if resp.StatusCode == http.StatusOK {
+ return
+ }
+ assert.Equal(t, http.StatusOK, resp.StatusCode)
+
+ body, err := io.ReadAll(resp.Body)
+ require.NoError(t, err)
+
+ assert.Fail(t, "test agent detected failures: \n", string(body))
+}

ddtrace/tracer/option.go

@@ -168,6 +168,19 @@ func (c *config) HasFeature(f string) bool {
  return ok
 }
 
+func (c *config) DumpForTestAgent() string {
+ envVars := map[string]string{
+ "DD_SERVICE": c.serviceName,
+ "DD_TRACE_SPAN_ATTRIBUTE_SCHEMA": fmt.Sprintf("v%d", c.spanAttributeSchemaVersion),
+ "DD_TRACE_PEER_SERVICE_DEFAULTS_ENABLED": strconv.FormatBool(c.peerServiceDefaultsEnabled),
+ }
+ values := make([]string, 0, len(envVars))
+ for k, v := range envVars {
+ values = append(values, fmt.Sprintf("%s=%s", k, v))
+ }
+ return strings.Join(values, ",")
+}
+
 // StartOption represents a function that can be provided as a parameter to Start.
 type StartOption func(*config)
 
@@ -343,7 +356,13 @@ func newConfig(opts ...StartOption) *config {
  }
  c.dogstatsdAddr = addr
  }
-
+ if testSessionToken := os.Getenv("CI_TEST_AGENT_SESSION_TOKEN"); testSessionToken != "" {
+ tr, ok := c.transport.(*httpTransport)
+ if ok {
+ tr.headers["X-Datadog-Trace-Env-Variables"] = c.DumpForTestAgent()
+ tr.headers["X-Datadog-Test-Session-Token"] = testSessionToken
+ }
+ }
  return c
 }
 

docker-compose.yaml

@@ -90,6 +90,18 @@ services:
  DD_API_KEY: invalid_key_but_this_is_fine
  ports:
  - "8126:8126"
+ testagent:
+ image: williamconti549/dd-apm-test-agent:service-naming
+ environment:
+ LOG_LEVEL: DEBUG
+ TRACE_LANGUAGE: golang
+ DISABLED_CHECKS: trace_content_length,meta_tracer_version_header
+ PORT: 9126
+ DD_SUPPRESS_TRACE_PARSE_ERRORS: true
+ DD_POOL_TRACE_CHECK_FAILURES: true
+ DD_DISABLE_ERROR_RESPONSES: true
+ ports:
+ - "127.0.0.1:9126:9126"
  mongodb:
  image: circleci/mongo:latest-ram
  ports: