Usa métricas del cliente para solucionar problemas de latencia alta

Si bien Memorystore para Redis proporciona métricas en tiempo real del servidor para supervisar la capacidad de procesamiento, el uso de CPU y el uso de memoria, es posible que estos datos por sí solos no expliquen por qué tu aplicación cliente experimenta una latencia alta dentro de sistemas distribuidos complejos.

Las métricas del cliente resuelven este problema, ya que proporcionan transparencia en el ciclo completo de solicitud-respuesta. Miden un comando desde el momento en que la aplicación lo inicia hasta que la aplicación procesa la respuesta. Si capturas estos puntos de datos, puedes determinar con precisión si la latencia se origina en la lógica de la aplicación, la ruta de red o el servidor de Redis.

Antes de comenzar

Asegúrate de que tu aplicación cliente use una cuenta de servicio y que se le asignen los siguientes roles de Identity and Access Management (IAM):

  • roles/cloudtrace.agent (agente de Cloud Trace)
  • roles/monitoring.metricWriter (escritor de métricas de Monitoring)

Para obtener más información para otorgar roles, consulta la guía de inicio rápido Otorga un rol de IAM con la Google Cloud consola.

Habilita la API de Cloud Monitoring

Para exportar métricas del cliente a Monitoring, tu aplicación requiere que la API de Monitoring esté habilitada. Exportar y visualizar estas métricas en Monitoring te permite identificar la causa raíz de los cuellos de botella para determinar si se origina la latencia.

Para habilitar la API de Monitoring, haz lo siguiente:

  1. En la Google Cloud consola de, ve a la página APIs y servicios.

    Ir a APIs y servicios

  2. Selecciona el proyecto en el que creaste la instancia de Memorystore para Redis.

  3. Haz clic en Habilitar las APIs y los servicios.

  4. Busca monitoring.

  5. En los resultados de la búsqueda, haz clic en API de Cloud Monitoring.

  6. Si aparece API habilitada, la API ya está habilitada. De lo contrario, haz clic en Habilitar.

Habilita la API de Cloud Trace

Para ver seguimientos distribuidos en Trace, debes habilitar la API de Trace. Luego, puedes usar el Explorador de Trace para ver estos seguimientos, diagnosticar cuellos de botella y aislar la fuente de latencia en tu aplicación.

Para habilitar la API de Trace, haz lo siguiente:

  1. En la Google Cloud consola de, ve a la página APIs y servicios.

    Ir a APIs y servicios

  2. Selecciona el proyecto en el que creaste la instancia de Memorystore para Redis.

  3. Haz clic en Habilitar las APIs y los servicios.

  4. Busca trace.

  5. En los resultados de la búsqueda, haz clic en API de Cloud Trace.

  6. Si aparece API habilitada, la API ya está habilitada. De lo contrario, haz clic en Habilitar.

Habilita las métricas del cliente

Para habilitar las métricas del cliente, agrega el OpenTelemetry de OpenTelemetry, el exportador de Cloud Monitoring y el exportador de Cloud Trace al código de tu aplicación. La instrumentación de OpenTelemetry, que se ejecuta directamente dentro de la biblioteca cliente de Redis de tu aplicación, captura las métricas. Esto permite que tu aplicación registre puntos de datos de latencia y los exporte a Monitoring y Trace para su visualización.

Para habilitar las métricas del cliente, puedes usar Go, Java, Node.js, o Python. La información para habilitar las métricas de cada lenguaje aparece en las siguientes pestañas.

Go

  1. Para instalar las dependencias requeridas de OpenTelemetry y Google Cloud exporter, ejecuta los siguientes comandos en tu terminal:

      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. Para habilitar las métricas del cliente, crea un archivo main.go y agrégale el siguiente código:

    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. Ejecuta tu aplicación durante al menos un minuto para darle al exportador el tiempo suficiente para agrupar y enviar las métricas publicadas a Monitoring.

Java

  1. Para instalar las dependencias requeridas de OpenTelemetry y Google Cloud exporter dependencies, agrega el siguiente código al archivo pom.xml de tu aplicación:

    <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. Para habilitar las métricas del cliente, crea un archivo RedisTelemetryApp.java y agrégale el siguiente código:

    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. Ejecuta tu aplicación durante al menos un minuto para darle al exportador el tiempo suficiente para agrupar y enviar las métricas publicadas a Monitoring.

Node.js

  1. Para instalar las dependencias requeridas de OpenTelemetry y Google Cloud exporter, ejecuta los siguientes comandos en tu terminal:

      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. Para habilitar las métricas del cliente, crea un archivo server.js y agrégale el siguiente código:

    
'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. Ejecuta tu aplicación durante al menos un minuto para darle al exportador el tiempo suficiente para agrupar y enviar las métricas publicadas a Monitoring.

Python

  1. Para instalar las dependencias requeridas de OpenTelemetry y Google Cloud exporter, ejecuta los siguientes comandos en tu terminal:

      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. Para habilitar las métricas del cliente, crea un archivo main.py y agrégale el siguiente código a tu aplicación:

    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. Ejecuta tu aplicación durante al menos un minuto para darle al exportador el tiempo suficiente para agrupar y enviar las métricas publicadas a Monitoring.

Visualiza métricas en Monitoring

Después de habilitar las métricas del cliente y ejecutar tu aplicación durante al menos un minuto para darle al exportador el tiempo suficiente para agrupar y enviar métricas a Monitoring, usa Monitoring para visualizar tus métricas, agruparlas por operación o instancia y aplicar agregadores para supervisar el rendimiento de tu aplicación.

Para ver las métricas en Monitoring, haz lo siguiente:

  1. En la Google Cloud consola de, ve a la página Explorador de métricas.

    Ir al Explorador de métricas

  2. Seleccionar tu Google Cloud proyecto.

  3. Haz clic en Selecciona una métrica.

  4. Busca workload.googleapis.com/redis.

  5. Selecciona una métrica del cliente. Agrupa los datos por operation y instance según sea necesario, y elige un agregador. Para explorar más opciones, consulta Selecciona métricas cuando uses el Explorador de métricas.

Visualiza seguimientos distribuidos en Trace

Después de que tu aplicación comience a exportar datos, puedes usar Trace para visualizar el ciclo completo de solicitud-respuesta de tus comandos de Redis. Ver tus seguimientos distribuidos en Trace te permite diagnosticar cuellos de botella para que puedas aislar rápidamente la fuente exacta de latencia en tu aplicación.

Para ver seguimientos distribuidos en Trace, haz lo siguiente:

  1. En la Google Cloud consola de, ve a la página Explorador de seguimiento.

    Ir al Explorador de seguimiento

  2. Selecciona un seguimiento reciente representado por un punto en el diagrama de dispersión.

  3. Examina la vista de cascada para aislar la fuente de latencia. Para ello, identifica los siguientes cuellos de botella:

    • Duración total de la solicitud: La barra de nivel superior (superior) muestra el tiempo total que debes esperar para que finalice la operación.

    • Latencia de red y servidor (RTT): Las barras secundarias (como las etiquetadas como GET o SET) muestran el tiempo que el comando pasó viajando por la red y ejecutándose en el servidor de Memorystore para Redis.

    • Bloqueo de conexión del cliente: Si hay un espacio horizontal grande y vacío antes de que comience el intervalo secundario de Redis, el subproceso de la aplicación se queda esperando una conexión TCP disponible del grupo de conexiones.

    • Bloqueo de análisis de la aplicación: Si hay un espacio horizontal grande y vacío después de que finaliza el intervalo secundario de Redis, la aplicación tiene problemas para analizar o procesar la carga útil que se muestra. Esto ocurre a menudo con las cadenas JSON de varios megabytes.

    • Reintentos: Si ves varios intervalos secundarios cortos para el mismo comando que ocurren de forma secuencial dentro del mismo seguimiento superior, es posible que tu cliente experimente una pérdida de paquetes de red y deba activar su bucle de reintento de retirada exponencial.

Solucionar problemas

En esta sección, se enumeran los problemas de rendimiento comunes que puedes identificar con las métricas del cliente, se explican sus causas raíz y se proporcionan instrucciones para solucionar los problemas.

Problema Causa Solucionar problemas

Tu aplicación experimenta un aumento repentino de la latencia, pero Memorystore para Redis parece estar en perfecto estado.

  • workload.googleapis.com/
    redis_client_blocking_latency     (métrica del cliente): aumento
  • workload.googleapis.com/redis_client_rtt (métrica del cliente): baja o típica
  • redis.googleapis.com/commands/
    usec_per_call     (métrica del servidor de Memorystore para Redis): baja o típica
  • redis.googleapis.com/clients/connected (métrica del servidor de Memorystore para Redis): se estabiliza en un número específico
El cuello de botella está estrictamente dentro de tu aplicación. Tus subprocesos intentan ejecutar comandos de Redis, pero el grupo de conexiones está completamente agotado. La alta redis_client_blocking_latency representa el tiempo que tu código pasa esperando un socket TCP disponible antes de que se envíe el comando a la red. Para controlar el tráfico simultáneo más alto, aumenta los límites de tamaño del grupo de conexiones en la configuración de tu cliente de Redis (por ejemplo, MaxActive para Go, MaxTotal para Java o max_connections para Node.js y Python).

La solicitud se completa, pero el extremo tarda mucho más de lo esperado No hay problemas asociados con el estado de tu red o servidor.

  • workload.googleapis.com/
    redis_application_blocking_latency     (métrica del cliente): aumento
  • workload.googleapis.com/redis_client_rtt (métrica del cliente): baja o típica
  • redis.googleapis.com/commands/
    usec_per_call     (métrica del servidor de Memorystore para Redis): baja o típica
  • redis.googleapis.com/stats/
    network_traffic     (Bytes de salida) (métrica del servidor de Memorystore para Redis): aumentos significativos
 Memorystore para Redis ejecuta el comando y la red transfiere la carga útil rápidamente (RTT bajo). Sin embargo, la carga útil que se muestra es grande (por ejemplo, una cadena JSON de 15 MB). Tu aplicación experimenta una redis_application_blocking_latency alta porque la aplicación consume recursos excesivos mientras asigna memoria y deserializa esa cadena grande en un objeto. Optimiza tu modelo de datos. No almacenes objetos JSON masivos en claves únicas. Divide los datos con hashes de Redis (HSET) y usa HGET o HMGET para recuperar solo los campos específicos que necesitas.

La latencia de tu aplicación orientada al usuario aumenta, pero tus métricas de Redis informan una latencia baja del servidor y extracciones típicas del grupo de conexiones.

  • workload.googleapis.com/redis_retry_count (métrica del cliente): aumentos
  • workload.googleapis.com/
    redis_connectivity_error_count     (métrica del cliente): puede mostrar incrementos transitorios
  • workload.googleapis.com/redis_client_rtt (métrica del cliente): baja o típica para solicitudes exitosas
  • redis.googleapis.com/commands/
    usec_per_call     (métrica del servidor de Memorystore para Redis): baja o típica
 Debido a que redis_client_rtt solo captura el RTT de las solicitudes exitosas, no refleja la duración del tiempo de espera de un paquete fallido. Cuando tu aplicación experimenta caídas de paquetes transitorias o restablecimientos de TCP, la lógica de reintento de tu cliente instrumentado incrementa el redis_retry_count y activa su bucle de retirada exponencial. Esto introduce un tiempo de suspensión entre los intentos (por ejemplo, 100ms, 200ms, o 400ms). El usuario experimenta una latencia total alta, pero la causa raíz subyacente es una pérdida de paquetes de red, que activa retrasos de suspensión del cliente. Consulta tus registros de flujo de VPC para ver si hay paquetes descartados, limitación de ancho de banda, o anomalías de enrutamiento entre regiones. Si experimentas tiempos de espera agresivos, asegúrate de que los tiempos de espera de conexión del cliente (socket_timeout o connect_timeout) sean mayores que el RTT esperado para tener en cuenta la fluctuación de red transitoria.

Todo se detiene y todas las capas de la canalización de telemetría informan latencia alta.

  • workload.googleapis.com/redis_client_rtt (métrica del cliente): alta
  • redis.googleapis.com/commands/
    usec_per_call     (métrica del servidor de Memorystore para Redis): alta
  • redis.googleapis.com/stats/
    cpu_utilization_main_thread     (métrica del servidor de Memorystore para Redis): alta (por ejemplo, cerca de 1 s/s, o 100%)
  • Cascada de Trace: Muestra un comando que tarda mucho tiempo
Redis tiene un solo subproceso. Cuando ejecutas un comando de complejidad temporal O(N) (como KEYS *, SMEMBERS en un conjunto masivo o HGETALL en un hash con millones de campos), el motor de Redis se detiene para cumplir con esa solicitud. Mientras se ejecuta ese comando, todas las demás solicitudes de la aplicación se ponen en cola, lo que provoca un aumento de la latencia en todo el sistema. Debido a que tu redis_client_rtt personalizado coincide con la latencia del servidor (commands/usec_per_call), el servidor que ejecuta el comando es el cuello de botella.

Abre Trace y observa los comandos de Redis en los intervalos lentos para identificar qué consulta causa el bloqueo. Reemplaza los comandos de bloqueo por comandos que no bloqueen en tu código.

Para iterar a través de conjuntos de datos grandes de forma incremental sin bloquear el subproceso del servidor, usa SCAN, SSCAN, o HSCAN.

Tu aplicación informa una latencia de referencia constante y elevada para todos los comandos de Redis, incluso cuando el tráfico es bajo.

  • workload.googleapis.com/redis_client_rtt (métrica del cliente): constantemente elevada (p50 y p99 son ambas ~30-100 ms+)
  • redis.googleapis.com/commands/
    usec_per_call     (métrica del servidor de Memorystore para Redis): extremadamente baja (< 1 ms)
  • workload.googleapis.com/
    redis_client_blocking_latency
    y redis_application_blocking_latency: baja o típica
 El servidor de Redis ejecuta comandos al instante, pero tu aplicación y tu instancia se implementan en diferentes regiones (por ejemplo, us-central1 y us-east1). Cada paquete de red debe viajar a través de la infraestructura de nube física de Google Cloud entre estos centros de datos geográficos. Esto da como resultado una penalización de latencia entre regiones obligatoria de la velocidad de la luz para cada viaje de ida y vuelta. Para reducir la latencia, implementa tu aplicación para que resida en la misma región y zona que tu instancia. Para ver la región de tu aplicación y instancia, usa la Google Cloud consola.

¿Qué sigue?