Menggunakan metrik sisi klien untuk memecahkan masalah latensi tinggi

Meskipun Memorystore for Redis menyediakan metrik sisi server real-time untuk memantau throughput, pemanfaatan CPU, dan penggunaan memori, data ini saja mungkin tidak menjelaskan mengapa aplikasi klien Anda mengalami latensi tinggi dalam sistem terdistribusi yang kompleks.

Metrik sisi klien menyelesaikan masalah ini dengan memberikan transparansi ke dalam siklus permintaan-respons penuh. Metrik ini mengukur perintah dari saat aplikasi memulainya hingga aplikasi memproses respons. Dengan merekam titik data ini, Anda dapat menentukan secara akurat apakah latensi berasal dari logika aplikasi, jalur jaringan, atau server Redis.

Sebelum memulai

Pastikan aplikasi klien Anda menggunakan akun layanan dan peran Identity and Access Management (IAM) berikut ditetapkan ke akun layanan tersebut:

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

Untuk mengetahui informasi selengkapnya tentang cara memberikan peran, lihat panduan memulai Memberikan peran IAM menggunakan konsol Google Cloud .

Aktifkan Cloud Monitoring API

Untuk mengekspor metrik sisi klien ke Monitoring, aplikasi Anda memerlukan Monitoring API diaktifkan. Dengan mengekspor dan memvisualisasikan metrik ini di Monitoring, Anda dapat menunjukkan penyebab utama hambatan untuk menentukan asal latensi.

Untuk mengaktifkan Monitoring API, lakukan hal berikut:

  1. Di konsol Google Cloud , buka halaman APIs & Services.

    Buka APIs and services

  2. Pilih project tempat Anda membuat instance Memorystore for Redis.

  3. Klik Enable APIs and services.

  4. Telusuri monitoring.

  5. Di hasil penelusuran, klik Cloud Monitoring API.

  6. Jika API enabled muncul, berarti API sudah diaktifkan. Jika tidak, klik Aktifkan.

Aktifkan Cloud Trace API

Untuk melihat rekaman aktivitas terdistribusi di Trace, Anda harus mengaktifkan Trace API. Kemudian, Anda dapat menggunakan Penjelajah Trace untuk melihat trace ini, mendiagnosis bottleneck, dan mengisolasi sumber latensi dalam aplikasi Anda.

Untuk mengaktifkan Trace API, lakukan hal berikut:

  1. Di konsol Google Cloud , buka halaman APIs & Services.

    Buka APIs and services

  2. Pilih project tempat Anda membuat instance Memorystore for Redis.

  3. Klik Enable APIs and services.

  4. Telusuri trace.

  5. Di hasil penelusuran, klik Cloud Trace API.

  6. Jika API enabled muncul, berarti API sudah diaktifkan. Jika tidak, klik Aktifkan.

Mengaktifkan metrik sisi klien

Untuk mengaktifkan metrik sisi klien, tambahkan SDK OpenTelemetry, pengekspor Cloud Monitoring, dan pengekspor Cloud Trace ke kode aplikasi Anda. Instrumentasi OpenTelemetry, yang berjalan langsung di dalam library klien Redis aplikasi Anda, akan merekam metrik. Hal ini memungkinkan aplikasi Anda merekam titik data latensi dan mengekspornya ke Monitoring dan Trace untuk visualisasi.

Untuk mengaktifkan metrik sisi klien, Anda dapat menggunakan Go, Java, Node.js, atau Python. Informasi untuk mengaktifkan metrik untuk setiap bahasa muncul di tab berikutnya.

Go

  1. Untuk menginstal dependensi OpenTelemetry dan Google Cloud exporter yang diperlukan, jalankan perintah berikut di terminal Anda:

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

  2. Untuk mengaktifkan metrik sisi klien, buat file main.go dan tambahkan kode berikut ke dalamnya:

    package main

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

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

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

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

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

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

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

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

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

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

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

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

	return client, shutdown, nil
}

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

  3. Jalankan aplikasi Anda setidaknya selama satu menit agar pengekspor memiliki cukup waktu untuk membuat batch dan mengirim metrik yang dipublikasikan ke Monitoring.

Java

  1. Untuk menginstal dependensi OpenTelemetry dan Google Cloud exporter yang diperlukan, tambahkan kode berikut ke file pom.xml aplikasi Anda:

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

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

  2. Untuk mengaktifkan metrik sisi klien, buat file RedisTelemetryApp.java dan tambahkan kode berikut ke dalamnya:

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

  3. Jalankan aplikasi Anda setidaknya selama satu menit agar pengekspor memiliki cukup waktu untuk membuat batch dan mengirim metrik yang dipublikasikan ke Monitoring.

Node.js

  1. Untuk menginstal dependensi OpenTelemetry dan Google Cloud exporter yang diperlukan, jalankan perintah berikut di terminal Anda:

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

  2. Untuk mengaktifkan metrik sisi klien, buat file server.js dan tambahkan kode berikut ke dalamnya:

    
'use strict';

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

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

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

const redis = require('redis');

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

  3. Jalankan aplikasi Anda setidaknya selama satu menit agar pengekspor memiliki cukup waktu untuk membuat batch dan mengirim metrik yang dipublikasikan ke Monitoring.

Python

  1. Untuk menginstal dependensi OpenTelemetry dan Google Cloud exporter yang diperlukan, jalankan perintah berikut di terminal Anda:

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

  2. Untuk mengaktifkan metrik sisi klien, buat file main.py dan tambahkan kode berikut ke aplikasi Anda:

    import os
import time

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




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

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

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

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

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

    return tracer, redis_metrics, tracer_provider, meter_provider


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

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


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

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

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

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

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

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

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

            return response

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

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

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

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

        tracer_provider.force_flush()
        meter_provider.force_flush()

  3. Jalankan aplikasi Anda setidaknya selama satu menit agar pengekspor memiliki cukup waktu untuk membuat batch dan mengirim metrik yang dipublikasikan ke Monitoring.

Melihat metrik di Monitoring

Setelah mengaktifkan metrik sisi klien dan menjalankan aplikasi selama setidaknya satu menit untuk memberi pengekspor cukup waktu untuk mengelompokkan dan mengirim metrik ke Monitoring, gunakan Monitoring untuk memvisualisasikan metrik, mengelompokkannya menurut operasi atau instance, dan menerapkan penggabung untuk memantau performa aplikasi Anda.

Untuk melihat metrik di Monitoring, lakukan hal berikut:

  1. Di konsol Google Cloud , buka halaman Metrics Explorer.

    Buka Metrics Explorer

  2. Pilih project Google Cloud Anda.

  3. Klik Select a metric.

  4. Telusuri workload.googleapis.com/redis.

  5. Pilih metrik sisi klien. Kelompokkan data menurut operation dan instance sesuai kebutuhan, lalu pilih agregator. Untuk mempelajari opsi lainnya, lihat Memilih metrik saat menggunakan Metrics Explorer.

Melihat rekaman aktivitas terdistribusi di Trace

Setelah aplikasi Anda mulai mengekspor data, Anda dapat menggunakan Trace untuk memvisualisasikan siklus permintaan-respons penuh perintah Redis Anda. Dengan melihat rekaman aktivitas terdistribusi di Trace, Anda dapat mendiagnosis hambatan sehingga dapat dengan cepat mengisolasi sumber latensi yang tepat dalam aplikasi.

Untuk melihat rekaman aktivitas terdistribusi di Trace, lakukan hal berikut:

  1. Di konsol Google Cloud , buka halaman Trace explorer.

    Buka Trace Explorer

  2. Pilih rekaman aktivitas terbaru yang diwakili oleh titik pada diagram sebar.

  3. Periksa tampilan waterfall untuk mengisolasi sumber latensi dengan mengidentifikasi hambatan berikut:

    • Total durasi permintaan: kolom tingkat teratas (induk) menampilkan total waktu yang harus Anda tunggu hingga operasi selesai.

    • Latensi jaringan dan server (RTT): batang turunan (seperti yang berlabel GET atau SET) menunjukkan waktu yang dihabiskan perintah untuk melintasi jaringan dan berjalan di server Memorystore for Redis.

    • Pemblokiran koneksi klien: jika ada jeda horizontal yang besar dan kosong sebelum rentang turunan Redis dimulai, maka thread aplikasi akan macet menunggu koneksi TCP yang tersedia dari kumpulan koneksi.

    • Pemblokiran parsing aplikasi: jika ada jeda horizontal yang besar dan kosong setelah rentang turunan Redis berakhir, berarti aplikasi kesulitan mem-parsing atau memproses payload yang ditampilkan. Hal ini sering terjadi pada string JSON multi-megabyte.

    • Percobaan ulang: jika Anda melihat beberapa rentang turunan pendek untuk perintah yang sama yang terjadi secara berurutan dalam rekaman aktivitas induk yang sama, klien Anda mungkin mengalami kehilangan paket jaringan dan harus memicu loop percobaan ulang dengan jeda eksponensial.

Memecahkan masalah

Bagian ini mencantumkan masalah performa umum yang dapat Anda identifikasi menggunakan metrik sisi klien, menjelaskan penyebab utamanya, dan memberikan panduan tentang cara memecahkan masalah tersebut.

Masalah Penyebab Memecahkan masalah

Aplikasi Anda mengalami lonjakan latensi yang tiba-tiba, tetapi Memorystore for Redis tampak berfungsi dengan baik.

  • workload.googleapis.com/
    redis_client_blocking_latency     (metrik sisi klien): melonjak
  • workload.googleapis.com/redis_client_rtt (metrik sisi klien): rendah / normal
  • redis.googleapis.com/commands/
    usec_per_call     (Metrik server Memorystore for Redis): rendah / khas
  • redis.googleapis.com/clients/connected (Metrik server Memorystore for Redis): tetap di angka tertentu
Hambatan sepenuhnya berada di dalam aplikasi Anda. thread Anda mencoba menjalankan perintah Redis, tetapi kumpulan koneksi sudah habis sepenuhnya. redis_client_blocking_latency tinggi menunjukkan waktu yang dihabiskan kode Anda untuk menunggu soket TCP yang tersedia sebelum perintah dikirim ke jaringan. Untuk menangani traffic serentak yang lebih tinggi, tingkatkan batas ukuran kumpulan koneksi dalam konfigurasi klien Redis Anda (misalnya, MaxActive untuk Go, MaxTotal untuk Java, atau max_connections untuk Node.js dan Python).

Permintaan selesai, tetapi endpoint memerlukan waktu yang jauh lebih lama dari yang diperkirakan. Tidak ada masalah yang terkait dengan kondisi jaringan atau server Anda.

  • workload.googleapis.com/
    redis_application_blocking_latency     (metrik sisi klien): melonjak
  • workload.googleapis.com/redis_client_rtt (metrik sisi klien): rendah / normal
  • redis.googleapis.com/commands/
    usec_per_call     (Metrik server Memorystore for Redis): rendah / khas
  • redis.googleapis.com/stats/
    network_traffic     (Bytes keluar) (Metrik server Memorystore for Redis): melonjak tajam
 Memorystore for Redis menjalankan perintah dan jaringan mentransfer payload dengan cepat (RTT rendah). Namun, payload yang ditampilkan berukuran besar (misalnya, string JSON 15 MB). Aplikasi Anda mengalami redis_application_blocking_latency yang tinggi karena aplikasi menggunakan resource yang berlebihan saat mengalokasikan memori dan mendeserialisasi string besar tersebut menjadi objek. Optimalkan model data Anda. Jangan menyimpan blob JSON besar dalam satu kunci. Pecah data menggunakan hash Redis (HSET) dan gunakan HGET atau HMGET untuk mengambil hanya kolom tertentu yang Anda butuhkan.

Latensi aplikasi yang terlihat pengguna meningkat, tetapi metrik Redis Anda melaporkan latensi server yang rendah dan checkout pool koneksi yang normal.

  • workload.googleapis.com/redis_retry_count (metrik sisi klien): lonjakan
  • workload.googleapis.com/
    redis_connectivity_error_count     (metrik sisi klien): mungkin menampilkan peningkatan sementara
  • workload.googleapis.com/redis_client_rtt (metrik sisi klien): rendah / khas untuk permintaan yang berhasil
  • redis.googleapis.com/commands/
    usec_per_call     (Metrik server Memorystore for Redis): rendah / umum
 Karena redis_client_rtt hanya merekam RTT permintaan yang berhasil, tidak mencerminkan durasi waktu tunggu paket yang gagal. Saat aplikasi Anda mengalami penurunan paket sementara atau reset TCP, logika percobaan ulang klien yang diinstrumentasikan akan meningkatkan redis_retry_count dan memicu loop backoff eksponensialnya. Hal ini menyebabkan waktu tunggu antara percobaan (misalnya, 100ms, 200ms, atau 400ms). Pengguna mengalami total latensi yang tinggi, tetapi penyebab utamanya adalah kehilangan paket jaringan, yang memicu penundaan tidur sisi klien. Periksa VPC Flow Logs Anda untuk mengetahui apakah ada paket yang dibatalkan, pembatasan bandwidth, atau anomali perutean lintas region. Jika Anda mengalami timeout agresif, pastikan timeout koneksi klien Anda (socket_timeout atau connect_timeout) lebih besar daripada RTT yang diharapkan untuk memperhitungkan jitter jaringan sementara.

Semua berhenti dan semua lapisan pipeline telemetri melaporkan latensi tinggi.

  • workload.googleapis.com/redis_client_rtt (metrik sisi klien): tinggi
  • redis.googleapis.com/commands/
    usec_per_call     (Metrik server Memorystore for Redis): tinggi
  • redis.googleapis.com/stats/
    cpu_utilization_main_thread     (Metrik server Memorystore for Redis): tinggi (misalnya, mendekati 1 s/s, atau 100%)
  • Waterfall rekaman aktivitas: menampilkan perintah yang membutuhkan waktu yang lama
 Redis adalah thread tunggal. Saat Anda menjalankan perintah kompleksitas waktu O(N), seperti KEYS *, SMEMBERS pada set data yang sangat besar, atau HGETALL pada hash dengan jutaan kolom, mesin Redis akan dijeda untuk memenuhi permintaan tersebut. Saat perintah tersebut berjalan, setiap permintaan aplikasi lainnya akan diantrekan, sehingga menyebabkan lonjakan latensi di seluruh sistem. Karena redis_client_rtt kustom Anda cocok dengan latensi server (commands/usec_per_call), server yang menjalankan perintah adalah hambatan.

Buka Trace dan lihat perintah Redis pada rentang yang lambat untuk mengidentifikasi kueri mana yang menyebabkan pemblokiran. Ganti perintah pemblokiran dengan perintah non-pemblokiran dalam kode Anda.

Untuk melakukan iterasi melalui set data besar secara inkremental tanpa mengunci thread server, gunakan SCAN, SSCAN, atau HSCAN.

Aplikasi Anda melaporkan latensi dasar yang konsisten dan tinggi untuk semua perintah Redis, bahkan saat traffic rendah.

  • workload.googleapis.com/redis_client_rtt (metrik sisi klien): terus meningkat (p50 dan p99 keduanya ~30-100 md+)
  • redis.googleapis.com/commands/
    usec_per_call     (Metrik server Memorystore for Redis): sangat rendah (< 1 md)
  • workload.googleapis.com/
    redis_client_blocking_latency
    dan redis_application_blocking_latency: rendah / normal
 Server Redis menjalankan perintah secara instan, tetapi aplikasi dan instance Anda di-deploy di region yang berbeda (misalnya, us-central1 dan us-east1). Setiap paket jaringan harus melintasi infrastruktur Google Cloud fisik di antara pusat data geografis ini. Hal ini akan menghasilkan penalti latensi lintas region yang wajib dan secepat cahaya untuk setiap perjalanan pulang pergi. Untuk mengurangi latensi, deploy aplikasi Anda agar berada di region dan zona yang sama dengan instance Anda. Untuk melihat region aplikasi dan instance, gunakan konsol Google Cloud .

Langkah berikutnya