研究调用链跟踪技术之jaeger

$ curl 'https://api.github.com/repos/openzipkin/zipkin' 2>/dev/null | grep -E 'created_at|updated_at|stargazers_count|watchers_count|forks_count'
  "created_at": "2012-06-06T18:26:16Z",
  "updated_at": "2018-07-22T13:36:01Z",
  "stargazers_count": 9039,
  "watchers_count": 9039,
  "forks_count": 1500,

$ curl 'https://api.github.com/repos/jaegertracing/jaeger' 2>/dev/null | grep -E 'created_at|updated_at|stargazers_count|watchers_count|forks_count'
  "created_at": "2016-04-15T18:49:02Z",
  "updated_at": "2018-07-22T10:46:39Z",
  "stargazers_count": 5184,
  "watchers_count": 5184,
  "forks_count": 414,

COMPOSE_PROJECT_NAME=jaeger_demo
MY_HOST_IP=${your_host_ip}

version: '3'

services:
  elasticsearch:
    image: elasticsearch
    command: -Enode.name=jaegerESNode
    restart: always
    volumes:
      - "esdata:/usr/share/elasticsearch/data"
    ports:
      - "9200:9200"

  jaeger-collector:
    image: jaegertracing/jaeger-collector
    restart: always
    environment:
      SPAN_STORAGE_TYPE: elasticsearch
      ES_SERVER_URLS: http://${MY_HOST_IP:-127.0.0.1}:9200
    ports:
      - "14267:14267"
      - "14268:14268"
      - "9411:9411"

  jaeger-query:
    image: jaegertracing/jaeger-query
    restart: always
    environment:
      SPAN_STORAGE_TYPE: elasticsearch
      ES_SERVER_URLS: http://${MY_HOST_IP:-127.0.0.1}:9200
    ports:
      - "16686:16686"

  jaeger-agent:
    image: jaegertracing/jaeger-agent
    restart: always
    command: --collector.host-port=${MY_HOST_IP:-127.0.0.1}:14267
    ports:
      - "5775:5775/udp"
      - "6831:6831/udp"
      - "6832:6832/udp"
      - "5778:5778/tcp"
      
  jaeger-spark-dependencies:
    image: jaegertracing/spark-dependencies
    restart: always
    environment:
      STORAGE: elasticsearch
      ES_NODES: http://${MY_HOST_IP:-127.0.0.1}:9200
      ES_NODES_WAN_ONLY: 'true'
      JAVA_OPTS: -Dspark.testing.memory=481859200

  example-hotrod:
    image: jaegertracing/example-hotrod
    restart: always
    command: all --jaeger-agent.host-port=${MY_HOST_IP:-127.0.0.1}:6831
    ports:
      - "8080-8083:8080-8083"

volumes:
  esdata:
    driver: local
    driver_opts:
      type: none
      o: bind
      device: "${PWD}/esdata"

// Init creates a new instance of Jaeger tracer.
func Init(serviceName string, metricsFactory metrics.Factory, logger log.Factory, backendHostPort string) opentracing.Tracer {
	cfg := config.Configuration{
		Sampler: &config.SamplerConfig{
			Type:  "const",
			Param: 1,
		},
	}
	// TODO(ys) a quick hack to ensure random generators get different seeds, which are based on current time.
	time.Sleep(100 * time.Millisecond)
	jaegerLogger := jaegerLoggerAdapter{logger.Bg()}
	var sender jaeger.Transport
	if strings.HasPrefix(backendHostPort, "http://") {
		sender = transport.NewHTTPTransport(
			backendHostPort,
			transport.HTTPBatchSize(1),
		)
	} else {
		if s, err := jaeger.NewUDPTransport(backendHostPort, 0); err != nil {
			logger.Bg().Fatal("cannot initialize UDP sender", zap.Error(err))
		} else {
			sender = s
		}
	}
	tracer, _, err := cfg.New(
		serviceName,
		config.Reporter(jaeger.NewRemoteReporter(
			sender,
			jaeger.ReporterOptions.BufferFlushInterval(1*time.Second),
			jaeger.ReporterOptions.Logger(jaegerLogger),
		)),
		config.Logger(jaegerLogger),
		config.Metrics(metricsFactory),
		config.Observer(rpcmetrics.NewObserver(metricsFactory, rpcmetrics.DefaultNameNormalizer)),
	)
	if err != nil {
		logger.Bg().Fatal("cannot initialize Jaeger Tracer", zap.Error(err))
	}
	return tracer
}

// Handle implements http.ServeMux#Handle
func (tm *TracedServeMux) Handle(pattern string, handler http.Handler) {
	middleware := nethttp.Middleware(
		tm.tracer,
		handler,
		nethttp.OperationNameFunc(func(r *http.Request) string {
			return "HTTP " + r.Method + " " + pattern
		}))
	tm.mux.Handle(pattern, middleware)
}

// HTTPClient wraps an http.Client with tracing instrumentation.
type HTTPClient struct {
	Tracer opentracing.Tracer
	Client *http.Client
}

// GetJSON executes HTTP GET against specified url and tried to parse
// the response into out object.
func (c *HTTPClient) GetJSON(ctx context.Context, endpoint string, url string, out interface{}) error {
	req, err := http.NewRequest("GET", url, nil)
	if err != nil {
		return err
	}
	req = req.WithContext(ctx)
	req, ht := nethttp.TraceRequest(c.Tracer, req, nethttp.OperationName("HTTP GET: "+endpoint))
	defer ht.Finish()

	res, err := c.Client.Do(req)
	if err != nil {
		return err
	}

	defer res.Body.Close()

	if res.StatusCode >= 400 {
		body, err := ioutil.ReadAll(res.Body)
		if err != nil {
			return err
		}
		return errors.New(string(body))
	}
	decoder := json.NewDecoder(res.Body)
	return decoder.Decode(out)
}

// simulate opentracing instrumentation of an SQL query
	if span := opentracing.SpanFromContext(ctx); span != nil {
		span := d.tracer.StartSpan("SQL SELECT", opentracing.ChildOf(span.Context()))
		tags.SpanKindRPCClient.Set(span)
		tags.PeerService.Set(span, "mysql")
		span.SetTag("sql.query", "SELECT * FROM customer WHERE customer_id="+customerID)
		defer span.Finish()
		ctx = opentracing.ContextWithSpan(ctx, span)
	}

type spanLogger struct {
	logger *zap.Logger
	span   opentracing.Span
}

func (sl spanLogger) Info(msg string, fields ...zapcore.Field) {
	sl.logToSpan("info", msg, fields...)
	sl.logger.Info(msg, fields...)
}

func (sl spanLogger) Error(msg string, fields ...zapcore.Field) {
	sl.logToSpan("error", msg, fields...)
	sl.logger.Error(msg, fields...)
}

func (sl spanLogger) Fatal(msg string, fields ...zapcore.Field) {
	sl.logToSpan("fatal", msg, fields...)
	tag.Error.Set(sl.span, true)
	sl.logger.Fatal(msg, fields...)
}

// With creates a child logger, and optionally adds some context fields to that logger.
func (sl spanLogger) With(fields ...zapcore.Field) Logger {
	return spanLogger{logger: sl.logger.With(fields...), span: sl.span}
}

func (sl spanLogger) logToSpan(level string, msg string, fields ...zapcore.Field) {
	// TODO rather than always converting the fields, we could wrap them into a lazy logger
	fa := fieldAdapter(make([]log.Field, 0, 2+len(fields)))
	fa = append(fa, log.String("event", msg))
	fa = append(fa, log.String("level", level))
	for _, field := range fields {
		field.AddTo(&fa)
	}
	sl.span.LogFields(fa...)
}

d.logger.For(ctx).Info("Loading customer", zap.String("customer_id", customerID))