^-^
【go工程组件】分布式链路追踪OpenTelemetry(施工)
最近更新：2025-01-09   |   字数总计：1.9k   |   阅读估时：10分钟   |   阅读量：次
  1. 分布式链路追踪OpenTelemetry
    1. 关键概念
    2. 多种导出器实现trace导出
    3. 多种导出器实现metric导出
    4. otel collector采集配置
    5. 微服务场景下追踪与度量中间件编程
    6. 微服务链路关联与数据传递

分布式链路追踪OpenTelemetry

  • 简称OTel，与供应商无关的开源可观测性框架，用于检测、生成、收集和导出遥测数据。
  • 能作什么？
    1. 每种语言提供了相对应的lib库
    2. 提供了中立的数据收集器，并支持多种方式部署
    3. 生成、发送、收集、处理和导出遥测数据
    4. 可以通过配置将数据并行发送到多个目的地
    5. 提供OpenTracing和OpenCensus项目提供垫片以向后兼容OpenTracing与OpenCensus项目

关键概念

  1. Traces
  2. Metrics
  3. logs

多种导出器实现trace导出

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
// /trace-basic/main.go
package main

import (
	"context"
	"errors"
	"flag"
	"fmt"
	"log"
	"time"
	"trace-basoc/bus"

	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/exporters/jaeger"
	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp"
	"go.opentelemetry.io/otel/exporters/zipkin"
	"go.opentelemetry.io/otel/sdk/resource"
	sdktrace "go.opentelemetry.io/otel/sdk/trace"
	semconv "go.opentelemetry.io/otel/semconv/v1.17.0"
	"go.opentelemetry.io/otel/trace"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
)

// span 有三个，一个方法就是一个span；main、sum、product都是span
// tracerProvider只有一个
// tracer由provider创建，一个包对应一个tracer，所以有2个tracer

// 在启用otl之前需要启动jaeger
// docker run -d --name jaeger \
//    -e COLLECTOR_ZIPKIN_HOST_PORT=:9411 \
//    -e COLLECTOR_OTLP_ENABLED=true \
//    -p  6831:6831/udp \
//    -p  6832:6832/udp \
//    -p  5778:5778 \
//    -p  16686:16686 \
//    -p  4317:4317 \
//    -p  4318:4318 \
//    -p  14250:14250 \
//    -p  14268:14268 \
//    -p  14269:14269 \
//    -p  9411:9411 \
//   quay.io/jaegertracing/all-in-one:1.65.0

// 浏览器访问 http://localhost:16686
// jaeger api http://localhost:14268/api/traces
// zipkin api http://localhost:9411/api/v2/spans
// otlp http api http://localhost:4318
// otlp grpc api http://localhost:4317

// go get go.opentelemetry.io/otel/sdk
// go get go.opentelemetry.io/otel/trace
// go get go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc
// go get go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp
// go get go.opentelemetry.io/otel/exporters/jaeger
// go get go.opentelemetry.io/otel/exporters/zipkin

var (
	otlpGrpcEndpoint = flag.String("otlp-grpc", "", "otlp协议grpc导出地址，例如：192.168.239.154: 4317")
	otlpHttpEndpoint = flag.String("otlp-http", "", "otlp协议http导出地址，例如：192.168.239.154: 4318")
	jaegerEndpoint   = flag.String("jaeger", "", "jaeger导出地址，例如：http://localhost:14268/api/traces")
	zipkinEndpoint   = flag.String("zipkin", "", "zipkin导出地址，例如：http://localhost:9411/api/v2/spans")
)

// go run main.go --otlp-grpc 134.175.250.62:4317

func main() {
	flag.Parse()
	// 初始化provider
	shutDown, err := initProvider(*otlpGrpcEndpoint, *otlpHttpEndpoint, *jaegerEndpoint, *zipkinEndpoint)
	if err != nil {
		log.Fatal(err)
	}
	defer shutDown(context.Background())
	ctx := context.Background()
	// 创建main包的tracer
	mainTrace := otel.Tracer("main-tracer")
	ctx, span := mainTrace.Start(ctx, "main", trace.WithAttributes(attribute.Key("method").String("main.main")))
	defer span.End()
	// 创建bus包的tracer，并且传给bus包
	busTrace := otel.Tracer("bus-tracer")

	// tracer创建完成后，tracer创建span
	b := bus.NewBus(busTrace)

	for i := 0; i < 5; i++ {
		c := b.Sum(ctx, i, i+1)
		fmt.Println(c)
		<-time.After(time.Microsecond * 150)
		c = b.Product(ctx, i, i+1)
		<-time.After(time.Microsecond * 150)
		fmt.Println(c)
	}
}

// 1. 初始化添加到provider的资源信息
// 2. 初始化Exporter导出器
// 3. 初始化tracer provider
// 4. 设置全局的tracer provider
func initProvider(otlpGrpcEndpoint, otlpHttpEndpoint, jaegerEndpoint, zipkinEndpoint string) (func(context.Context) error, error) {
	// 初始化资源
	ctx := context.Background()
	res, err := resource.New(ctx, resource.WithOS(), resource.WithHost(), resource.WithAttributes(
		semconv.ServiceName("trace-basic"),
		semconv.ServiceVersion("1.0.0"),
		attribute.String("env", "dev"),
		attribute.String("author", "lin"),
	))
	if err != nil {
		log.Println(err)
		return nil, err
	}
	// 初始化导出器
	var traceExporter sdktrace.SpanExporter
	{
		ctx, cancel := context.WithTimeout(ctx, time.Second)
		defer cancel()
		if otlpGrpcEndpoint != "" {
			conn, err := grpc.DialContext(ctx, otlpGrpcEndpoint, grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithBlock())
			if err != nil {
				log.Println(err)
				return nil, err
			}
			traceExporter, err = otlptracegrpc.New(ctx, otlptracegrpc.WithGRPCConn(conn), otlptracegrpc.WithInsecure())
			if err != nil {
				log.Println(err)
				return nil, err
			}
		} else if otlpHttpEndpoint != "" {
			traceExporter, err = otlptracehttp.New(ctx, otlptracehttp.WithEndpoint(otlpHttpEndpoint), otlptracehttp.WithInsecure())
			if err != nil {
				log.Println(err)
				return nil, err
			}
		} else if jaegerEndpoint != "" {
			traceExporter, err = jaeger.New(jaeger.WithCollectorEndpoint(jaeger.WithEndpoint(jaegerEndpoint)))
			if err != nil {
				log.Println(err)
				return nil, err
			}
		} else if zipkinEndpoint != "" {
			traceExporter, err = zipkin.New(jaegerEndpoint)
			if err != nil {
				log.Println(err)
				return nil, err
			}
		} else {
			err := errors.New("没用可用的exporter端点")
			log.Println(err)
			return nil, err
		}
	}
	// 初始化provider
	tracerProvider := sdktrace.NewTracerProvider(
		sdktrace.WithSampler(sdktrace.AlwaysSample()),
		sdktrace.WithResource(res),
		sdktrace.WithSpanProcessor(sdktrace.NewBatchSpanProcessor(traceExporter)),
	)
	// 设置全局provider
	otel.SetTracerProvider(tracerProvider)
	return tracerProvider.Shutdown, nil
}
// /trace-basic/bus.go
package bus

import (
	"context"
	"time"

	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/trace"
)

type Bus interface {
	Sum(ctx context.Context, a, b int) int
	Product(ctx context.Context, a, b int) int
}

type bus struct {
	tracer trace.Tracer
}

func NewBus(tracer trace.Tracer) Bus {
	return &bus{
		tracer: tracer,
	}
}

func (bus *bus) Sum(ctx context.Context, a, b int) int {
	_, span := bus.tracer.Start(ctx, "sum", trace.WithAttributes(attribute.Int("a", a), attribute.Int("b", b)))
	defer span.End()
	c := a + b
	span.SetAttributes(attribute.Int("c", c))
	<-time.After(time.Microsecond * 100)
	return c
}

func (bus *bus) Product(ctx context.Context, a, b int) int {
	_, span := bus.tracer.Start(ctx, "product", trace.WithAttributes(attribute.Int("a", a), attribute.Int("b", b)))
	defer span.End()
	c := a * b
	span.SetAttributes(attribute.Int("c", c))
	<-time.After(time.Microsecond * 200)
	return c
}

alt text

多种导出器实现metric导出

  • /metric-basic/main.go
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    86
    87
    88
    89
    90
    91
    92
    93
    94
    95
    96
    97
    98
    99
    100
    101
    102
    103
    104
    105
    106
    107
    108
    109
    110
    111
    112
    113
    114
    115
    116
    117
    118
    119
    120
    121
    122
    123
    124
    125
    126
    127
    128
    129
    130
    131
    132
    133
    134
    135
    136
    137
    138
    139
    140
    141
    142
    143
    144
    145
    146
    147
    148
    149
    150
    151
    152
    153
    154
    155
    156
    157
    158
    159
    160
    161
    162
    163
    164
    165
    166
    167
    168
    169
    170
    171
    172
    173
    174
    175
    176
    177
    178
    179
    180
    181
    182
    183
    184
    185
    186
    187
    188
    189
    package main

    import (
    	"context"
    	"flag"
    	"log"
    	"math/rand"
    	"net/http"
    	"os"
    	"os/signal"
    	"time"

    	"github.com/prometheus/client_golang/prometheus/promhttp"
    	"go.opentelemetry.io/otel"
    	"go.opentelemetry.io/otel/attribute"
    	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc"
    	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp"
    	"go.opentelemetry.io/otel/exporters/prometheus"
    	api "go.opentelemetry.io/otel/metric"
    	"go.opentelemetry.io/otel/sdk/instrumentation"
    	"go.opentelemetry.io/otel/sdk/metric"
    	"go.opentelemetry.io/otel/sdk/resource"
    	semconv "go.opentelemetry.io/otel/semconv/v1.17.0"
    	"google.golang.org/grpc"
    	"google.golang.org/grpc/credentials/insecure"
    )

    // go get go.opentelemetry.io/otel/exporters/prometheus
    // go get go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp
    // go get go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc
    // go get go.opentelemetry.io/otel/sdk/metric/aggregation

    // go run .
    var (
    	otlpGrpcEndpoint = flag.String("otlp-grpc", "", "otlp协议grpc导出地址，例如：192.168.239.154: 4317")
    	otlpHttpEndpoint = flag.String("otlp-http", "", "otlp协议http导出地址，例如：192.168.239.154: 4318")
    )

    func main() {
    	flag.Parse()
    	ctx := context.Background()
    	// 初始化meter provider 全局仅一次
    	shutdown, err := initProvider(*otlpGrpcEndpoint, *otlpHttpEndpoint)
    	if err != nil {
    		log.Fatal(err)
    	}
    	defer shutdown(ctx)
    	// 通过provider创建meter，可以基于应用或服务创建，也可以基于lib库或者模块创建
    	// 通过meter创建具体的指标（cunter、gauge、直方图（histogram））
    	getMetrics()
    	// 启动promethues http
    	if *otlpGrpcEndpoint == "" && *otlpHttpEndpoint == "" {
    		go serveMetrics()
    	}
    	ctx, stop := signal.NotifyContext(ctx, os.Kill, os.Interrupt)
    	defer stop()
    	<-ctx.Done()
    }

    // 1. 初始化添加到provider的资源信息
    // 2. 初始化Exporter导出器
    // 3. 初始化meter provider
    // 4. 设置全局的meter provider
    func initProvider(otlpGrpcEndpoint, otlpHttpEndpoint string) (func(context.Context) error, error) {
    	// 初始化资源
    	ctx := context.Background()
    	res, err := resource.New(ctx, resource.WithOS(), resource.WithHost(), resource.WithAttributes(
    		semconv.ServiceName("meter-basic"),
    		semconv.ServiceVersion("1.0.0"),
    		attribute.String("env", "dev"),
    		attribute.String("author", "lin"),
    	))
    	if err != nil {
    		log.Println(err)
    		return nil, err
    	}
    	var metricReader metric.Reader
    	{
    		if otlpGrpcEndpoint != "" {
    			conn, err := grpc.DialContext(ctx, otlpGrpcEndpoint, grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithBlock())
    			if err != nil {
    				log.Println(err)
    				return nil, err
    			}
    			exporter, err := otlpmetricgrpc.New(ctx, otlpmetricgrpc.WithGRPCConn(conn), otlpmetricgrpc.WithInsecure())
    			if err != nil {
    				log.Println(err)
    				return nil, err
    			}
    			metricReader = metric.NewPeriodicReader(exporter, metric.WithInterval(time.Second*5))
    		} else if otlpHttpEndpoint != "" {
    			exporter, err := otlpmetrichttp.New(context.Background(), otlpmetrichttp.WithEndpoint(otlpHttpEndpoint), otlpmetrichttp.WithInsecure())
    			if err != nil {
    				log.Println(err)
    				return nil, err
    			}
    			metricReader = metric.NewPeriodicReader(exporter, metric.WithInterval(time.Second*5))
    		} else {
    			// 初始化Prometheus导出器
    			metricReader, err = prometheus.New()
    			if err != nil {
    				log.Println(err)
    				return nil, err
    			}
    		}
    	}
    	view := metric.NewView(metric.Instrument{
    		// 指定作用的指标名称
    		Name: "custom_histogram",
    		// 作用范围，仅针对metrics-basic-meter
    		Scope: instrumentation.Scope{Name: "metrics-basic-meter"},
    	}, metric.Stream{
    		Name: "myhistogram",
    		Aggregation: metric.AggregationExplicitBucketHistogram{
    			Boundaries: []float64{2, 4, 8, 16, 32, 64, 128, 256, 512},
    		},
    	})
    	// 初始化provider
    	provider := metric.NewMeterProvider(metric.WithResource(res), metric.WithReader(metricReader), metric.WithView(view))
    	// 设置全局provider
    	otel.SetMeterProvider(provider)
    	return provider.Shutdown, nil
    }

    // 获取系统及自定义指标
    func getMetrics() {
    	ctx := context.Background()
    	meter := otel.Meter("metrics-basic-meter")
    	attrs := []attribute.KeyValue{
    		attribute.Key("A").String("B"),
    		attribute.Key("C").String("D"),
    	}
    	// counter，累计指标
    	counter, err := meter.Float64Counter("counter", api.WithDescription("累计指标"))
    	if err != nil {
    		log.Fatal(err)
    	}
    	counter.Add(ctx, 5, api.WithAttributes(attrs...))
    	// 异步counter，每次收集数据时才会更新最新值
    	_, err = meter.Float64ObservableCounter("counter1", api.WithDescription("异步累计指标"), api.WithFloat64Callback(func(ctx context.Context, fo api.Float64Observer) error {
    		fo.Observe(float64(time.Now().Unix()))
    		return nil
    	}))
    	if err != nil {
    		log.Fatal(err)
    	}
    	// gauge实时指标，每次获取数据的时候获取当前值
    	rng := rand.New(rand.NewSource(time.Now().UnixMicro()))
    	gauge, err := meter.Float64ObservableGauge("gauge", api.WithDescription("实时指标"))
    	if err != nil {
    		log.Fatal(err)
    	}
    	meter.RegisterCallback(func(ctx context.Context, o api.Observer) error {
    		n := rng.Intn(100)
    		o.ObserveFloat64(gauge, float64(n))
    		return nil
    	}, gauge)
    	// histogram直方图
    	histogram, err := meter.Float64Histogram("histogram", api.WithDescription("直方图"))
    	if err != nil {
    		log.Fatal(err)
    	}
    	histogram.Record(ctx, 233)
    	histogram.Record(ctx, 23)
    	histogram.Record(ctx, 253)
    	histogram.Record(ctx, 73)
    	histogram.Record(ctx, 8)
    	histogram.Record(ctx, 238)

    	cushistogram, err := meter.Float64Histogram("custom_histogram", api.WithDescription("自定义视图直方图"))
    	if err != nil {
    		log.Fatal(err)
    	}
    	cushistogram.Record(ctx, 233)
    	cushistogram.Record(ctx, 23)
    	cushistogram.Record(ctx, 253)
    	cushistogram.Record(ctx, 73)
    	cushistogram.Record(ctx, 8)
    	cushistogram.Record(ctx, 238)
    }

    // 启动prometheus系统指标采集点
    func serveMetrics() {
    	http.Handle("/metrics", promhttp.Handler())
    	err := http.ListenAndServe(":2223", nil)
    	if err != nil {
    		log.Fatal(err)
    	}
    }

otel collector采集配置

微服务场景下追踪与度量中间件编程

微服务链路关联与数据传递

2024-12-13 该篇文章被 Cleofwine 归为分类: Go云原生

上一篇：【go分布式中间件】Redis 下一篇：【go工程组件】配置解决方案viper
© 2023-2026 Cleofwine / Email / GitHub / 浙ICP备2024136127号
🌙赠我弯弯一枚月，也赠予我晚星⭐ 本站由Hexo驱动｜使用Hexo-theme-A4主题