Performance tests - Span ingestion at the Agent

[jpkrohling]

Similar to #989, we need a set of performance tests for the Agent, to assess its performance at ingesting spans.

To exercise that, we might just need a Golang application that generates reasonably sized spans and dispatches them to the Agent running at localhost. At the end of the test, the metrics endpoint at the agent can be used to query the amount of spans ingested, comparing with the number of spans dispatched by the client.

The test is to be executed with different values for server-queue-size, to better understand its performance impact.

[yurishkuro]

I had this tracegen util internally that we used to stress test the agent (it needs a bit of cleanup for metrics to use jaeger-lib):

package main

import (
        "flag"
        "fmt"
        "sync"
        "sync/atomic"
        "time"

        "github.com/opentracing/opentracing-go"
        "github.com/opentracing/opentracing-go/ext"
        "github.com/uber/jaeger-client-go"
        "github.com/uber/jaeger-client-go/config"
        "go.uber.org/zap"

        "{some internal package}/metrics"
        "{some internal package}/m3metrics"
)

var logger, _ = zap.NewDevelopment()

// Sample configuration for the application
type appConfig struct {
        Metrics metrics.Configuration
        Jaeger  config.Configuration
}

type worker struct {
        id       int             // worker id
        traces   int             // how many traces the worker has to generate (only when duration==0)
        marshall bool            // whether the worker needs to marshall trace context via HTTP headers
        debug    bool            // whether to set DEBUG flag on the spans
        duration time.Duration   // how long to run the test for (overrides `traces`)
        pause    time.Duration   // how long to pause before finishing the trace
        running  *uint32         // pointer to shared flag that indicates it's time to stop the test
        wg       *sync.WaitGroup // notify when done
}

var fakeIP uint32 = 1<<24 | 2<<16 | 3<<8 | 4

func (w worker) simulateTraces() {
        tracer := opentracing.GlobalTracer()
        var i int
        for atomic.LoadUint32(w.running) == 1 {
                sp := tracer.StartSpan("lets-go")
                ext.SpanKindRPCClient.Set(sp)
                ext.PeerHostIPv4.Set(sp, fakeIP)
                ext.PeerService.Set(sp, "tracegen-server")
                if w.debug {
                        ext.SamplingPriority.Set(sp, 100)
                }

                childCtx := sp.Context()
                if w.marshall {
                        m := make(map[string]string)
                        c := opentracing.TextMapCarrier(m)
                        if err := tracer.Inject(sp.Context(), opentracing.TextMap, c); err == nil {
                                c := opentracing.TextMapCarrier(m)
                                childCtx, err = tracer.Extract(opentracing.TextMap, c)
                                if err != nil {
                                        logger.Error("cannot extract from TextMap", zap.Error(err))
                                }
                        } else {
                                logger.Error("cannot inject span", zap.Error(err))
                        }
                }
                child := opentracing.StartSpan(
                        "okey-dokey",
                        ext.RPCServerOption(childCtx),
                )
                ext.PeerHostIPv4.Set(child, fakeIP)
                ext.PeerService.Set(child, "tracegen-client")

                time.Sleep(w.pause)

                if w.pause == 0 {
                        child.Finish()
                        sp.Finish()
                } else {
                        opt := opentracing.FinishOptions{FinishTime: time.Now().Add(123 * time.Microsecond)}
                        child.FinishWithOptions(opt)
                        sp.FinishWithOptions(opt)
                }

                i++
                if w.traces != 0 {
                        if i >= w.traces {
                                break
                        }
                }
        }
        logger.Info(fmt.Sprintf("Worker %d generated %d traces", w.id, i))
        w.wg.Done()
}

func main() {
        var (
                workers   = flag.Int("workers", 1, "Number of workers (goroutines) to run")
                traces    = flag.Int("traces", 1, "Number of traces to generate in each worker (ignored if duration is provided")
                marshall  = flag.Bool("marshall", false, "Whether to marshall trace context via HTTP headers")
                debug     = flag.Bool("debug", false, "Whether to set DEBUG flag on the spans to prevent downsampling")
                pause     = flag.Duration("pause", time.Microsecond, "How long to pause before finishing trace")
                duration  = flag.Duration("duration", 0, "For how long to run the test")
                stats     = flag.Bool("stats", false, "Use local backend for stats and print it on exit")
                agentPort = flag.Int("agent-port", 0, "Override for default jaeger-agent UDP port")
        )

        flag.Parse()

        cfg := appConfig{
                Metrics: metrics.Configuration{
                        M3: &metrics.M3Configuration{
                                HostPort:    "127.0.0.1:9052",
                                Service:     "tracegen",
                                IncludeHost: true,
                        },
                },
        }

        if *duration > 0 {
                *traces = 0
        } else if *traces <= 0 {
                logger.Fatal("Either `traces` or `duration` must be greater than 0")
        }

        var scope metrics.Scope
        if *stats {
                logger.Info("Using local stats backend")
                statsBackend := metrics.NewLocalBackend()
                scope = metrics.NewRootScope(statsBackend)
                defer func() {
                        counters, gauges := statsBackend.Snapshot()
                        logger.Info(fmt.Sprintf("Counters accumulated: %+v", counters))
                        logger.Info(fmt.Sprintf("Gauges accumulated: %+v", gauges))
                }()
        } else {
                if m, err := cfg.Metrics.New(); err != nil {
                        logger.Fatal("Metrics.New failed", zap.Error(err))
                } else {
                        logger.Info(fmt.Sprintf("Using M3 stats scope, %+v", m))
                        scope = m
                        if bp, ok := m.(metrics.BackendProvider); ok {
                                logger.Info(fmt.Sprintf("Using M3 stats backend, %+v", bp.Backend()))
                        }
                }
        }

        agentHostPort := ""
        if *agentPort != 0 {
                agentHostPort = fmt.Sprintf(":%d", *agentPort)
        }
        sender, err := jaeger.NewUDPTransport(agentHostPort, 0)
        if err != nil {
                logger.Fatal("Cannot initialize UDPSender", zap.Error(err))
        }
        // TODO wrap zap logger
        // jLogger := tl.WrapLogger(logger.DefaultLogger())
        jMetrics := jaeger.NewMetrics(m3metrics.NewFactory(scope), nil)
        tracer, tCloser := jaeger.NewTracer("tracegen",
                jaeger.NewConstSampler(true),
                jaeger.NewRemoteReporter(
                        sender,
                        jaeger.ReporterOptions.Metrics(jMetrics),
                ),
                jaeger.TracerOptions.Metrics(jMetrics),
        )
        opentracing.InitGlobalTracer(tracer)
        logger.Info("Initialized global tracer")

        wg := &sync.WaitGroup{}
        var running uint32 = 1
        for i := 0; i < *workers; i++ {
                wg.Add(1)
                w := worker{
                        id:       i,
                        traces:   *traces,
                        marshall: *marshall,
                        debug:    *debug,
                        pause:    *pause,
                        duration: *duration,
                        running:  &running,
                        wg:       wg,
                }

                go w.simulateTraces()
        }
        if *duration > 0 {
                time.Sleep(*duration)
                atomic.StoreUint32(&running, 0)
        }
        wg.Wait()
        tCloser.Close()

        logger.Info("Waiting 1.5sec for metrics to flush")
        time.Sleep(3 * time.Second / 2)
}

[yurishkuro]

old issue