Pub/Sub OpenTelemetry 跟踪

借助 OpenTelemetry 跟踪,您可以识别和跟踪各种 Pub/Sub 客户端库操作(例如批处理、租约管理和流控制)的延迟时间。收集此信息有助于您调试客户端库问题。

OpenTelemetry 跟踪的一些潜在用例包括:

  • 您的服务发布延迟时间高于正常水平。
  • 您收到的消息重新传送次数过多。
  • 对订阅者客户端的回调函数进行更改后,处理时间比平时长。

准备工作

在配置 OpenTelemetry 之前,请完成以下任务:

所需的角色

为确保服务帐号具有将跟踪记录导出到 Cloud Trace 所需的 权限, 请让您的管理员为服务帐号授予项目的 以下 IAM 角色:

如需详细了解如何授予角色,请参阅管理对项目、文件夹和组织的访问权限

这些预定义角色包含 将跟踪记录导出到 Cloud Trace 所需的权限。如需查看所需的确切权限,请展开所需权限部分:

所需权限

将跟踪记录导出到 Cloud Trace 需要以下权限:

  • 全部: cloudtrace.traces.patch

您的管理员也可以使用自定义角色或其他预定义角色为服务帐号授予这些权限。

OpenTelemetry 跟踪工作流

如需设置 OpenTelemetry 跟踪,您可以使用 Pub/Sub 客户端库和 OpenTelemetry SDK。使用 SDK 时,您必须先设置跟踪记录导出器和跟踪器提供程序,然后才能连接到 Pub/Sub 库。在某些库中,设置跟踪器提供程序是可选的。

  • 跟踪记录导出器。OpenTelemetry SDK 使用跟踪记录导出器来确定将跟踪记录发送到何处。

  • 跟踪器提供程序。Pub/Sub 客户端库使用跟踪器提供程序来创建跟踪记录。

以下步骤简要介绍了如何设置跟踪:

  1. 实例化 Cloud Trace OpenTelemetry 导出器。
  2. 如果需要,请使用 OpenTelemetry SDK 实例化并注册跟踪器提供程序。
  3. 使用“启用 OpenTelemetry 跟踪”选项配置客户端。
  4. 使用 Pub/Sub 客户端库发布消息。

跟踪功能的工作原理

对于发布的每条消息,客户端库都会创建一个新的跟踪记录。此跟踪记录表示消息的整个生命周期,从您发布消息的那一刻到消息被确认的那一刻。跟踪记录封装了操作时长、父 span 和子 span 以及关联的 span 等信息。

跟踪记录由根 span 及其对应的子 span 组成。这些 span 表示客户端库在处理消息时所做的工作。每条消息跟踪记录都包含以下内容:

  • 对于发布 。流控制、排序键调度、批处理以及发布 RPC 的长度。
  • 对于订阅 。并发控制、排序键调度和租约管理。

为了将信息从发布端传播到订阅端,客户端库会在发布端注入跟踪记录专用属性。只有在启用跟踪记录且以 googclient_ 前缀开头时,上下文传播机制才会启用。

发布带有跟踪记录的消息

以下代码示例展示了如何使用 Pub/Sub 客户端库和 OpenTelemetry SDK 启用跟踪记录。在此示例中,跟踪记录结果会导出到 Cloud Trace。

注意事项

实例化跟踪器提供程序时,您可以使用 OpenTelemetry SDK 配置采样率。此比率决定了 SDK 应采样多少跟踪记录。较低的采样率有助于降低结算费用,并防止 您的服务超出 Cloud Trace span 配额

Go

import (
	"context"
	"fmt"
	"io"

	"cloud.google.com/go/pubsub/v2"
	"go.opentelemetry.io/otel"
	"google.golang.org/api/option"

	texporter "github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/trace"
	"go.opentelemetry.io/otel/sdk/resource"
	sdktrace "go.opentelemetry.io/otel/sdk/trace"
	semconv "go.opentelemetry.io/otel/semconv/v1.26.0"
)

// publishOpenTelemetryTracing publishes a single message with OpenTelemetry tracing
// enabled, exporting to Cloud Trace.
func publishOpenTelemetryTracing(w io.Writer, projectID, topicID string, sampling float64) error {
	// projectID := "my-project-id"
	// topicID := "my-topic"
	ctx := context.Background()

	exporter, err := texporter.New(texporter.WithProjectID(projectID),
		// Disable spans created by the exporter.
		texporter.WithTraceClientOptions(
			[]option.ClientOption{option.WithTelemetryDisabled()},
		),
	)
	if err != nil {
		return fmt.Errorf("error instantiating exporter: %w", err)
	}

	resources := resource.NewWithAttributes(
		semconv.SchemaURL,
		semconv.ServiceNameKey.String("publisher"),
	)

	// Instantiate a tracer provider with the following settings
	tp := sdktrace.NewTracerProvider(
		sdktrace.WithBatcher(exporter),
		sdktrace.WithResource(resources),
		sdktrace.WithSampler(
			sdktrace.ParentBased(sdktrace.TraceIDRatioBased(sampling)),
		),
	)

	defer tp.ForceFlush(ctx) // flushes any pending spans
	otel.SetTracerProvider(tp)

	// Create a new client with tracing enabled.
	client, err := pubsub.NewClientWithConfig(ctx, projectID, &pubsub.ClientConfig{
		EnableOpenTelemetryTracing: true,
	})
	if err != nil {
		return fmt.Errorf("pubsub: NewClient: %w", err)
	}
	defer client.Close()

	// client.Publisher can be passed a topic ID (e.g. "my-topic") or
	// a fully qualified name (e.g. "projects/my-project/topics/my-topic").
	// If a topic ID is provided, the project ID from the client is used.
	// Reuse this publisher for all publish calls to send messages in batches.
	publisher := client.Publisher(topicID)
	result := publisher.Publish(ctx, &pubsub.Message{
		Data: []byte("Publishing message with tracing"),
	})
	if _, err := result.Get(ctx); err != nil {
		return fmt.Errorf("pubsub: result.Get: %w", err)
	}
	fmt.Fprintln(w, "Published a traced message")
	return nil
}

C++

// Create a few namespace aliases to make the code easier to read.
namespace gc = ::google::cloud;
namespace otel = gc::otel;
namespace pubsub = gc::pubsub;

// This example uses a simple wrapper to export (upload) OTel tracing data
// to Google Cloud Trace. More complex applications may use different
// authentication, or configure their own OTel exporter.
auto project = gc::Project(project_id);
auto configuration = otel::ConfigureBasicTracing(project);

auto publisher = pubsub::Publisher(pubsub::MakePublisherConnection(
    pubsub::Topic(project_id, topic_id),
    // Configure this publisher to enable OTel tracing. Some applications may
    // chose to disable tracing in some publishers or to dynamically enable
    // this option based on their own configuration.
    gc::Options{}.set<gc::OpenTelemetryTracingOption>(true)));

// After this point, use the Cloud Pub/Sub C++ client library as usual.
// In this example, we will send a few messages and configure a callback
// action for each one.
std::vector<gc::future<void>> ids;
for (int i = 0; i < 5; i++) {
  auto id = publisher.Publish(pubsub::MessageBuilder().SetData("Hi!").Build())
                .then([](gc::future<gc::StatusOr<std::string>> f) {