Google 會運用 AI 技術將內容翻譯成你偏好的語言，但可能會出錯。

使用用戶端指標排解高延遲問題

Memorystore for Redis 提供即時伺服器端指標，可監控總處理量、CPU 使用率和記憶體用量，但光是這些資料可能無法解釋，為何用戶端應用程式在複雜的分散式系統中會發生高延遲問題。

用戶端指標可提供完整要求-回應週期的透明度，解決這個問題。這類指標會測量從應用程式啟動指令到處理回應的時間。擷取這些資料點後，您就能準確判斷延遲時間是源自應用程式邏輯、網路路徑還是 Redis 伺服器。

事前準備

請確認用戶端應用程式使用服務帳戶，並為該帳戶指派下列 Identity and Access Management (IAM) 角色：

roles/cloudtrace.agent (Cloud Trace 代理程式)
roles/monitoring.metricWriter (Monitoring 指標寫入者)

如要進一步瞭解如何授予角色，請參閱「使用 Google Cloud 控制台授予 IAM 角色」快速入門指南。

啟用 Cloud Monitoring API

如要將用戶端指標匯出至 Monitoring，應用程式必須啟用 Monitoring API。在 Monitoring 中匯出及以圖表呈現這些指標，有助於找出瓶頸的根本原因，判斷延遲的來源。

如要啟用 Monitoring API，請按照下列步驟操作：

在 Google Cloud 控制台中，前往「APIs & Services」(API 與服務) 頁面。

前往「API 和服務」頁面
選取您建立 Memorystore for Redis 執行個體的專案。
按一下「啟用 API 和服務」。
搜尋 monitoring。
在搜尋結果中，點選「Cloud Monitoring API」。
如果畫面顯示「API 已啟用」，代表 API 已啟用。否則請按一下「啟用」。

啟用 Cloud Trace API

如要在「追蹤記錄」中查看分散式追蹤記錄，請啟用 Trace API。接著，您可以使用「Trace 探索工具」查看這些追蹤記錄、診斷瓶頸，並找出應用程式中的延遲來源。

如要啟用 Trace API，請按照下列步驟操作：

在 Google Cloud 控制台中，前往「APIs & Services」(API 與服務) 頁面。

前往「API 和服務」頁面
選取您建立 Memorystore for Redis 執行個體的專案。
按一下「啟用 API 和服務」。
搜尋 trace。
在搜尋結果中，按一下「Cloud Trace API」。
如果畫面顯示「API 已啟用」，代表 API 已啟用。否則請按一下「啟用」。

啟用用戶端指標

如要啟用用戶端指標，請將 OpenTelemetry SDK、Cloud Monitoring 匯出工具和 Cloud Trace 匯出工具新增至應用程式的程式碼。OpenTelemetry 檢測會直接在應用程式的 Redis 用戶端程式庫中執行，並擷取指標。應用程式可藉此記錄延遲時間資料點，並匯出至 Monitoring 和 Trace 以供視覺化。

如要啟用用戶端指標，可以使用 Go、Java、Node.js 或 Python。如要瞭解如何為各語言啟用指標，請參閱下方的分頁標籤。

Go

如要安裝必要的 OpenTelemetry 和 Google Cloud exporter 依附元件，請在終端機中執行下列指令：

  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

如要啟用用戶端指標，請建立 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)
	}
}

執行應用程式至少一分鐘，讓匯出工具有足夠的時間批次處理並將發布的指標傳送至 Monitoring。

Java

如要安裝必要的 OpenTelemetry 和 Google Cloud exporter 依附元件，請在應用程式的 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>

如要啟用用戶端指標，請建立 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();
        }
    }
}

執行應用程式至少一分鐘，讓匯出工具有足夠的時間批次處理並將發布的指標傳送至 Monitoring。

Node.js

如要安裝必要的 OpenTelemetry 和 Google Cloud exporter 依附元件，請在終端機中執行下列指令：

  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

如要啟用用戶端指標，請建立 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,
};

執行應用程式至少一分鐘，讓匯出工具有足夠的時間批次處理並將發布的指標傳送至 Monitoring。

Python

如要安裝必要的 OpenTelemetry 和 Google Cloud exporter 依附元件，請在終端機中執行下列指令：

  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

如要啟用用戶端指標，請建立 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()

執行應用程式至少一分鐘，讓匯出工具有足夠的時間批次處理並將發布的指標傳送至 Monitoring。

在 Monitoring 中查看指標

啟用用戶端指標並執行應用程式至少一分鐘，讓匯出工具有足夠時間批次處理指標並傳送至 Monitoring 後，即可使用 Monitoring 視覺化呈現指標、依作業或執行個體分組，以及套用匯總工具來監控應用程式的效能。

如要在 Monitoring 中查看指標，請按照下列步驟操作：

前往 Google Cloud 控制台的「Metrics Explorer」頁面。

前往 Metrics Explorer
選取 Google Cloud 專案。
按一下「Select a metric」(選取指標)。
搜尋 workload.googleapis.com/redis。
選取用戶端指標。視需要依 operation 和 instance 將資料分組，然後選擇匯總器。如要瞭解更多選項，請參閱「在使用 Metrics Explorer 時選取指標」。

重要事項：延遲時間很少會保持一致，因此請勿依賴平均值 (平均數)。繪製 redis_client_rtt 圖表時，請將匯總設定為顯示第 95 個百分位數 (p95) 或第 99 個百分位數 (p99)。這會顯示最慢要求所造成的延遲。

在 Trace 中查看分散式追蹤記錄

應用程式開始匯出資料後，您可以使用 Trace 視覺化呈現 Redis 指令的完整要求-回應週期。在 Trace 中查看分散式追蹤記錄，有助於診斷瓶頸，以便快速找出應用程式中延遲的確切來源。

如要在 Trace 中查看分散式追蹤記錄，請按照下列步驟操作：

前往 Google Cloud 控制台的「Trace Explorer」頁面。

前往「Trace Explorer」
在散布圖上選取以圓點表示的近期追蹤記錄。
檢查瀑布圖，找出下列瓶頸，以找出延遲來源：
- 要求總時間長度：頂層 (父項) 長條會顯示作業完成前必須等待的總時間。
- 網路和伺服器延遲 (RTT)：子長條 (例如標示為 GET 或 SET 的長條) 會顯示指令在網路中傳輸，以及在 Memorystore for Redis 伺服器上執行的時間。
- 用戶端連線遭到封鎖：如果 Redis 子時距開始之前出現大片的空白水平間隙，表示應用程式執行緒卡住，正在等待連線集區中的可用 TCP 連線。
- 應用程式剖析作業遭到封鎖：如果 Redis 子時距結束後出現大範圍的空白水平間隙，表示應用程式難以剖析或處理傳回的酬載。這種情況通常發生在數百萬位元組的 JSON 字串。
- 重試：如果同一個指令有多個短暫的子項範圍，且這些範圍在同一個父項追蹤記錄中依序發生，則用戶端可能發生網路封包遺失的情況，因此必須觸發指數輪詢重試迴圈。

疑難排解

本節列出可使用用戶端指標識別的常見效能問題，說明問題的根本原因，並提供疑難排解指引。

問題	原因	疑難排解
應用程式的延遲時間突然大幅增加，但 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 用戶端設定中增加連線集區大小限制 (例如 Go 的 `MaxActive`、Java 的 `MaxTotal`，或 Node.js 和 Python 的 `max_connections`)。
要求完成，但端點所需時間遠超出預期。網路或伺服器健康狀態沒有問題。 `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` (Bytes out) (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` 只會擷取成功要求的回合時間，因此不會反映封包失敗的逾時時間長度。如果應用程式發生暫時性封包遺失或 TCP 重設，經過檢測的用戶端重試邏輯會遞增 `redis_retry_count`，並觸發指數輪詢迴圈。這會在嘗試之間導入休眠時間 (例如 `100ms`、`200ms` 或 `400ms`)。使用者會感受到總延遲時間較長，但根本原因在於網路封包遺失，這會觸發用戶端休眠延遲。	檢查虛擬私有雲流量記錄檔，確認是否有封包遭捨棄、頻寬受到節流，或跨區域路由異常。如果發生逾時問題，請確保用戶端連線逾時 (`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%) 追蹤瀑布圖：顯示耗費大量時間的指令	Redis 是單一執行緒，當您對大型集合執行 `O(N)` 時間複雜度指令 (例如 `KEYS *`、`SMEMBERS` 或對含有數百萬個欄位的雜湊執行 `HGETALL`) 時，Redis 引擎會暫停以滿足該要求。執行該指令時，所有其他應用程式要求都會排隊，導致系統延遲時間大幅增加。由於自訂 `redis_client_rtt` 與伺服器的延遲時間 (`commands/usec_per_call`) 相符，因此執行指令的伺服器是瓶頸。	開啟 Trace，查看緩慢跨度中的 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 雲端基礎架構傳輸。因此，每次來回都會產生強制性的光速跨區域延遲懲罰。	如要縮短延遲時間，請將應用程式部署至與執行個體相同的區域和可用區。如要查看應用程式和執行個體的區域，請使用 Google Cloud 控制台。

後續步驟

進一步瞭解用戶端指標。
瞭解 Memorystore for Redis 適用的用戶端指標。

使用用戶端指標排解高延遲問題 透過集合功能整理內容 你可以依據偏好儲存及分類內容。

事前準備

啟用 Cloud Monitoring API

啟用 Cloud Trace API

啟用用戶端指標

Go

Java

Node.js

Python

在 Monitoring 中查看指標

在 Trace 中查看分散式追蹤記錄

疑難排解

後續步驟

使用用戶端指標排解高延遲問題