使用客户端指标排查高延迟时间问题

虽然 Memorystore for Redis 提供实时服务器端指标来监控吞吐量、CPU 利用率和内存用量，但仅凭这些数据可能无法解释为什么您的客户端应用在复杂的分布式系统中会遇到高延迟。

客户端指标通过提供对完整请求-响应周期的透明度来解决此问题。它们会衡量从应用启动命令到应用处理响应的时间。通过捕获这些数据点，您可以准确确定延迟是源自应用逻辑、网络路径还是 Redis 服务器。

准备工作

确保您的客户端应用使用服务帐号，并且已为其分配以下 Identity and Access Management (IAM) 角色：

  • roles/cloudtrace.agent (Cloud Trace Agent)
  • roles/monitoring.metricWriter (Monitoring Metric Writer)

如需详细了解如何授予角色，请参阅使用 Google Cloud 控制台授予 IAM 角色快速入门。

启用 Cloud Monitoring API

如需将客户端指标导出到 Monitoring， 您的应用需要启用 Monitoring API。 通过在 Monitoring 中导出和可视化这些指标，您可以找出瓶颈的根本原因，以确定延迟的来源。

如需启用 Monitoring API，请执行以下操作：

  1. 在 Google Cloud 控制台中，前往 API 和服务 页面。

    前往“API 和服务”

  2. 选择您在其中创建 Memorystore for Redis 实例的项目。

  3. 点击启用 API 和服务 。

  4. 搜索 monitoring。

  5. 在搜索结果中，点击 Cloud Monitoring API 。

  6. 如果显示 API 已启用 ，则表示此 API 已经启用。否则，请点击启用 。

启用 Cloud Trace API

如需在 Trace 中查看分布式跟踪记录，您必须 启用 Trace API。然后，您可以使用 Trace Explorer 查看这些跟踪记录、诊断瓶颈并隔离应用中的延迟来源。

如需启用 Trace API，请执行以下操作：

  1. 在 Google Cloud 控制台中，前往 API 和服务 页面。

    前往“API 和服务”

  2. 选择您在其中创建 Memorystore for Redis 实例的项目。

  3. 点击启用 API 和服务 。

  4. 搜索 trace。

  5. 在搜索结果中，点击 Cloud Trace API 。

  6. 如果显示 API 已启用 ，则表示此 API 已经启用。否则，请点击启用 。

启用客户端指标

如需启用客户端指标，请将 OpenTelemetry SDK、Cloud Monitoring 导出器和 Cloud Trace 导出器添加到应用的代码中。OpenTelemetry 插桩直接在应用的 Redis 客户端库内运行，用于捕获指标。这样，您的应用就可以记录延迟数据点，并将其导出到 Monitoring 和 Trace 以进行可视化。

如需启用客户端指标，您可以使用 Go、 Java、Node.js、 或 Python。如需了解如何为每种语言启用指标，请参阅以下标签页。

Go

  1. 如需安装所需的 OpenTelemetry 和 Google Cloud 导出器 依赖项，请在终端中运行以下命令：

      go get github.com/gomodule/redigo/redis@latest
  go get go.opentelemetry.io/otel
  go get go.opentelemetry.io/otel/sdk/trace
  go get go.opentelemetry.io/otel/sdk/metric
  go get github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/trace
  go get github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/metric

  2. 如需启用客户端指标，请创建 main.go 文件，并向其中添加以下代码：

    package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"time"

	"github.com/gomodule/redigo/redis"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/codes"
	"go.opentelemetry.io/otel/metric"
	"go.opentelemetry.io/otel/trace"

	gcpmetric "github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/metric"
	gcptrace "github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/trace"
	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
	sdktrace "go.opentelemetry.io/otel/sdk/trace"
)

// MetricClient encapsulates the tracer and metric histograms to avoid package-level globals.
type MetricClient struct {
	tracer           trace.Tracer
	rttHist          metric.Float64Histogram
	clientBlockHist  metric.Float64Histogram
	appBlockHist     metric.Float64Histogram
	retryCounter     metric.Int64Counter
	connErrorCounter metric.Int64Counter
}

// sleep hook enables lightning-fast unit tests by stubbing out real time.Sleep
var sleep = time.Sleep

// sinceMs calculates elapsed time in fractional milliseconds to avoid truncating sub-millisecond durations.
func sinceMs(start time.Time) float64 {
	return float64(time.Since(start).Microseconds()) / 1000.0
}

func initTelemetry(ctx context.Context) (*MetricClient, func(), error) {
	traceExporter, err := gcptrace.New()
	if err != nil {
		return nil, nil, fmt.Errorf("gcptrace.New: %w", err)
	}
	tp := sdktrace.NewTracerProvider(sdktrace.WithBatcher(traceExporter))
	otel.SetTracerProvider(tp)
	tracer := tp.Tracer("redigo.client")

	metricExporter, err := gcpmetric.New()
	if err != nil {
		return nil, nil, fmt.Errorf("gcpmetric.New: %w", err)
	}
	mp := sdkmetric.NewMeterProvider(sdkmetric.WithReader(sdkmetric.NewPeriodicReader(metricExporter, sdkmetric.WithInterval(10*time.Second))))
	otel.SetMeterProvider(mp)
	meter := mp.Meter("redigo.metrics")

	rttHist, err := meter.Float64Histogram("redis_client_rtt", metric.WithUnit("ms"))
	if err != nil {
		return nil, nil, fmt.Errorf("redis_client_rtt histogram: %w", err)
	}
	clientBlockHist, err := meter.Float64Histogram("redis_client_blocking_latency", metric.WithUnit("ms"))
	if err != nil {
		return nil, nil, fmt.Errorf("redis_client_blocking_latency histogram: %w", err)
	}
	appBlockHist, err := meter.Float64Histogram("redis_application_blocking_latency", metric.WithUnit("ms"))
	if err != nil {
		return nil, nil, fmt.Errorf("redis_application_blocking_latency histogram: %w", err)
	}
	retryCounter, err := meter.Int64Counter("redis_retry_count")
	if err != nil {
		return nil, nil, fmt.Errorf("redis_retry_count counter: %w", err)
	}
	connErrorCounter, err := meter.Int64Counter("redis_connectivity_error_count")
	if err != nil {
		return nil, nil, fmt.Errorf("redis_connectivity_error_count counter: %w", err)
	}

	client := &MetricClient{
		tracer:           tracer,
		rttHist:          rttHist,
		clientBlockHist:  clientBlockHist,
		appBlockHist:     appBlockHist,
		retryCounter:     retryCounter,
		connErrorCounter: connErrorCounter,
	}

	initAttrs := metric.WithAttributes(attribute.String("operation", "startup"))
	client.retryCounter.Add(ctx, 0, initAttrs)
	client.connErrorCounter.Add(ctx, 0, initAttrs)

	shutdown := func() {
		tp.Shutdown(ctx)
		mp.Shutdown(ctx)
	}

	return client, shutdown, nil
}

func (c *MetricClient) smartRedisCall(ctx context.Context, pool *redis.Pool, operationName string, commandName string, args ...interface{}) (interface{}, error) {
	// Create a dedicated child span for the Redis command
	ctx, span := c.tracer.Start(ctx, operationName)
	span.SetAttributes(attribute.String("redis.command", commandName))
	defer span.End()

	maxRetries := 3
	attempt := 0
	metricOpts := metric.WithAttributes(attribute.String("operation", operationName))
	var lastErr error

	for attempt < maxRetries {
		poolStart := time.Now()
		// Use GetContext to respect context deadlines and cancellation
		conn, err := pool.GetContext(ctx)
		c.clientBlockHist.Record(ctx, sinceMs(poolStart), metricOpts)

		if err != nil {
			c.connErrorCounter.Add(ctx, 1, metricOpts)
			c.retryCounter.Add(ctx, 1, metricOpts)
			span.RecordError(err)
			span.SetStatus(codes.Error, err.Error())
			lastErr = err
			attempt++
			if attempt >= maxRetries {
				break
			}
			sleep(time.Duration(100<<attempt) * time.Millisecond)
			continue
		}

		// Check if the connection is dead
		if err := conn.Err(); err != nil {
			conn.Close()
			c.connErrorCounter.Add(ctx, 1, metricOpts)
			c.retryCounter.Add(ctx, 1, metricOpts)
			span.RecordError(err)
			span.SetStatus(codes.Error, err.Error())
			lastErr = err
			attempt++
			if attempt >= maxRetries {
				break
			}
			sleep(time.Duration(100<<attempt) * time.Millisecond)
			continue
		}

		reqStart := time.Now()
		// Redigo has no native DoContext; pass timeouts using redis.DoWithTimeout when context has a deadline
		var reply interface{}
		if deadline, ok := ctx.Deadline(); ok {
			reply, err = redis.DoWithTimeout(conn, time.Until(deadline), commandName, args...)
		} else {
			reply, err = conn.Do(commandName, args...)
		}
		c.rttHist.Record(ctx, sinceMs(reqStart), metricOpts)
		conn.Close()

		if err != nil {
			c.retryCounter.Add(ctx, 1, metricOpts)
			span.RecordError(err)
			span.SetStatus(codes.Error, err.Error())
			lastErr = err
			attempt++
			if attempt >= maxRetries {
				break
			}
			sleep(time.Duration(100<<attempt) * time.Millisecond)
			continue
		}

		appStart := time.Now()
		// Replace fmt.Sprintf to remove unnecessary string formatting overhead
		sleep(2 * time.Millisecond)
		c.appBlockHist.Record(ctx, sinceMs(appStart), metricOpts)

		// Reset span status to Ok if the retry or execution eventually succeeds
		span.SetStatus(codes.Ok, "")

		return reply, nil
	}
	return nil, fmt.Errorf("max retries reached for %s: %w", operationName, lastErr)
}

func main() {
	ctx := context.Background()
	client, shutdown, err := initTelemetry(ctx)
	if err != nil {
		log.Printf("Failed to initialize telemetry: %v", err)
		os.Exit(1)
	}
	defer shutdown()

	redisHost := os.Getenv("REDISHOST")
	redisPort := os.Getenv("REDISPORT")
	if redisPort == "" {
		redisPort = "6379"
	}

	pool := &redis.Pool{
		MaxIdle:     10,
		MaxActive:   20,
		IdleTimeout: 240 * time.Second,
		Wait:        true,
		Dial: func() (redis.Conn, error) {
			return redis.Dial("tcp", fmt.Sprintf("%s:%s", redisHost, redisPort))
		},
	}
	defer pool.Close()

	ctx, span := client.tracer.Start(ctx, "fetch_data_span")
	defer span.End()

	// Simple write and read operations
	_, err = client.smartRedisCall(ctx, pool, "set_user", "SET", "user:123", "active")
	if err != nil {
		log.Printf("Error setting data: %v", err)
	}
	val, err := client.smartRedisCall(ctx, pool, "get_user", "GET", "user:123")
	if err != nil {
		log.Printf("Error fetching data: %v", err)
	} else {
		log.Printf("Retrieved value: %s", val)
	}
}

  3. 运行应用至少 1 分钟，以便导出器有足够的时间将已发布的指标批量发送到 Monitoring。

Java

  1. 如需安装所需的 OpenTelemetry 和 Google Cloud 导出器 依赖项，请将以下代码添加到应用的 pom.xml 文件中：

    <dependencies>
    <dependency>
        <groupId>redis.clients</groupId>
        <artifactId>jedis</artifactId>
        <version>5.1.0</version>
    </dependency>
    <dependency>
        <groupId>io.opentelemetry</groupId>
        <artifactId>opentelemetry-api</artifactId>
        <version>1.36.0</version>
    </dependency>
    <dependency>
        <groupId>io.opentelemetry</groupId>
        <artifactId>opentelemetry-sdk</artifactId>
        <version>1.36.0</version>
    </dependency>
    <dependency>
        <groupId>com.google.cloud.opentelemetry</groupId>
        <artifactId>exporter-trace</artifactId>
        <version>0.28.0</version>
    </dependency>
    <dependency>
        <groupId>com.google.cloud.opentelemetry</groupId>
        <artifactId>exporter-metrics</artifactId>
        <version>0.28.0</version>
    </dependency>

    <!-- Testing Dependencies -->
    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.13.2</version>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.mockito</groupId>
        <artifactId>mockito-core</artifactId>
        <version>4.11.0</version>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-simple</artifactId>
        <version>1.7.36</version>
        <scope>test</scope>
    </dependency>
</dependencies>

  2. 如需启用客户端指标，请创建 RedisTelemetryApp.java 文件，并向其中添加以下代码：

    import com.google.cloud.opentelemetry.metric.GoogleCloudMetricExporter;
import com.google.cloud.opentelemetry.trace.TraceExporter;
import io.opentelemetry.api.OpenTelemetry;
import io.opentelemetry.api.common.AttributeKey;
import io.opentelemetry.api.common.Attributes;
import io.opentelemetry.api.metrics.DoubleHistogram;
import io.opentelemetry.api.metrics.LongCounter;
import io.opentelemetry.api.metrics.Meter;
import io.opentelemetry.api.trace.Span;
import io.opentelemetry.api.trace.Tracer;
import io.opentelemetry.sdk.OpenTelemetrySdk;
import io.opentelemetry.sdk.metrics.export.MetricExporter;
import io.opentelemetry.sdk.metrics.SdkMeterProvider;
import io.opentelemetry.sdk.metrics.export.PeriodicMetricReader;
import io.opentelemetry.sdk.trace.SdkTracerProvider;
import io.opentelemetry.sdk.trace.export.BatchSpanProcessor;
import io.opentelemetry.sdk.trace.export.SpanExporter;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.exceptions.JedisConnectionException;

import java.time.Duration;
import java.util.function.Function;

/**
 * Sample application demonstrating client-side metrics and tracing for
 * Google Cloud Memorystore for Redis.
 */
public final class RedisTelemetryApp {
    /** Attribute key for Redis operation names. */
    private static final AttributeKey<String> ATTR_OPERATION =
            AttributeKey.stringKey("operation");

    /** Maximum number of Redis reconnection attempts. */
    private static final int MAX_RETRIES = 3;

    /** Maximum total connections for the Jedis pool. */
    private static final int POOL_MAX_TOTAL = 20;

    /** Interval in seconds for exporting metrics to Google Cloud. */
    private static final long METRIC_INTERVAL_SECONDS = 10L;

    /** Base multiplier for exponential backoff sleep (in milliseconds). */
    private static final long RETRY_BACKOFF_BASE_MS = 100L;

    /** Conversion factor from Nanoseconds to Milliseconds. */
    private static final double NANO_TO_MS = 1_000_000.0;

    /** Default Redis port. */
    private static final int DEFAULT_REDIS_PORT = 6379;

    /** OpenTelemetry Tracer instance for recording trace spans. */
    private static Tracer tracer;

    /** OpenTelemetry Histogram for Redis round-trip time. */
    private static DoubleHistogram rttHist;

    /** OpenTelemetry Histogram for pool blocking latency. */
    private static DoubleHistogram clientBlockHist;

    /** OpenTelemetry Histogram for application logic blocking latency. */
    private static DoubleHistogram appBlockHist;

    /** OpenTelemetry Counter for Redis reconnection retry events. */
    private static LongCounter retryCounter;

    /** OpenTelemetry Counter for Redis connectivity errors. */
    private static LongCounter connErrorCounter;

    /** Shared Jedis connection pool. */
    private static JedisPool jedisPool;

    /**
     * Private constructor to prevent instantiation of this utility class.
     */
    private RedisTelemetryApp() {
    }

    /**
     * Main entry point for running the sample application.
     *
     * @param args Command line arguments (not used).
     */
    public static void main(final String[] args) {
        setupTelemetry();

        final String host = System.getenv()
                .getOrDefault("REDISHOST", "localhost");
        final int port = Integer.parseInt(System.getenv()
                .getOrDefault("REDISPORT",
                        String.valueOf(DEFAULT_REDIS_PORT)));

        final JedisPoolConfig poolConfig = new JedisPoolConfig();
        poolConfig.setMaxTotal(POOL_MAX_TOTAL);
        poolConfig.setBlockWhenExhausted(true);
        jedisPool = new JedisPool(poolConfig, host, port);

        try {
            run();
        } finally {
            if (jedisPool != null) {
                jedisPool.close();
            }
        }
    }

    /**
     * Executes the core business logic of reading and writing to Redis.
     *
     * @return The string retrieved from the Redis 'get' operation.
     */
    static String run() {
        final Span span = tracer.spanBuilder("process_user_span")
                .startSpan();
        try {
            smartRedisCall("set_user", jedis ->
                    jedis.set("user:123", "active"));

            final String result = smartRedisCall("get_user", jedis ->
                    jedis.get("user:123"));
            System.out.println("Retrieved: " + result);
            return result;
        } catch (Exception e) {
            span.recordException(e);
            throw e;
        } finally {
            span.end();
        }
    }

    /**
     * Injects mocked or no-op telemetry and pool instances for unit testing.
     *
     * @param pool                The mocked or test JedisPool instance.
     * @param testOpenTelemetry The OpenTelemetry instance to use for testing.
     */
    static void initForTest(
            final JedisPool pool,
            final OpenTelemetry testOpenTelemetry) {
        jedisPool = pool;
        tracer = testOpenTelemetry.getTracer("jedis.client");
        final Meter meter = testOpenTelemetry.getMeter("jedis.metrics");

        rttHist = meter.histogramBuilder("redis_client_rtt")
                .setUnit("ms").build();
        clientBlockHist = meter
                .histogramBuilder("redis_client_blocking_latency")
                .setUnit("ms").build();
        appBlockHist = meter
                .histogramBuilder("redis_application_blocking_latency")
                .setUnit("ms").build();
        retryCounter = meter.counterBuilder("redis_retry_count").build();
        connErrorCounter = meter
                .counterBuilder("redis_connectivity_error_count")
                .build();

        retryCounter.add(0, Attributes.of(ATTR_OPERATION, "startup"));
        connErrorCounter.add(0, Attributes.of(ATTR_OPERATION, "startup"));
    }

    /**
     * Configures the production OpenTelemetry SDK to export Traces and Metrics
     * to Google Cloud Operations.
     */
    private static void setupTelemetry() {
        final SpanExporter traceExporter =
                TraceExporter.createWithDefaultConfiguration();
        final SdkTracerProvider tracerProvider =
                SdkTracerProvider.builder()
                        .addSpanProcessor(
                                BatchSpanProcessor.builder(traceExporter)
                                        .build())
                        .build();

        final MetricExporter metricExporter =
                GoogleCloudMetricExporter.createWithDefaultConfiguration();
        final SdkMeterProvider meterProvider =
                SdkMeterProvider.builder()
                        .registerMetricReader(
                                PeriodicMetricReader.builder(metricExporter)
                                        .setInterval(Duration.ofSeconds(
                                                METRIC_INTERVAL_SECONDS))
                                        .build())
                        .build();

        final OpenTelemetry openTelemetry = OpenTelemetrySdk.builder()
                .setTracerProvider(tracerProvider)
                .setMeterProvider(meterProvider)
                .buildAndRegisterGlobal();

        tracer = openTelemetry.getTracer("jedis.client");
        final Meter meter = openTelemetry.getMeter("jedis.metrics");

        rttHist = meter.histogramBuilder("redis_client_rtt")
                .setUnit("ms").build();
        clientBlockHist = meter
                .histogramBuilder("redis_client_blocking_latency")
                .setUnit("ms").build();
        appBlockHist = meter
                .histogramBuilder("redis_application_blocking_latency")
                .setUnit("ms").build();
        retryCounter = meter.counterBuilder("redis_retry_count").build();
        connErrorCounter = meter
                .counterBuilder("redis_connectivity_error_count")
                .build();

        retryCounter.add(0, Attributes.of(ATTR_OPERATION, "startup"));
        connErrorCounter.add(0, Attributes.of(ATTR_OPERATION, "startup"));
    }

    /**
     * Wraps a Redis operation with latency metrics, reconnection retry logic,
     * and trace spans.
     *
     * @param <T>           The return type of the Redis operation.
     * @param operationName The name of the operation for metric attributes.
     * @param operation     The Redis command lambda to execute safely.
     * @return The return value from the Redis command.
     */
    private static <T> T smartRedisCall(
            final String operationName,
            final Function<Jedis, T> operation) {
        int attempt = 0;
        final Attributes attrs = Attributes.of(ATTR_OPERATION,
                operationName);

        final Span span = tracer.spanBuilder(operationName).startSpan();

        try {
            while (attempt < MAX_RETRIES) {
                final long poolStart = System.nanoTime();
                try (Jedis jedis = jedisPool.getResource()) {
                    clientBlockHist.record((System.nanoTime() - poolStart)
                            / NANO_TO_MS, attrs);

                    final long reqStart = System.nanoTime();
                    final T response = operation.apply(jedis);
                    rttHist.record((System.nanoTime() - reqStart)
                            / NANO_TO_MS, attrs);

                    final long appStart = System.nanoTime();
                    @SuppressWarnings("unused")
                    final String dummy = String.valueOf(response);
                    appBlockHist.record((System.nanoTime() - appStart)
                            / NANO_TO_MS, attrs);

                    return response;
                } catch (JedisConnectionException e) {
                    attempt++;
                    connErrorCounter.add(1, attrs);
                    retryCounter.add(1, attrs);
                    span.recordException(e);
                    if (attempt >= MAX_RETRIES) {
                        throw e;
                    }
                    try {
                        Thread.sleep((long) (Math.pow(2, attempt)
                                * RETRY_BACKOFF_BASE_MS));
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                    }
                }
            }
            return null;
        } finally {
            span.end();
        }
    }
}

  3. 运行应用至少 1 分钟，以便导出器有足够的时间将已发布的指标批量发送到 Monitoring。

Node.js

  1. 如需安装所需的 OpenTelemetry 和 Google Cloud 导出器 依赖项，请在终端中运行以下命令：

      npm install redis@^4.6.0 @opentelemetry/api@^1.9.0
  @opentelemetry/sdk-trace-node@^2.1.0
  @opentelemetry/sdk-trace-base@^2.1.0
  @opentelemetry/sdk-metrics@^2.1.0
  @opentelemetry/instrumentation@^0.205.0
  @opentelemetry/instrumentation-redis@^0.67.0
  @google-cloud/opentelemetry-cloud-trace-exporter@^3.0.0
  @google-cloud/opentelemetry-cloud-monitoring-exporter@^0.21.0
  @opentelemetry/resources@^2.1.0

  2. 如需启用客户端指标，请创建 server.js 文件，并向其中添加以下代码：

    
'use strict';

const {trace, metrics} = require('@opentelemetry/api');
const {NodeTracerProvider} = require('@opentelemetry/sdk-trace-node');
const {BatchSpanProcessor} = require('@opentelemetry/sdk-trace-base');
const {
  TraceExporter,
} = require('@google-cloud/opentelemetry-cloud-trace-exporter');
const {
  MeterProvider,
  PeriodicExportingMetricReader,
} = require('@opentelemetry/sdk-metrics');
const {
  MetricExporter,
} = require('@google-cloud/opentelemetry-cloud-monitoring-exporter');
const {RedisInstrumentation} = require('@opentelemetry/instrumentation-redis');
const {registerInstrumentations} = require('@opentelemetry/instrumentation');
const {performance} = require('perf_hooks');

// FIX: Pass spanProcessors in the constructor options for NodeTracerProvider in SDK 2.x
const provider = new NodeTracerProvider({
  spanProcessors: [new BatchSpanProcessor(new TraceExporter())],
});
provider.register();

registerInstrumentations({
  instrumentations: [new RedisInstrumentation()],
});

const redis = require('redis');

const metricExporter = new MetricExporter();
const metricReader = new PeriodicExportingMetricReader({
  exporter: metricExporter,
  exportIntervalMillis: 10000,
});
const meterProvider = new MeterProvider({readers: [metricReader]});
metrics.setGlobalMeterProvider(meterProvider);

const tracer = trace.getTracer('redis.client.node');
const meter = metrics.getMeter('redis.metrics.node');

const rttHist = meter.createHistogram('redis_client_rtt', {unit: 'ms'});
const appBlockHist = meter.createHistogram(
  'redis_application_blocking_latency',
  {unit: 'ms'}
);
const retryCounter = meter.createCounter('redis_retry_count');
const connErrorCounter = meter.createCounter('redis_connectivity_error_count');

retryCounter.add(0, {operation: 'startup'});
connErrorCounter.add(0, {operation: 'startup'});

const REDISHOST = process.env.REDISHOST || 'localhost';
const REDISPORT = process.env.REDISPORT || 6379;

const client = redis.createClient({
  socket: {
    host: REDISHOST,
    port: REDISPORT,
    reconnectStrategy: retries => {
      connErrorCounter.add(1, {error: 'socket_reconnect'});
      if (retries > 5) return new Error('Max retries reached');
      return Math.min(retries * 100, 3000);
    },
  },
});
client.on('error', err => console.log('Redis Client Error', err));

async function smartRedisCall(operationName, func, ...args) {
  let attempt = 0;
  while (attempt < 3) {
    try {
      const reqStart = performance.now();
      const response = await func(...args);
      rttHist.record(performance.now() - reqStart, {operation: operationName});

      const appParseStart = performance.now();
      // eslint-disable-next-line no-unused-vars
      const _ = String(response);
      appBlockHist.record(performance.now() - appParseStart, {
        operation: operationName,
      });

      return response;
    } catch (e) {
      attempt++;
      retryCounter.add(1, {operation: operationName});
      if (attempt >= 3) throw e;
      await new Promise(resolve =>
        setTimeout(resolve, Math.pow(2, attempt) * 100)
      );
    }
  }
}

async function main() {
  await client.connect();

  await tracer.startActiveSpan('process_user_span', async span => {
    try {
      // Simple write and read operations
      await smartRedisCall(
        'set_user',
        client.set.bind(client),
        'user:123',
        'active'
      );

      const result = await smartRedisCall(
        'get_user',
        client.get.bind(client),
        'user:123'
      );
      console.log('Retrieved:', result);
    } catch (e) {
      span.recordException(e);
    } finally {
      span.end();
    }
  });

  await client.quit();
  await provider.forceFlush();
  await meterProvider.forceFlush();
}

// Only run the script automatically if it is executed directly (e.g. `node server.js`)
if (require.main === module) {
  main().catch(console.error);
}

// Export for testability
module.exports = {
  main,
  smartRedisCall,
};

  3. 运行应用至少 1 分钟，以便导出器有足够的时间将已发布的指标批量发送到 Monitoring。

Python

  1. 如需安装所需的 OpenTelemetry 和 Google Cloud 导出器 依赖项，请在终端中运行以下命令：

      pip install redis==7.0.1 opentelemetry-api==1.39.1
  opentelemetry-sdk==1.39.1
  opentelemetry-instrumentation-redis==0.60b1
  opentelemetry-exporter-gcp-trace==1.11.0
  opentelemetry-exporter-gcp-monitoring==1.11.0a0

  2. 如需启用客户端指标，请创建 main.py 文件，并向其中添加以下代码：

    import os
import time

from opentelemetry import metrics, trace
from opentelemetry.exporter.cloud_monitoring import (
    CloudMonitoringMetricsExporter,
)
from opentelemetry.exporter.cloud_trace import CloudTraceSpanExporter
from opentelemetry.instrumentation.redis import RedisInstrumentor
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
import redis
from redis.exceptions import ConnectionError, TimeoutError




def init_telemetry():
    """Initializes OpenTelemetry with GCP Exporters and returns the SDK objects."""
    # 1. Initialize Tracing
    tracer_provider = TracerProvider()
    tracer_provider.add_span_processor(
        BatchSpanProcessor(CloudTraceSpanExporter())
    )
    trace.set_tracer_provider(tracer_provider)
    tracer = trace.get_tracer("redis.client")

    # 2. Initialize Metrics
    metrics_exporter = CloudMonitoringMetricsExporter()
    metric_reader = PeriodicExportingMetricReader(
        metrics_exporter, export_interval_millis=10000
    )
    meter_provider = MeterProvider(metric_readers=[metric_reader])
    metrics.set_meter_provider(meter_provider)
    meter = metrics.get_meter("redis.metrics")

    # Bundle all metric handlers safely into a dictionary
    redis_metrics = {
        "rtt_hist": meter.create_histogram("redis_client_rtt", unit="ms"),
        "client_block_hist": meter.create_histogram(
            "redis_client_blocking_latency", unit="ms"
        ),
        "app_block_hist": meter.create_histogram(
            "redis_application_blocking_latency", unit="ms"
        ),
        "retry_counter": meter.create_counter("redis_retry_count"),
        "conn_error_counter": meter.create_counter(
            "redis_connectivity_error_count"
        ),
    }

    redis_metrics["retry_counter"].add(0, {"operation": "startup"})
    redis_metrics["conn_error_counter"].add(0, {"operation": "startup"})

    # 3. Setup Redis Auto-Instrumentation
    RedisInstrumentor().instrument()

    return tracer, redis_metrics, tracer_provider, meter_provider


def init_redis_pool():
    """Initializes and returns the Redis ConnectionPool and Client."""
    redis_host = os.environ.get("REDISHOST", "localhost")
    redis_port = int(os.environ.get("REDISPORT", 6379))

    redis_pool = redis.ConnectionPool(
        host=redis_host,
        port=redis_port,
        max_connections=10,
        decode_responses=True,
    )
    redis_client = redis.Redis(connection_pool=redis_pool)
    return redis_pool, redis_client


def smart_redis_call(
    operation_name, func, redis_pool, metrics, *args, **kwargs
):
    """Executes a Redis operation with metrics and retry handling (No Globals!)."""
    max_retries = 3
    attempt = 0

    pool_start = time.time()
    try:
        conn = redis_pool.get_connection()
        redis_pool.release(conn)
    except Exception:
        pass

    if metrics and metrics.get("client_block_hist"):
        metrics["client_block_hist"].record(
            (time.time() - pool_start) * 1000, {"operation": operation_name}
        )

    while attempt < max_retries:
        try:
            req_start = time.time()
            response = func(*args, **kwargs)

            if metrics and metrics.get("rtt_hist"):
                metrics["rtt_hist"].record(
                    (time.time() - req_start) * 1000,
                    {"operation": operation_name},
                )

            app_start = time.time()
            _ = str(response)

            if metrics and metrics.get("app_block_hist"):
                metrics["app_block_hist"].record(
                    (time.time() - app_start) * 1000,
                    {"operation": operation_name},
                )

            return response

        except (ConnectionError, TimeoutError) as e:
            attempt += 1
            if metrics and metrics.get("conn_error_counter"):
                metrics["conn_error_counter"].add(
                    1, {"operation": operation_name}
                )
            if metrics and metrics.get("retry_counter"):
                metrics["retry_counter"].add(1, {"operation": operation_name})
            if attempt >= max_retries:
                raise e
            time.sleep((2**attempt) * 0.1)

if __name__ == "__main__":
    tracer, redis_metrics, tracer_provider, meter_provider = init_telemetry()
    redis_pool, redis_client = init_redis_pool()

    if tracer:
        with tracer.start_as_current_span("process_user_span"):
            try:
                # Simple write and read operations
                smart_redis_call(
                    "set_user",
                    redis_client.set,
                    redis_pool,
                    redis_metrics,
                    "user:123",
                    "active",
                )

                result = smart_redis_call(
                    "get_user",
                    redis_client.get,
                    redis_pool,
                    redis_metrics,
                    "user:123",
                )
                print(f"Retrieved: {result}")
            except Exception as e:
                print(f"Error: {e}")

        tracer_provider.force_flush()
        meter_provider.force_flush()

  3. 运行应用至少 1 分钟，以便导出器有足够的时间将已发布的指标批量发送到 Monitoring。

在 Monitoring 中查看指标

启用客户端指标并运行应用至少 1 分钟，以便导出器有足够的时间将指标批量发送到 Monitoring 后，您可以使用 Monitoring 可视化指标，按操作或实例对指标进行分组，并应用聚合器来监控应用的性能。

如需在 Monitoring 中查看指标，请执行以下操作：

  1. 在 Google Cloud 控制台中，前往 Metrics Explorer 页面。

    进入 Metrics Explorer

  2. 选择您的 Google Cloud 项目。

  3. 点击选择指标 。

  4. 搜索 workload.googleapis.com/redis。

  5. 选择客户端指标。根据需要按 operation 和 instance 对数据进行分组，然后选择一个聚合器。如需了解更多选项，请参阅使用 Metrics Explorer 时选择指标。

在 Trace 中查看分布式跟踪记录

应用开始导出数据后，您可以使用 Trace 可视化 Redis 命令的完整请求-响应周期。通过在 Trace 中查看分布式跟踪记录，您可以诊断瓶颈，以便快速隔离应用中延迟的确切来源。

如需在 Trace 中查看分布式跟踪记录，请执行以下操作：

  1. 在 Google Cloud 控制台中，前往 Trace Explorer 页面。

    前往 Trace Explorer

  2. 在散点图上选择以点表示的最近跟踪记录。

  3. 检查瀑布图，通过识别以下瓶颈来隔离延迟来源：

    • 总请求时长：顶层（父级）条形图显示了您必须等待操作完成的总 时间。

    • 网络和服务器延迟 (RTT)：子条形图（例如标有 GET 或 SET 的条形图）显示了命令在网络上传输和在 Memorystore for Redis 服务器上运行所花费的时间。

    • 客户端连接阻塞：如果 Redis 子 span 开始之前存在较大的空白水平间隙，则表示应用线程卡在等待连接池中的可用 TCP 连接。

    • 应用解析阻塞：如果 Redis 子 span 结束后存在较大的空白水平间隙，则表示应用难以解析 或难以处理返回的载荷。这种情况通常发生在多兆字节 JSON 字符串中。

    • 重试：如果您看到同一父级跟踪记录中按顺序出现同一命令的多个短子 span，则表示您的客户端可能会遇到网络丟包，并且必须触发其指数退避重试循环。

问题排查

本部分列出了您可以使用客户端指标识别的常见性能问题，解释了这些问题的根本原因，并提供了有关问题排查的指南。

问题 原因 问题排查

您的应用遇到突发延迟峰值，但 Memorystore for Redis 看起来完全正常。

  • workload.googleapis.com/
    redis_client_blocking_latency     （客户端指标）：峰值
  • workload.googleapis.com/redis_client_rtt （客户端指标）：低 / 典型
  • redis.googleapis.com/commands/
    usec_per_call     （Memorystore for Redis 服务器指标）：低 / 典型
  • redis.googleapis.com/clients/connected （Memorystore for Redis 服务器指标）：在特定数字处趋于平稳
瓶颈严格位于您的应用内。您的 线程尝试运行 Redis 命令，但连接池已完全 耗尽。较高的 redis_client_blocking_latency 表示您的代码在将命令发送到网络之前等待可用 TCP 套接字所花费的时间。 如需处理更高的并发流量，请在 Redis 客户端配置中增加连接池大小限制（例如，MaxActive for Go, MaxTotal for Java, or max_connections for Node.js and Python）。

请求完成，但端点所花费的时间远超预期 。您的网络或服务器的运行状况没有相关问题。

  • workload.googleapis.com/
    redis_application_blocking_latency     （客户端指标）：峰值
  • workload.googleapis.com/redis_client_rtt（客户端 指标）：低 / 典型
  • redis.googleapis.com/commands/
    usec_per_call     （Memorystore for Redis 服务器指标）：低 / 典型
  • redis.googleapis.com/stats/
    network_traffic     （传出字节数）（Memorystore for Redis 服务器指标）：大幅峰值
 Memorystore for Redis 运行命令，网络快速传输载荷（低 RTT）。但是，返回的载荷很大（例如，15 MB 的 JSON 字符串）。您的应用遇到较高的 redis_application_blocking_latency，因为应用 在分配内存并将该 大型字符串反序列化为对象时消耗了过多的资源。 优化您的数据模型。请勿在单个键中存储大量 JSON blob。 使用 Redis 哈希 (HSET) 分解数据，并使用 HGET 或 HMGET 仅检索所需的特定字段 。

面向用户的应用延迟出现峰值，但 Redis 指标 报告的服务器延迟较低，并且连接池签出正常。

  • workload.googleapis.com/redis_retry_count （客户端指标）：峰值
  • workload.googleapis.com/
    redis_connectivity_error_count     （客户端指标）：可能会显示瞬时增量
  • workload.googleapis.com/redis_client_rtt （客户端指标）：成功请求的低 / 典型值
  • redis.googleapis.com/commands/
    usec_per_call     （Memorystore for Redis 服务器指标）：低 / 典型
 由于 redis_client_rtt 仅捕获 成功请求的 RTT，因此它不会反映失败 数据包的超时时长。当您的应用遇到瞬时数据包丢弃或 TCP 重置时，插桩客户端的重试逻辑会增加 redis_retry_count 并触发其指数退避算法循环。这会在尝试之间引入休眠时间（例如， 100ms、200ms 或 400ms）。用户 会遇到较高的总延迟，但根本原因是网络 丟包，这会触发客户端休眠延迟。 检查 VPC 流日志中是否存在丢弃的数据包、带宽限制、 或跨区域路由异常。如果您遇到激进的超时， 请确保您的客户端连接超时（socket_timeout或 connect_timeout）大于预期 RTT，以应对瞬时网络抖动。

一切停止，遥测流水线的所有层级都报告 高延迟。

  • workload.googleapis.com/redis_client_rtt（客户端 指标）：高
  • redis.googleapis.com/commands/
    usec_per_call     （Memorystore for Redis 服务器指标）：高
  • redis.googleapis.com/stats/
    cpu_utilization_main_thread     （Memorystore for Redis 服务器指标）：高（例如，接近 1 秒/秒， 或 100%）
  • Trace 瀑布图：显示命令花费大量时间
Redis 是单线程的。当您运行 O(N) 时间复杂度的命令（例如，KEYS *、对大型集执行 SMEMBERS 或对包含数百万个 字段的哈希执行 HGETALL）时，Redis 引擎会暂停以满足该请求。在该命令 运行时，所有其他应用请求都会排队，从而导致系统范围内的延迟 峰值。由于您的自定义 redis_client_rtt 与服务器的延迟 (commands/usec_per_call) 匹配，因此运行该命令的服务器是瓶颈。

打开 Trace，查看慢速 span 上的 Redis 命令，以确定哪个查询导致阻塞。将代码中的阻塞命令 替换为非阻塞命令。

如需以增量方式遍历大型数据集而不锁定 服务器线程，请使用 SCAN、SSCAN 或 HSCAN。

您的应用报告所有 Redis 命令的基准延迟始终较高，即使流量较低也是如此。

  • workload.googleapis.com/redis_client_rtt （客户端指标）：始终较高（p50 和 p99 均为 ~30-100 毫秒以上）
  • redis.googleapis.com/commands/
    usec_per_call     （Memorystore for Redis 服务器指标）：极低（< 1 毫秒）
  • workload.googleapis.com/
    redis_client_blocking_latency
    和 redis_application_blocking_latency：低 / 典型
 Redis 服务器会立即运行命令，但您的应用和您的 实例部署在不同的区域（例如， us-central1 和 us-east1）。每个网络数据包 都必须在这些地理数据中心之间通过物理 Google Cloud 基础架构传输。这会导致每次往返都必须承担光速 跨区域延迟惩罚。 如需缩短延迟，请将应用部署到与实例相同的区域 和可用区。如需查看应用和 实例的区域，请使用 Google Cloud 控制台。

后续步骤