基于 Consul 服务发现构建动态自适应的 Vector 可观测性管道

可观测性

文章字数: 3.2k

阅读时长: 14 分

在跨多个云服务商部署的动态微服务环境中，维护一套静态的可观测性配置是一项艰巨且易错的任务。每当一个服务实例上线、下线或迁移，都需要手动更新 Prometheus 的抓取目标或类似配置，这不仅效率低下，还极易导致监控盲点或配置漂移。真正的挑战在于，如何让可观测性基础设施能够像服务本身一样，动态地感知整个集群拓扑的变化并自动适应。

我们的目标是构建一个“活”的管道：一个能够自动从服务注册中心学习拓扑、动态配置数据采集代理、并将统一的指标流推送到中心化平台进行可视化的系统。这个系统不应该关心服务运行在哪家云服务商的哪个虚拟机或容器里，只要它在 Consul 中注册，就应该被自动纳入监控。

flowchart TD
    subgraph "Cloud Provider A (e.g., AWS)"
        ServiceA1[Service A - Instance 1]
        ServiceB1[Service B - Instance 1]
        VectorAgentA[Vector Agent]
    end

    subgraph "Cloud Provider B (e.g., Azure)"
        ServiceA2[Service A - Instance 2]
        ServiceC1[Service C - Instance 1]
        VectorAgentB[Vector Agent]
    end

    subgraph "Central Observability Stack"
        Grafana[Grafana]
        Prometheus[Prometheus / Mimir]
    end

    Consul[Consul Cluster]

    ServiceA1 -- "Register with metadata" --> Consul
    ServiceB1 -- "Register with metadata" --> Consul
    ServiceA2 -- "Register with metadata" --> Consul
    ServiceC1 -- "Register with metadata" --> Consul

    VectorAgentA -- "1. Query services API" --> Consul
    VectorAgentB -- "1. Query services API" --> Consul

    Consul -- "2. Return list of local services" --> VectorAgentA
    Consul -- "2. Return list of local services" --> VectorAgentB

    VectorAgentA -- "3. Dynamically configure scrape jobs" --> VectorAgentA
    VectorAgentB -- "3. Dynamically configure scrape jobs" --> VectorAgentB

    VectorAgentA -- "4. Scrape metrics" --> ServiceA1
    VectorAgentA -- "4. Scrape metrics" --> ServiceB1

    VectorAgentB -- "4. Scrape metrics" --> ServiceA2
    VectorAgentB -- "4. Scrape metrics" --> ServiceC1

    VectorAgentA -- "5. Push metrics" --> Prometheus
    VectorAgentB -- "5. Push metrics" --> Prometheus

    Grafana -- "6. Query & Visualize" --> Prometheus

技术选型决策

服务发现: Consul
在我们的现有架构中，Consul 已经是服务注册与发现的标准。它不仅存储了服务的网络位置（IP和端口），还通过健康检查持续维护服务的可用状态。这是我们动态拓扑的唯一真实来源（Single Source of Truth）。
采集代理: Vector
虽然 Prometheus 自带服务发现机制，但在复杂的异构环境中，我们需要一个更灵活、更轻量且厂商中立的代理。Vector 是一个用 Rust 编写的高性能可观测性数据管道。它的杀手级特性是其强大的数据转换能力（Vector Remap Language, VRL）和丰富的源（Source）与汇（Sink）集成。特别是，Vector 内置的 consul_services 源可以直接轮询 Consul API，这正是我们实现动态配置的核心。
可视化: Grafana
Grafana 是业界标准，无需赘述。关键在于如何设计一个能适应服务动态增减的仪表盘。
配置测试: Vitest
动态配置的逻辑会逐渐变得复杂。我们将使用一段 TypeScript 脚本来生成部分复杂的 Vector 配置模板，并使用 Vitest 为这个生成器编写单元测试。在真实项目中，直接手写复杂的 VRL 或 TOML 配置而不进行测试，是导致生产事故的常见原因。

核心实现：动态的 Vector 管道

首先，我们需要一个本地环境来模拟这个场景。使用 Docker Compose 启动 Consul 和几个带 /metrics 端点的示例服务。

docker-compose.yml

version: '3.8'

services:
  consul:
    image: hashicorp/consul:1.15
    container_name: consul
    ports:
      - "8500:8500"
      - "8600:8600/udp"
    command: "agent -server -ui -node=server-1 -bootstrap-expect=1 -client=0.0.0.0"

  # 模拟一个提供 Prometheus 指标的服务
  prometheus-mock-service-A1:
    image: prom/node-exporter:v1.5.0
    container_name: mock_service_a1
    command: '--web.listen-address=:8080'
    # 这个服务在 Consul 中的注册信息
    labels:
      - "SERVICE_NAME=service-a"
      - "SERVICE_TAGS=production,cloud-a"
      - "SERVICE_8080_CHECK_HTTP=/metrics"
      - "SERVICE_8080_CHECK_INTERVAL=10s"

  prometheus-mock-service-B1:
    image: prom/node-exporter:v1.5.0
    container_name: mock_service_b1
    command: '--web.listen-address=:8080'
    labels:
      - "SERVICE_NAME=service-b"
      - "SERVICE_TAGS=staging,cloud-a"
      - "SERVICE_8080_CHECK_HTTP=/metrics"
      - "SERVICE_8080_CHECK_INTERVAL=10s"

  # 使用 registrator 自动将容器注册到 Consul
  registrator:
    image: gliderlabs/registrator:latest
    container_name: registrator
    network_mode: "host" # 需要访问 docker socket 和 consul
    volumes:
      - "/var/run/docker.sock:/tmp/docker.sock"
    command: "consul://localhost:8500"
    depends_on:
      - consul

在这个环境中，registrator 会自动读取容器的 labels 并将其注册到 Consul。我们通过 SERVICE_TAGS 注入了环境和云提供商等关键元数据。

接下来是 Vector 的配置，这是整个系统的核心。

vector.toml

# vector.toml
# 数据源部分
[sources.consul_discovered_services]
  # 关键：使用 consul_services 源来动态发现服务
  type = "consul_services"
  # Consul agent 的地址
  address = "http://consul:8500"
  # 每隔 15 秒轮询一次 Consul API
  interval_secs = 15
  # 定义要发现的服务。星号表示所有服务。
  # 在生产环境中，可以指定具体服务，例如 ["api-gateway", "user-service"]
  services = ["*"]
  # 只包含健康检查通过的服务实例
  passing = true

# 数据转换部分：将 Consul API 的响应转换为 Prometheus 的抓取配置
[transforms.build_prometheus_targets]
  type = "remap"
  inputs = ["consul_discovered_services"]
  # 使用 VRL 进行核心逻辑处理
  source = """
  # . represents the event from consul_services source.
  # It's a JSON object describing a service change (upsert/delete).
  # We only care about services being added or updated.
  if .kind == "upsert" {
      # Extract metadata. Provide defaults to prevent errors.
      service_name = .service.name
      service_id = .service.id
      node_name = .node.name
      address = .service.address
      port = .service.port
      
      # The tags are crucial for labeling in Prometheus.
      # We convert the array of tags into a key-value map for easier access.
      tags, err = object(.service.tags)
      if err != null {
          # Log error if tags are not in key=value format, but don't drop the event.
          log("Failed to parse tags for service " + service_name + ": " + err, level: "warn")
          tags = {}
      }

      # Construct the scrape endpoint.
      # A common mistake is not handling services without a standard /metrics path.
      # Here we assume a convention, but a real-world implementation might get this path from Consul metadata.
      scrape_url = "http://" + address + ":" + to_string(port) + "/metrics"

      # This is where we shape the final event that the prometheus_scrape source will receive.
      # We are essentially creating a new event with a specific structure.
      . = {
          "__meta_consul_service": service_name,
          "__meta_consul_service_id": service_id,
          "__meta_consul_node": node_name,
          "job": service_name,
          "instance": service_id,
          "scrape_url": scrape_url,
          # Forward custom tags as Prometheus labels.
          # This allows us to filter by cloud, environment, etc. in Grafana.
          "cloud": tags.cloud || "unknown",
          "env": tags.env || "default"
      }
  } else {
      # If the service is deleted from Consul, we drop the event.
      # The prometheus_scrape source will eventually remove the target.
      abort
  }
  """

# 采集器部分：一个特殊的 source，它接收配置并执行抓取
[sources.prometheus_scraper]
  type = "prometheus_scrape"
  # 关键：这里的 endpoints 是由 `build_prometheus_targets` transform 动态注入的
  # 当 `build_prometheus_targets` 生成一个新事件时，这个 source 会添加一个抓取目标。
  # target_config.path 必须匹配 transform 输出的字段名
  endpoint_config_path = "build_prometheus_targets"
  scrape_interval_secs = 30

# 数据汇部分：将抓取到的指标发送到远端存储
[sinks.prometheus_remote]
  type = "prometheus_remote_write"
  inputs = ["prometheus_scraper"]
  endpoint = "http://your-prometheus-or-mimir:9090/api/v1/write"
  # 在生产环境中，需要添加认证
  # [sinks.prometheus_remote.auth]
  #   strategy = "basic"
  #   user = "..."
  #   password = "..."

这段配置的精髓在于 consul_services -> remap -> prometheus_scrape 的链条。

sources.consul_discovered_services 定期从 Consul 拉取全量服务列表。对于每个服务的每个实例，它会生成一个事件（upsert 或 delete）。
transforms.build_prometheus_targets 接收这些事件。VRL 脚本是这里的“大脑”，它解析 Consul 返回的复杂 JSON，提取服务名、地址、端口以及我们自定义的 tags，然后重塑成一个包含 scrape_url 和 Prometheus labels 的新事件。这里的错误处理（如 tags, err = object(.service.tags)）在生产环境中至关重要，它能防止格式错误的 tag 导致整个管道崩溃。
sources.prometheus_scraper 是一个特殊的源，它本身不产生数据，而是消费其他流（通过 endpoint_config_path 指定）来动态配置自己的抓取目标。这是 Vector 实现动态抓取的核心机制。

测试配置生成逻辑

当 remap 中的 VRL 逻辑变得复杂时，例如需要根据不同的服务类型应用不同的抓取路径或标签规则，直接在 vector.toml 中维护大段 VRL 脚本会变得难以测试和维护。一个更稳健的方法是使用脚本生成这部分配置。

假设我们用 TypeScript 来编写一个配置生成器。

src/configGenerator.ts

// src/configGenerator.ts

// 模拟从 Consul API 获取的服务实例数据
export interface ConsulServiceInstance {
  kind: 'upsert' | 'delete';
  service: {
    id: string;
    name: string;
    address: string;
    port: number;
    tags: string[];
  };
  node: {
    name: string;
  };
}

// 定义生成的目标 Prometheus 配置结构
export interface PrometheusTarget {
  __meta_consul_service: string;
  __meta_consul_service_id: string;
  __meta_consul_node: string;
  job: string;
  instance: string;
  scrape_url: string;
  cloud: string;
  env: string;
  // 可能还有其他从 tag 派生的标签
  tier?: 'frontend' | 'backend';
}

/**
 * 将 Consul 服务实例转换为 Prometheus 抓取目标配置
 * @param event 从 Vector 的 consul_services source 接收的事件
 * @returns 转换后的 Prometheus 目标对象，如果事件应被忽略则返回 null
 */
export function transformConsulEvent(event: ConsulServiceInstance): PrometheusTarget | null {
  if (event.kind !== 'upsert') {
    return null; // 忽略删除事件
  }

  const { service, node } = event;

  // 一个常见的错误是没有处理空的 address 字段，当 Consul 发现服务但网络不可达时可能发生
  if (!service.address || service.port <= 0) {
    console.warn(`Invalid network address for service ${service.id}: ${service.address}:${service.port}`);
    return null;
  }

  const tags: Record<string, string> = {};
  for (const tag of service.tags) {
    const [key, value] = tag.split('=');
    if (key && value) {
      tags[key] = value;
    }
  }

  const scrapePath = tags.metrics_path || '/metrics';
  const scrapeUrl = `http://${service.address}:${service.port}${scrapePath}`;

  const target: PrometheusTarget = {
    __meta_consul_service: service.name,
    __meta_consul_service_id: service.id,
    __meta_consul_node: node.name,
    job: service.name,
    instance: service.id,
    scrape_url: scrapeUrl,
    cloud: tags.cloud || 'unknown',
    env: tags.env || 'default',
  };
  
  // 复杂的业务逻辑：根据 tag 决定服务层级
  if (tags.tier === 'frontend' || tags.tier === 'backend') {
    target.tier = tags.tier;
  }

  return target;
}

现在，我们可以用 Vitest 来为这个转换逻辑编写单元测试。

src/configGenerator.test.ts

// src/configGenerator.test.ts
import { describe, it, expect } from 'vitest';
import { transformConsulEvent, ConsulServiceInstance } from './configGenerator';

describe('transformConsulEvent', () => {
  
  it('should transform a standard service upsert event correctly', () => {
    const event: ConsulServiceInstance = {
      kind: 'upsert',
      service: {
        id: 'service-a-1',
        name: 'service-a',
        address: '10.0.1.10',
        port: 8080,
        tags: ['env=production', 'cloud=aws', 'tier=backend', 'metrics_path=/custom-metrics'],
      },
      node: { name: 'node-1' },
    };

    const result = transformConsulEvent(event);

    expect(result).not.toBeNull();
    expect(result).toEqual({
      __meta_consul_service: 'service-a',
      __meta_consul_service_id: 'service-a-1',
      __meta_consul_node: 'node-1',
      job: 'service-a',
      instance: 'service-a-1',
      scrape_url: 'http://10.0.1.10:8080/custom-metrics',
      cloud: 'aws',
      env: 'production',
      tier: 'backend',
    });
  });

  it('should return null for delete events', () => {
    const event: ConsulServiceInstance = {
      kind: 'delete',
      service: { id: 'service-a-1', name: 'service-a', address: '', port: 0, tags: [] },
      node: { name: 'node-1' },
    };
    expect(transformConsulEvent(event)).toBeNull();
  });

  it('should handle missing or malformed tags gracefully', () => {
    const event: ConsulServiceInstance = {
      kind: 'upsert',
      service: {
        id: 'service-b-1',
        name: 'service-b',
        address: '10.0.2.20',
        port: 9090,
        tags: ['env=staging', 'malformed-tag'], // 'cloud' tag is missing
      },
      node: { name: 'node-2' },
    };

    const result = transformConsulEvent(event);
    expect(result?.cloud).toBe('unknown');
    expect(result?.env).toBe('staging');
    expect(result?.scrape_url).toBe('http://10.0.2.20:9090/metrics'); // Uses default path
    expect(result?.tier).toBeUndefined();
  });

  it('should return null for services with invalid network details', () => {
    const event: ConsulServiceInstance = {
      kind: 'upsert',
      service: {
        id: 'service-c-1',
        name: 'service-c',
        address: '', // Invalid address
        port: 8000,
        tags: [],
      },
      node: { name: 'node-3' },
    };
    expect(transformConsulEvent(event)).toBeNull();
  });
});

通过这种方式，我们可以将核心的、易变的业务逻辑从静态配置文件中剥离出来，用更强大的编程语言进行实现，并用单元测试保证其正确性。最终，可以通过 CI/CD 流程运行这个脚本来生成 vector.vrl 文件，再注入到 Vector 的主配置中。

Grafana 动态仪表盘

有了富含元数据（job, instance, cloud, env）的指标流，我们可以在 Grafana 中创建动态仪表盘。

创建变量 (Variables):
在仪表盘设置中，创建几个查询变量：
- $cloud: label_values(up, cloud)
- $env: label_values(up{cloud="$cloud"}, env)
- $job: label_values(up{cloud="$cloud", env="$env"}, job)
配置面板:
在面板的查询中使用这些变量，并设置 Repeat 选项。例如，创建一个显示 CPU 使用率的面板：
- 查询: rate(process_cpu_seconds_total{job="$job", env="$env", cloud="$cloud"}[5m])
- 在面板的 “Repeat options” 中，设置 “Repeat for” 为 $job 变量，方向为水平。

这样，当你在仪表盘顶部选择一个云提供商和一个环境后，Grafana 会自动查询 Consul 中注册过的所有服务（jobs），并为每个服务动态生成一个面板。当新服务 service-d 在该环境中上线并被 Vector 发现后，无需修改仪表盘，它会自动出现在上面。

方案的局限性与未来展望

这套方案虽然实现了高度自动化，但在真实生产环境中，它并非没有边界。

首先，系统的韧性与 Consul 集群的健康状况强相关。Consul 的抖动或不可用，将直接导致 Vector 无法更新其抓取目标，可能造成短暂的监控数据丢失。为缓解此问题，需要确保 Consul 集群自身的高可用性。

其次，对于服务实例变更极为频繁（例如每秒钟都有大量实例启停）的场景，Consul API 的轮询模式可能会带来一定的延迟。Vector 的配置热加载虽然高效，但过于频繁的更新也可能增加代理自身的负载。在这种极端场景下，或许需要探索基于 Consul Watch 的事件驱动推送模型，而非轮询。

最后，当前的实现为所有服务提供了统一的指标采集，但对于需要深度定制、采集非标准指标或进行复杂多阶段抓取的应用，仍然需要更精细化的配置。一个可行的优化路径是，在 Consul 的服务元数据中定义更丰富的抓取指令，然后让 VRL 脚本或配置生成器去解析这些指令，从而为不同类型的服务生成差异化的抓取配置，实现更深层次的自适应。

Grafana Consul Vector 云服务商 Vitest

采用TDD方法构建集成Pandas、Milvus与MobX的实时特征探索系统

2023-11-15 架构与设计

FastAPI MobX TDD Pandas Milvus

构建基于 MLflow 与 Google Cloud Functions 的可观测 Serverless 推理架构

2023-10-27 MLOps

MLflow Google Cloud Functions 可观测性 Serverless OpenTelemetry