Google은 AI 기술을 사용하여 콘텐츠를 사용자의 기본 언어로 번역합니다. AI 번역에는 오류가 있을 수 있습니다.

클라이언트 측 측정항목을 사용하여 높은 지연 시간 문제 해결

Redis용 Memorystore는 처리량, CPU 사용률, 메모리 사용량을 모니터링하기 위한 실시간 서버 측 측정항목을 제공하지만, 이 데이터만으로는 복잡한 분산 시스템 내에서 클라이언트 애플리케이션의 지연 시간이 긴 이유를 설명하지 못할 수 있습니다.

클라이언트 측 측정항목은 전체 요청-응답 주기에 대한 투명성을 제공하여 이 문제를 해결합니다. 애플리케이션이 명령어를 시작한 시점부터 애플리케이션이 응답을 처리할 때까지 명령어를 측정합니다. 이러한 데이터 포인트를 캡처하면 지연 시간이 애플리케이션 로직, 네트워크 경로 또는 Redis 서버에서 발생하는지 정확하게 확인할 수 있습니다.

시작하기 전에

클라이언트 애플리케이션이 서비스 계정을 사용하고 다음 Identity and Access Management (IAM) 역할이 할당되어 있는지 확인합니다.

roles/cloudtrace.agent (Cloud Trace 에이전트)
roles/monitoring.metricWriter (Monitoring 측정항목 작성자)

역할 부여에 대한 자세한 내용은 콘솔을 사용하여 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를 보려면 Trace API를 사용 설정해야 합니다. 그런 다음 Trace 탐색기를 사용하여 이러한 trace를 보고, 병목 현상을 진단하고, 애플리케이션에서 지연 시간의 소스를 격리할 수 있습니다.

Trace API를 사용 설정하려면 다음을 수행합니다.

Google Cloud 콘솔에서 API 및 서비스 페이지로 이동합니다.

API 및 서비스로 이동
Memorystore for Redis 인스턴스를 만든 프로젝트를 선택합니다.
API 및 서비스 사용 설정 을 클릭합니다.
trace를 검색합니다.
검색 결과에서 Cloud Trace API 를 클릭합니다.
API 사용 설정됨 이 표시되면 API가 이미 사용 설정된 것입니다. 그렇지 않으면 사용 설정 을 클릭합니다.

클라이언트 측 측정항목 사용 설정

클라이언트 측 측정항목을 사용 설정하려면 OpenTelemetry SDK, Cloud Monitoring 내보내기 도구, Cloud Trace 내보내기 도구를 애플리케이션의 코드에 추가합니다. 애플리케이션의 Redis 클라이언트 라이브러리 내에서 직접 실행되는 OpenTelemetry 계측은 측정항목을 캡처합니다. 이렇게 하면 애플리케이션에서 지연 시간 데이터 포인트를 기록하고 시각화를 위해 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 측정항목 탐색기 페이지로 이동합니다.

측정항목 탐색기로 이동
Google Cloud 프로젝트를 선택합니다.
측정항목 선택 을 클릭합니다.
workload.googleapis.com/redis를 검색합니다.
클라이언트 측 측정항목을 선택합니다. 필요에 따라 데이터를 operation 및 instance로 그룹화하고 집계기를 선택합니다. 더 많은 옵션을 알아보려면 측정항목 탐색기 사용 시 측정항목 선택을 참조하세요.

중요: 지연 시간은 거의 균일하지 않으므로 평균에 의존하지 마세요. redis_client_rtt를 그래프로 표시할 때는 집계를 구성하여 95번째 (p95) 또는 99번째 (p99) 백분위수를 표시합니다. 이렇게 하면 가장 느린 요청에서 발생하는 지연 시간이 표시됩니다.

Trace에서 분산 trace 보기

애플리케이션이 데이터 내보내기를 시작한 후 Trace를 사용하여 Redis 명령어의 전체 요청-응답 주기를 시각화할 수 있습니다. Trace에서 분산 trace를 보면 병목 현상을 진단하여 애플리케이션에서 지연 시간의 정확한 소스를 빠르게 격리할 수 있습니다.

Trace에서 분산 trace를 보려면 다음 안내를 따르세요.

Google Cloud 콘솔에서 Trace 탐색기 페이지로 이동합니다.

trace 탐색기로 이동
분산형 차트에서 점으로 표시된 최근 trace를 선택합니다.
다음 병목 현상을 식별하여 지연 시간의 소스를 격리하려면 폭포식 뷰를 검사합니다.
- 총 요청 기간: 최상위 (상위) 막대는 작업이 완료될 때까지 기다려야 하는 총 시간을 보여줍니다.
- 네트워크 및 서버 지연 시간 (RTT): 하위 막대 (예: GET 또는 SET 라벨이 지정된 막대)는 명령어가 네트워크를 통해 이동하고 Memorystore for Redis 서버에서 실행되는 데 걸린 시간을 보여줍니다.
- 클라이언트 연결 차단: Redis 하위 스팬이 시작되기 전에 크고 빈 가로 간격이 있으면 애플리케이션 스레드가 연결 풀에서 사용 가능한 TCP 연결을 기다리는 동안 멈춰 있습니다.
- 애플리케이션 파싱 차단: Redis 하위 스팬이 종료된 후에 크고 빈 가로 간격이 있으면 애플리케이션이 반환된 페이로드를 파싱 하거나 처리하는 데 어려움을 겪습니다. 이 문제는 멀티 메가바이트 JSON 문자열에서 자주 발생합니다.
- 재시도: 동일한 상위 trace 내에서 동일한 명령어에 대해 여러 개의 짧은 하위 스팬이 순차적으로 발생하면 클라이언트에서 네트워크 패킷 손실이 발생할 수 있으며 지수 백오프 재시도 루프를 트리거해야 합니다.

문제 해결

이 섹션에서는 클라이언트 측 측정항목을 사용하여 식별할 수 있는 일반적인 성능 문제를 나열하고, 근본 원인을 설명하고, 문제를 해결하는 방법에 대한 안내를 제공합니다.

문제	원인	문제 해결
애플리케이션에 지연 시간이 갑자기 급증하지만 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 낮음). 하지만 반환되는 페이로드가 큽니다 (예: 15MB 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%) trace 폭포식: 명령어가 많은 시간 소요하는 것으로 표시됨	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~100ms 이상) `redis.googleapis.com/commands/ usec_per_call` (Memorystore for Redis 서버 측정항목): 매우 낮음 (< 1ms) `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에서 분산 trace 보기

문제 해결

다음 단계

클라이언트 측 측정항목을 사용하여 높은 지연 시간 문제 해결