Google は AI 技術を使用して、コンテンツをご希望の言語に翻訳しています。AI 翻訳には誤りが含まれる場合があります。

クライアントサイドの指標を使用してレイテンシの高い問題をトラブルシューティングする

Memorystore for Redis は、スループット、CPU 使用率、メモリ使用量をモニタリングするためのリアルタイムのサーバーサイド指標を提供しますが、このデータだけでは、複雑な分散システム内でクライアントアプリケーションのレイテンシが高くなる理由を説明できない場合があります。

クライアントサイドの指標は、リクエストとレスポンスのサイクル全体を可視化することで、この問題を解決します。アプリケーションがコマンドを開始してから、アプリケーションがレスポンスを処理するまでの時間を測定します。これらのデータポイントをキャプチャすることで、レイテンシがアプリケーションロジック、ネットワークパス、Redis サーバーのいずれに起因するかを正確に判断できます。

始める前に

クライアントアプリケーションがサービスアカウントを使用し、次の Identity and Access Management（IAM）ロールが割り当てられていることを確認します。

roles/cloudtrace.agent（Cloud Trace エージェント）
roles/monitoring.metricWriter（モニタリング指標の書き込み）

ロール付与の詳細については、コンソールを使用して IAM ロールを付与する Google Cloud クイックスタートをご覧ください。

Cloud Monitoring API を有効にする

クライアントサイドの指標を Monitoring にエクスポートするには、アプリケーションで Monitoring API を有効にする必要があります。これらの指標を Monitoring でエクスポートして可視化することで、ボトルネックの根本原因を特定し、レイテンシの発生源を特定できます。

Monitoring API を有効にするには、次の操作を行います。

コンソール Google Cloud で、[API とサービス] ページに移動します。

[API とサービス] に移動
Memorystore for Redis インスタンスを作成したプロジェクトを選択します。
[API とサービスを有効化] をクリックします。
monitoring を検索します。
検索結果で、[Cloud Monitoring API] をクリックします。
[API が有効です] が表示されている場合、API はすでに有効になっています。表示されていない場合は、[有効にする] をクリックします。

Cloud Trace API を有効にする

Trace で分散トレースを表示するには、Trace API を有効にする必要があります。Trace エクスプローラを使用してこれらのトレースを表示し、ボトルネックを診断して、アプリケーションのレイテンシの原因を特定できます。

Trace API を有効にするには、次の操作を行います。

コンソール Google Cloud で、[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 エクスポータの依存関係をインストールするには、ターミナルで次のコマンドを実行します。

  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 に送信するのに十分な時間を確保するため、アプリケーションを 1 分以上実行します。

Java

必要な 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>

クライアントサイドの指標を有効にするには、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 に送信するのに十分な時間を確保するため、アプリケーションを 1 分以上実行します。

Node.js

必要な 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

クライアントサイドの指標を有効にするには、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 に送信するのに十分な時間を確保するため、アプリケーションを 1 分以上実行します。

Python

必要な 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

クライアントサイドの指標を有効にするには、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 に送信するのに十分な時間を確保するため、アプリケーションを 1 分以上実行します。

Monitoring で指標を表示する

クライアントサイドの指標を有効にし、エクスポータが指標をバッチ処理して Monitoring に送信するのに十分な時間を確保するため、アプリケーションを 1 分以上実行したら、Monitoring を使用して指標を可視化し、オペレーションまたはインスタンスごとにグループ化して、アグリゲータを適用してアプリケーションのパフォーマンスをモニタリングします。

Monitoring で指標を表示するには、次の操作を行います。

コンソール Google Cloud で、Metrics Explorer のページに移動します。

Metrics Explorer に移動
プロジェクト Google Cloud を選択します。
[指標を選択] をクリックします。
workload.googleapis.com/redis を検索します。
クライアントサイドの指標を選択します。必要に応じて、データを operation と instance でグループ化し、アグリゲータを選択します。その他のオプションについては、Metrics Explorer を使用して指標を選択するをご覧ください。

重要: レイテンシは均一ではないため、平均値に依存しないでください。redis_client_rtt をグラフ化する場合は、95 パーセンタイル（p95）または 99 パーセンタイル（p99）を表示するように集計を構成します。これにより、最も遅いリクエストで発生するレイテンシが明らかになります。

Trace で分散トレースを表示する

アプリケーションがデータのエクスポートを開始したら、Trace を使用して Redis コマンドのリクエストとレスポンスのサイクル全体を可視化できます。Trace で分散トレースを表示すると、ボトルネックを診断して、アプリケーションのレイテンシの正確な原因を特定できます。

Trace で分散トレースを表示するには、次の操作を行います。

Google Cloud コンソールで、[**Trace エクスプローラ**] ページに移動します。

Trace エクスプローラに移動
散布図の点で表される最近のトレースを選択します。
ウォーターフォールビューを調べて、次のボトルネックを特定することで、レイテンシの原因を特定します。
- リクエストの合計時間: 最上位（親）のバーは、オペレーションが完了するまで待機する必要がある合計時間を示します。
- ネットワークとサーバーのレイテンシ（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` （送信バイト数）（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 秒あたり 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 Cloud インフラストラクチャを通過する必要があります。これにより、往復ごとに、光速のリージョン間レイテンシペナルティが強制的に発生します。	レイテンシを短縮するには、インスタンスと同じリージョンとゾーンにアプリケーションをデプロイします。アプリケーションとインスタンスのリージョンを表示するには、 Google Cloud コンソールを使用します。

次のステップ

クライアントサイドの指標の詳細を確認する。
Memorystore for Redis で使用できるクライアントサイドの指標について学習する。

クライアントサイドの指標を使用してレイテンシの高い問題をトラブルシューティングする コレクションでコンテンツを整理 必要に応じて、コンテンツの保存と分類を行います。

始める前に

Cloud Monitoring API を有効にする

Cloud Trace API を有効にする

クライアントサイドの指標を有効にする

Go

Java

Node.js

Python

Monitoring で指標を表示する

Trace で分散トレースを表示する

トラブルシューティング

次のステップ

クライアントサイドの指標を使用してレイテンシの高い問題をトラブルシューティングする