OpenTelemetry Collector 管线

概念定义

应用发射、后端消费，collector 坐在中间并拥有两者之外的一切：协议翻译、批处理、富化、过滤、采样、扇出、重试。

工作机制

管线由接收器（OTLP、Jaeger、syslog）、处理器（batch、attributes、k8s-metadata、尾采样、filter）与导出器（OTLP 到各厂商、Prometheus）组成。队列与重试设置限定内存并熬过后端故障；网关层聚合各 agent、昂贵策略只执行一次、再扇出到多个后端。

使用场景

永远作为后端迁移与厂商评估的解耦层；需要尾采样、成本过滤或集中脱敏时加网关层；节点本地发货延迟与韧性要紧时上 agent。

常见误区

用 OpenTelemetry 不强制用 collector——SDK 可直连后端；collector 附加的是路由、处理与生命周期解耦。横切策略——脱敏、采样、成本控制——属于一处而不是四十个服务。collector 自身会成为关键组件：要监控它自己的 RED 指标与队列深度，否则遥测断流会伪装成应用沉默；把尾采样这类昂贵处理器塞进每个节点 agent 会把路由器变瓶颈。

代码示例

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/redis/go-redis/v9"
)

func seedAndMeasure(ctx context.Context, client *redis.Client) error {
	keys := []string{"metric:hits", "metric:misses", "metric:errors"}

	// 独立命令走管道：一次网络往返
	pipe := client.Pipeline()
	incrCmds := make([]*redis.IntCmd, len(keys))
	for i, key := range keys {
		incrCmds[i] = pipe.Incr(ctx, key)
	}
	if _, err := pipe.Exec(ctx); err != nil {
		return fmt.Errorf("pipeline exec: %w", err)
	}

	// 事务管道：命令打包进 MULTI/EXEC 原子执行
	tx := client.TxPipeline()
	expireCmds := make([]*redis.BoolCmd, len(keys))
	for i, key := range keys {
		expireCmds[i] = tx.Expire(ctx, key, time.Hour)
	}
	if _, err := tx.Exec(ctx); err != nil {
		return fmt.Errorf("tx exec: %w", err)
	}

	for i, key := range keys {
		fmt.Printf("%s=%d expires=%v\n",
			key, incrCmds[i].Val(), expireCmds[i].Val())
	}
	return nil
}

func main() {
	client := redis.NewClient(&redis.Options{
		Addr:        "127.0.0.1:6379",
		PoolSize:    10,
		ReadTimeout: 500 * time.Millisecond,
	})
	defer client.Close()

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()

	if err := seedAndMeasure(ctx, client); err != nil {
		fmt.Println("failed:", err)
	}
}

import { createReadStream, createWriteStream } from 'node:fs';
import { Transform } from 'node:stream';
import { pipeline } from 'node:stream/promises';
import { createGzip } from 'node:zlib';

class LineCount extends Transform {
  constructor() {
    super({ objectMode: false });
    this.lines = 0;
  }

  _transform(chunk, encoding, callback) {
    const text = chunk.toString('utf8');
    this.lines += (text.match(/\n/g) ?? []).length;
    this.push(text.toUpperCase());
    callback();
  }

  _flush(callback) {
    this.push('\n# total lines: ' + this.lines + '\n');
    callback();
  }
}

await pipeline(
  createReadStream('input.log', { encoding: 'utf8', highWaterMark: 64 * 1024 }),
  new LineCount(),
  createGzip({ level: 6 }),
  createWriteStream('output.log.gz'),
);

console.log('pipeline finished');