数据采集采集源配置通过 jaeger 采集链路追踪数据

通过 jaeger 采集链路追踪数据

简介

接收符合 Jaeger 协议格式的链路数据，并把数据经过统一转换成 DataFlux 的链路数据后上报到 DataFlux 中。

前置条件

已安装 DataKit（DataKit 安装文档）
部署对应语言的 Jaeger Client Library，并配置数据上报地址为 DataKit 提供的地址

配置

使用 Jaeger 采集链路数据发送到 DataKit 一共需要三步：

第一步：在 DataKit 中开启链路数据接收服务

第二步：配置和获取链路数据监听的地址和端口

第三步：按项目需求进行埋点，并配置数据上报地址为 DataKit 的链路数据接收地址

开启链路数据接收服务

进入 DataKit 安装目录下的 conf.d/traceJaeger 目录，复制 traceJaeger.conf.sample 并命名为 traceJaeger.conf。示例如下：如果需要对链路数据添加其他自定义的标签，可以去掉 [inputs.traceJaeger.tags] 的注释，并添加自定义标签。

设置：

[inputs.traceJaeger]
       path = "/api/traces"        # trace数据接收路径，默认与Jeager官方定义路径相同
       [inputs.traceJaeger.tags]   # 自定义标签组
               tag1 = "tag1"       # 自定义标签1
               tag2 = "tag2"       # 自定义标签2
               tag3 = "tag3"       # 自定义标签3

配置和链路数据监听的地址和端口

打开 DataKit 安装目录下的 datakit.conf，找到 http_server_addr 配置项，配置数据接收的监听地址和端口，默认监听地址是 0.0.0.0，端口为 9529

完成以上配置后即可获取链路数据的接收地址：HTTP协议:绑定地址:链路端口/api/traces，例如 DataWay 的地址是 1.2.3.4，配置的监听地址和端口是 0.0.0.0:9529，则链路数据的接收地址是 http://1.2.3.4:9529/api/traces，如果是本机也可以使用 localhost 或 127.0.0.1，如果是内网也可以使用内网地址。

注意：

请求的默认路径是 /api/traces
需保证数据采集端能访问该地址

埋点配置

通过 Jaeger 采集链路数据需要根据当前项目的开发语言对应 Jaeger Client Library，然后按照 Jaeger Client Library 的文档进行埋点，对于 DataFlux 的链路数据采集来说，只需要将 Client Library 的数据上报地址配置为上一步中获取到的链路数据接收地址即可（相当于将 DataKit 当做 Jaeger Server）。

这里以 Jaeger 的 go 客户端作为示范

第一步，引入相关客户端包

import (
    "github.com/opentracing/opentracing-go"
    "github.com/uber/jaeger-client-go/config"
)

第二步，初始化设置

    //获取缺省配置
    cfg,err:= config.FromEnv()
    if err!=nil {
        return
    }

    //设置采样率
    cfg.Sampler=&config.SamplerConfig{
        Type:  "const",// 使用const采样器
        Param: 1,      // 采样所有追踪
    }

    //设置项目名
    tags := []opentracing.Tag {
        {
            Key: "project",
            Value: "myproject",
        },
    }
    cfg.Tags = tags

    //设置服务名
    cfg.ServiceName = "cd"
    reportCfg := new(config.ReporterConfig)

    # 将 CollectorEndpoint 配置为 DataKit 的链路数据接收地址
    reportCfg.CollectorEndpoint = "http://1.2.3.4:9529/api/traces"
    cfg.Reporter                = reportCfg

    //创建Tracer
    tracer, closer, err := cfg.NewTracer()
    if err != nil {
        return
    }

    //设置全局tracer
    opentracing.SetGlobalTracer(tracer)

第三步，埋点创建span信息

    //创建一个span，若创建span时候没有指定父span，则新建一条trace
    clientSpan := opentracing.StartSpan("operationName")
    //向collector发送追踪数据
    defer clientSpan.Finish()

除了在应用初始化时设置项目名外，还可通过如下两种方式设置：

通过环境变量设置

export JAEGER_TAGS="project=your_project_name"

通过采集器自定义标签设置

[inputs.traceJaeger]
     path = "/api/traces"                        # trace数据接收路径，默认与Jeager官方定义路径相同
     [inputs.traceJaeger.tags]                   # 自定义标签组
             project = "your_project_name"       # 设置项目名

示例代码

以 Go 语言作为示例代码，其它编程语言与此类似，示例中 SERVICE_A 调用 SERVICE_B 提供的HTTP服务。

SERVICE_A

package main

import (
    "fmt"
    "net/http"
    "time"

    "github.com/opentracing/opentracing-go"
    "github.com/uber/jaeger-client-go/config"
)

var testCnt = 100

func main(){
    //获取缺省配置
    cfg,err:= config.FromEnv()
    if err!=nil {
        fmt.Println(err)
    }

    //设置采样率
    cfg.Sampler=&config.SamplerConfig{
        Type:  "const",// 使用const采样器
        Param: 1,      // 采样所有追踪
    }
    
    //设置项目名
    tags := []opentracing.Tag {
        {
            Key: "project",
            Value: "myproject",
        },
    }
    cfg.Tags = tags    

    // 设置服务名
    cfg.ServiceName = "SERVICE_A"
    reportCfg := new(config.ReporterConfig)
    reportCfg.CollectorEndpoint = "http://10.100.64.106:9529/api/traces"
    cfg.Reporter                = reportCfg
    // 创建Tracer
    tracer, closer, err := cfg.NewTracer()
    if err != nil {
        fmt.Printf("Create Trace err: %v\n", err)
        return
    }
    defer closer.Close()
    //设置全局tracer
    opentracing.SetGlobalTracer(tracer)

    for i:=0; i < testCnt; i++ {
        app_process()
    }
}

func app_process() {
    //创建一个span，若创建span时候没有指定父span，则此spanid即为traceid
    clientSpan := opentracing.StartSpan("a1")
    //向agent发送追踪数据
    defer clientSpan.Finish()

    //延时一段时间
    time.Sleep((100)*time.Millisecond)

    //创建一个HTTP客户端
    client := &http.Client{
    }

    //创建一个HTTP请求
    req, err := http.NewRequest("GET", "http://127.0.0.1:12345/x", nil)
    if err != nil {
        fmt.Printf("Request err: %v\n", err)
        return
    }

    //当前span设置一个tag
    clientSpan.SetTag("clientSpanTag", "clientSpanValue")
    //当前span设置一个baggage
    clientSpan.SetBaggageItem("clientSpanBaggage", "clientSpanBaggageValue")

    //取得请求头
    carrier := opentracing.HTTPHeadersCarrier(req.Header)
    //把当前span的context上下文写入HTTP请求头，即写入traceid，baggage等信息
    clientSpan.Tracer().Inject(clientSpan.Context(), opentracing.HTTPHeaders, carrier)

    //发送请求
    resp, err := client.Do(req)
    if err != nil {
        fmt.Printf("Get response err: %v\n", err)
        return
    }

    //打印HTTP响应
    fmt.Printf("Response: %v\n", resp)
}

SERVICE_B

package main

import (
    "fmt"
    "net/http"
    "time"

    "github.com/opentracing/opentracing-go"
    "github.com/opentracing/opentracing-go/log"
    "github.com/uber/jaeger-client-go/config"
)


func authMock(s opentracing.Span) {
    //创建authspan，并且为testspan的子span
    authspan := opentracing.StartSpan("auth", opentracing.ChildOf(s.Context()))

    //authspan结束后发送其追踪信息到http://127.0.0.1:12345/y
    defer authspan.Finish()

    //为testspan设置log信息
    authspan.LogFields(
        log.String("event", "start auth"),
        log.String("type", "name/password"),
        log.Int("ID", 1500))

    //模拟操作耗时
    time.Sleep(100*time.Millisecond)
}

func dataMock(s opentracing.Span) {
    //创建dataspan，并且为testspan的子span
    dataspan := opentracing.StartSpan("data", opentracing.ChildOf(s.Context()))

    //authspan结束后发送其追踪信息到http://127.0.0.1:12345/y
    defer dataspan.Finish()

    //为dataspan设置log信息
    dataspan.LogKV("action", "Process")

    //为dataspan设置baggage信息
    dataspan.SetBaggageItem("dataSpanBaggage", "dataSpanBaggageValue")


    //延时一段时间
    time.Sleep(100*time.Millisecond)
}

func indexhandler(w http.ResponseWriter,r *http.Request)  {
    //获取tracer，即通过opentracing.SetGlobalTracer(tracer)设置的tracer
    tracer  := opentracing.GlobalTracer()

    //获取clientspan的context上下文载体
    carrier := opentracing.HTTPHeadersCarrier(r.Header)

    //提取clientspan的context上下文
    clientContext, err := tracer.Extract(opentracing.HTTPHeaders, carrier)
    fmt.Printf("Head: %v\n", r.Header)
    if err != nil {
        fmt.Printf("Extract err: %v\n", err)
    }

    //创建子span，并且为clientspan的子span
    testSpan := opentracing.StartSpan("index", opentracing.ChildOf(clientContext))

    //TestSpan结束后发送其追踪信息到http://127.0.0.1:12345/y
    defer testSpan.Finish()

    //为testspan设置baggage信息
    testSpan.SetBaggageItem("testSpanBaggage", "testSpanBaggageValue")

    //模拟请求验证
    authMock(testSpan)

    //模拟数据访问
    dataMock(testSpan)

    //HTTP响应
    fmt.Fprintln(w,"hello world")
}

func main() {
    //获取缺省配置
    cfg,err:= config.FromEnv()
    if err!=nil {
        fmt.Println(err)
    }
    //设置采样率
    cfg.Sampler=&config.SamplerConfig{
        Type:  "const",// 使用const采样器
        Param: 1,      // 采样所有追踪
    }
    
    //设置项目名
    tags := []opentracing.Tag {
        {
            Key: "project",
            Value: "myproject",
        },
    }
    cfg.Tags = tags    
    
    // 设置服务名
    cfg.ServiceName = "SERVICE_B"


    reportCfg := new(config.ReporterConfig)
    reportCfg.CollectorEndpoint = "http://10.100.64.106:9529/api/traces"
    cfg.Reporter                = reportCfg
    // 创建Tracer
    tracer, closer, err := cfg.NewTracer()
    if err != nil {
        fmt.Println(err)
    }
    defer closer.Close()
    //设置全局tracer
    opentracing.SetGlobalTracer(tracer)

    //设置HTTP服务API路由
    //client 应用发送请求到/x，请求头中带有spancontext信息
    http.HandleFunc("/x", indexhandler)

    http.ListenAndServe("127.0.0.1:12345",nil)
}

采集数据

具体请参考链路数据采集文档中的字段说明

事件

该数据源无关键事件数据

日志

该数据源无日志

FAQ

配置好后为什么在 DataFlux Studio 中看不到数据？

请按照下面的项依次进行检查：

确认更改了 DataKit 的相关配置文件后是否重启 DataKit
请确认在 DataKit 中配置的上报地址是否正确
请确认采集链路数据的网络环境是否能访问 DataKit 的链路数据接收地址
开启 DataKit Debug 模式（在 datakit.conf 中将 log_level 设置为 debug，重启），查看日志 datakit.log，查看是否有链路数据的日志（grep trace）