feat: implement diagnostic channels for observability #3195

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open

logaretm wants to merge 37 commits into redis:master from logaretm:feat/tracing-channel

packages/client/index.ts

-Original file line number
+Diff line change
@@ Expand Up @@
     export { BasicClientSideCache, BasicPooledClientSideCache } from './lib/client/cache';
     export { OpenTelemetry } from './lib/opentelemetry';
+    export {
+      CHANNELS,
+      type ChannelEvents,
+      type CommandTraceContext,
+      type BatchCommandTraceContext,
+      type BatchOperationContext,
+      type ConnectTraceContext,
+      type ConnectionReadyEvent,
+      type ConnectionClosedEvent,
+      type ConnectionRelaxedTimeoutEvent,
+      type ConnectionHandoffEvent,
+      type ConnectionWaitContext,
+      type ClientErrorEvent,
+      type MaintenanceNotificationEvent,
+      type PubSubMessageEvent,
+      type CacheRequestEvent,
+      type CacheEvictionEvent,
+      type CommandReplyEvent,
+    } from './lib/client/tracing';

packages/client/lib/client/cache.ts

-Original file line number
+Diff line change
@@ Expand Up / @@ -2,7 +2,7 @@ import { EventEmitter } from 'stream'; @@
     import RedisClient from '.';
     import { RedisArgument, ReplyUnion, TransformReply, TypeMapping } from '../RESP/types';
     import { BasicCommandParser } from './parser';
-    import { OTelMetrics, CSC_RESULT, CSC_EVICTION_REASON } from '../opentelemetry';
+    import { publish, CHANNELS } from './tracing';
     /**
      * A snapshot of cache statistics.
@@ Expand Down Expand Up @@
           // If instanceof is "too slow", can add a "type" and then use an "as" cast to call proper getters.
           if (cacheEntry instanceof ClientSideCacheEntryValue) { // "2b1"
             this.#statsCounter.recordHits(1);
-            OTelMetrics.instance.clientSideCacheMetrics.recordCacheRequest(
-              CSC_RESULT.HIT,
-              client._clientId,
-            );
-            OTelMetrics.instance.clientSideCacheMetrics.recordNetworkBytesSaved(
-              cacheEntry.value,
-              client._clientId,
-            );
+            publish(CHANNELS.CACHE_REQUEST, () => ({ result: 'hit', clientId: client._clientId }));
             return structuredClone(cacheEntry.value);
           } else if (cacheEntry instanceof ClientSideCacheEntryPromise) { // 2b2
             // This counts as a miss since the value hasn't been fully loaded yet.
             this.#statsCounter.recordMisses(1);
-            OTelMetrics.instance.clientSideCacheMetrics.recordCacheRequest(
-              CSC_RESULT.MISS,
-              client._clientId,
-            );
+            publish(CHANNELS.CACHE_REQUEST, () => ({ result: 'miss', clientId: client._clientId }));
             reply = await cacheEntry.promise;
           } else {
             throw new Error("unknown cache entry type");
           }
         } else { // 3/3a
           this.#statsCounter.recordMisses(1);
-          OTelMetrics.instance.clientSideCacheMetrics.recordCacheRequest(
-            CSC_RESULT.MISS,
-            client._clientId,
-          );
+          publish(CHANNELS.CACHE_REQUEST, () => ({ result: 'miss', clientId: client._clientId }));
           const startTime = performance.now();
           const promise = fn();
@@ Expand Down Expand Up @@
           this.clear(false);
           // Record invalidations as server-initiated evictions
           if (oldSize > 0) {
-            OTelMetrics.instance.clientSideCacheMetrics.recordCacheEviction(CSC_EVICTION_REASON.INVALIDATION, oldSize);
+            publish(CHANNELS.CACHE_EVICTION, () => ({ reason: 'invalidation', count: oldSize }));
           }
           this.emit("invalidate", key);
@@ Expand All @@
           this.#keyToCacheKeySetMap.delete(key.toString());
           if (deletedCount > 0) {
             // Record invalidations as server-initiated evictions
-            OTelMetrics.instance.clientSideCacheMetrics.recordCacheEviction(CSC_EVICTION_REASON.INVALIDATION, deletedCount);
+            publish(CHANNELS.CACHE_EVICTION, () => ({ reason: 'invalidation', count: deletedCount }));
           }
         }
@@ Expand Down Expand Up @@
           this.delete(cacheKey);
           this.#statsCounter.recordEvictions(1);
           // Entry failed validation - this is TTL expiry since invalidation marks are handled separately
-          OTelMetrics.instance.clientSideCacheMetrics.recordCacheEviction(CSC_EVICTION_REASON.TTL);
+          publish(CHANNELS.CACHE_EVICTION, () => ({ reason: 'ttl', count: 1 }));
           this.emit("cache-evict", cacheKey);
           return undefined;
@@ Expand Down Expand Up @@
           this.deleteOldest();
           this.#statsCounter.recordEvictions(1);
           // Eviction due to cache capacity limit
-          OTelMetrics.instance.clientSideCacheMetrics.recordCacheEviction(CSC_EVICTION_REASON.FULL);
+          publish(CHANNELS.CACHE_EVICTION, () => ({ reason: 'full', count: 1 }));
         }
         this.#cacheKeyToEntryMap.set(cacheKey, cacheEntry);
@@ Expand Down @@

packages/client/lib/client/enterprise-maintenance-manager.ts

            
                      Original file line number
                      Diff line number
                      Diff line change
                  
    @@ -7,8 +7,7 @@ import { setTimeout } from "node:timers/promises";
  
    import { RedisTcpSocketOptions } from "./socket";

    import diagnostics_channel from "node:diagnostics_channel";

    import { RedisArgument } from "../RESP/types";

    import {  OTelMetrics } from "../opentelemetry";

    import { METRIC_ERROR_ORIGIN } from "../opentelemetry/types";

    import { publish, CHANNELS } from "./tracing";

    type RedisType = RedisClient<any, any, any, any, any>;

    @@ -124,12 +123,12 @@ export default class EnterpriseMaintenanceManager {
  
          errorHandler: (error: Error) => {

            dbgMaintenance("handshake failed:", error);

            OTelMetrics.instance.resiliencyMetrics.recordClientErrors({

            publish(CHANNELS.ERROR, () => ({

              error,

              origin: METRIC_ERROR_ORIGIN.CLIENT,

              origin: 'client',

              internal: true,

              clientId,

            });

            }));

            if (options.maintNotifications === "enabled") {

              throw error;

    @@ -160,10 +159,10 @@ export default class EnterpriseMaintenanceManager {
  
        const type = String(push[0]);

        OTelMetrics.instance.resiliencyMetrics.recordMaintenanceNotifications(

          type,

          this.#client._clientId,

        );

        publish(CHANNELS.MAINTENANCE, () => ({

          notification: type,

          clientId: this.#client._clientId,

        }));

        emitDiagnostics({

          type,

    @@ -306,9 +305,7 @@ export default class EnterpriseMaintenanceManager {
  
        dbgMaintenance("Resume writing");

        this.#client._unpause();

        this.#onMigrated();

        OTelMetrics.instance.connectionBasicMetrics.recordConnectionHandoff(

          this.#client._clientId,

        );

        publish(CHANNELS.CONNECTION_HANDOFF, () => ({ clientId: this.#client._clientId }));

      };

      #onMigrating = () => {

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: implement diagnostic channels for observability #3195

Uh oh!

Diff view

Diff view

There are no files selected for viewing

Uh oh!

Uh oh!

feat: implement diagnostic channels for observability #3195

Are you sure you want to change the base?

Uh oh!

feat: implement diagnostic channels for observability #3195

Uh oh!

Uh oh!

Diff view

Diff view

There are no files selected for viewing

Uh oh!

Uh oh!