Metrics experiment

.changeset/tough-planes-end.md

@@ -0,0 +1,5 @@
+---
+"@apollo/client": patch
+---
+
+add a MetricsLink

config/bundlesize.ts

@@ -3,7 +3,7 @@ import { join } from "path";
 import { gzipSync } from "zlib";
 import bytes from "bytes";
 
-const gzipBundleByteLengthLimit = bytes("33.02KB");
+const gzipBundleByteLengthLimit = bytes("33.12KB");
 const minFile = join("dist", "apollo-client.min.cjs");
 const minPath = join(__dirname, "..", minFile);
 const gzipByteLen = gzipSync(readFileSync(minPath)).byteLength;

src/__tests__/__snapshots__/exports.ts.snap

@@ -353,6 +353,7 @@ Array [
   "DEV",
   "DeepMerger",
   "Observable",
+  "Subject",
   "addTypenameToDocument",
   "argumentsObjectFromField",
   "asyncMap",

src/core/ApolloClient.ts

@@ -642,4 +642,8 @@ export class ApolloClient<TCacheShape> implements DataProxy {
   public setLink(newLink: ApolloLink) {
     this.link = this.queryManager.link = newLink;
   }
+
+  public get metrics(){
+    return this.queryManager.metrics
+  }
 }

src/core/QueryManager.ts

@@ -30,19 +30,20 @@ import {
   makeUniqueId,
   isDocumentNode,
   isNonNullObject,
-} from '../utilities';
-import { mergeIncrementalData } from '../utilities/common/incrementalResult';
-import { ApolloError, isApolloError, graphQLResultHasProtocolErrors } from '../errors';
+  Subject,
+} from "../utilities";
+import { mergeIncrementalData } from "../utilities/common/incrementalResult";
+import { ApolloError, isApolloError, graphQLResultHasProtocolErrors } from "../errors";
 import {
   QueryOptions,
   WatchQueryOptions,
   SubscriptionOptions,
   MutationOptions,
   ErrorPolicy,
   MutationFetchPolicy,
-} from './watchQueryOptions';
-import { ObservableQuery, logMissingFieldErrors } from './ObservableQuery';
-import { NetworkStatus, isNetworkRequestInFlight } from './networkStatus';
+} from "./watchQueryOptions";
+import { ObservableQuery, logMissingFieldErrors } from "./ObservableQuery";
+import { NetworkStatus, isNetworkRequestInFlight } from "./networkStatus";
 import {
   ApolloQueryResult,
   OperationVariables,
@@ -52,16 +53,12 @@ import {
   InternalRefetchQueriesOptions,
   InternalRefetchQueriesResult,
   InternalRefetchQueriesMap,
-} from './types';
-import { LocalState } from './LocalState';
+  MetricsEvents,
+} from "./types";
+import { LocalState } from "./LocalState";
 
-import {
-  QueryInfo,
-  QueryStoreValue,
-  shouldWriteResult,
-  CacheWriteBehavior,
-} from './QueryInfo';
-import { PROTOCOL_ERRORS_SYMBOL, ApolloErrorOptions } from '../errors';
+import { QueryInfo, QueryStoreValue, shouldWriteResult, CacheWriteBehavior } from "./QueryInfo";
+import { PROTOCOL_ERRORS_SYMBOL, ApolloErrorOptions } from "../errors";
 
 const { hasOwnProperty } = Object.prototype;
 
@@ -86,10 +83,13 @@ interface TransformCacheEntry {
 
 type DefaultOptions = import("./ApolloClient").DefaultOptions;
 
+export const traceIdSymbol = Symbol.for("apollo-trace-id");
+
 export class QueryManager<TStore> {
   public cache: ApolloCache<TStore>;
   public link: ApolloLink;
   public defaultOptions: DefaultOptions;
+  public readonly metrics = new Subject<MetricsEvents>();
 
   public readonly assumeImmutableResults: boolean;
   public readonly ssrMode: boolean;
@@ -735,6 +735,11 @@ export class QueryManager<TStore> {
     ).finally(() => this.stopQuery(queryId));
   }
 
+  private traceIdCounter = 1;
+  public generateTraceId() {
+    return String(this.traceIdCounter++);
+  }
+
   private queryIdCounter = 1;
   public generateQueryId() {
     return String(this.queryIdCounter++);
@@ -1187,6 +1192,7 @@ export class QueryManager<TStore> {
     const query = this.transform(options.query).document;
     const variables = this.getVariables(query, options.variables) as TVars;
     const queryInfo = this.getQuery(queryId);
+    const traceId = this.generateTraceId()
 
     const defaults = this.defaultOptions.watchQuery;
     let {
@@ -1204,7 +1210,7 @@ export class QueryManager<TStore> {
       errorPolicy,
       returnPartialData,
       notifyOnNetworkStatusChange,
-      context,
+      context: Object.assign({}, context, { [traceIdSymbol]: traceId }),
     });
 
     const fromVariables = (variables: TVars) => {
@@ -1272,6 +1278,8 @@ export class QueryManager<TStore> {
 
     concast.promise.then(cleanupCancelFn, cleanupCancelFn);
 
+    this.metrics.next({ ...normalized, type: 'request', cacheHit: !containsDataFromLink });
+
     return {
       concast,
       fromLink: containsDataFromLink,

src/core/types.ts

@@ -7,13 +7,14 @@ import { QueryInfo } from './QueryInfo';
 import { NetworkStatus } from './networkStatus';
 import { Resolver } from './LocalState';
 import { ObservableQuery } from './ObservableQuery';
-import { QueryOptions } from './watchQueryOptions';
+import { QueryOptions, WatchQueryOptions } from "./watchQueryOptions";
 import { Cache } from '../cache';
-import { IsStrictlyAny } from '../utilities';
+import { IsStrictlyAny } from "../utilities";
+import { traceIdSymbol } from "./QueryManager";
 
 export { TypedDocumentNode } from '@graphql-typed-document-node/core';
 
-export interface DefaultContext extends Record<string, any> {};
+export interface DefaultContext extends Record<string, any> {[traceIdSymbol]?: string};
 
 export type QueryListener = (queryInfo: QueryInfo) => void;
 
@@ -194,3 +195,6 @@ export interface Resolvers {
     [ field: string ]: Resolver;
   };
 }
+
+export type MetricsEvents = { type: "request"; cacheHit: boolean } & WatchQueryOptions &
+  Pick<Required<WatchQueryOptions>, "context">;

src/link/metrics-link/index.ts

@@ -0,0 +1,157 @@
+import { invariant } from "../../utilities/globals";
+
+import { ApolloLink, Operation, FetchResult, NextLink } from "../core";
+import { Observable, Subject, getOperationDefinition, getOperationName } from "../../utilities";
+import { ApolloClient, DefaultContext, MetricsEvents } from "../../core";
+import { Subscription } from "zen-observable-ts";
+import { traceIdSymbol } from "../../core/QueryManager";
+
+interface RequestMetrics {
+  endedAt: number;
+  responseCode: number | null;
+  finishedAs: "success" | "error";
+  errors?: readonly unknown[];
+}
+
+interface BaseMetrics {
+  operationType: "query" | "mutation" | "subscription";
+  operationName: string;
+  variables: unknown;
+  traceId: string;
+  startedAt: number;
+}
+
+export type MetricsLinkEvents =
+  | ({ type: "cacheHit" } & BaseMetrics)
+  | ({ type: "request" } & BaseMetrics & RequestMetrics);
+
+type ExtractInfoData =
+  | {
+      type: "cacheHit";
+      context: DefaultContext;
+    }
+  | {
+      type: "request";
+      context: DefaultContext;
+      operation: Operation;
+    };
+
+type ExtractInfo<AdditionalMetrics extends Record<string, unknown>> = (
+  data: ExtractInfoData
+) => AdditionalMetrics;
+
+type MetricsLinkOptions<AdditionalMetrics extends Record<string, unknown>> =
+  keyof AdditionalMetrics extends never
+    ? {
+        extractInfo?: ExtractInfo<AdditionalMetrics>;
+      }
+    : {
+        extractInfo: ExtractInfo<AdditionalMetrics>;
+      };
+
+export class MetricsLink<AdditionalMetrics extends Record<string, unknown> = {}> extends ApolloLink {
+  public metrics = new Subject<MetricsLinkEvents>();
+  private clientSubscription: Subscription | undefined;
+  private runningRequests = new Map<string, BaseMetrics & Partial<RequestMetrics>>();
+  private extractInfo?: ExtractInfo<AdditionalMetrics>;
+
+  constructor(options: MetricsLinkOptions<AdditionalMetrics>) {
+    super();
+    if (options && "extractInfo" in options) {
+      this.extractInfo = options.extractInfo;
+    }
+  }
+
+  public registerClient(client: ApolloClient<any>) {
+    if (this.clientSubscription) {
+      this.clientSubscription.unsubscribe();
+    }
+    client.metrics.subscribe(this.onClientRequest);
+  }
+  private onClientRequest = (ev: MetricsEvents) => {
+    if (ev.type === "request") {
+      const { query, cacheHit, context, variables } = ev;
+      const operationType = getOperationDefinition(query)?.operation;
+      const operationName = getOperationName(query);
+      const traceId = getTraceId(context);
+      invariant(operationType && operationName, "invalid operation forwarded");
+      const baseMetrics = {
+        traceId,
+        operationType,
+        operationName,
+        variables,
+        startedAt: Date.now(),
+      };
+
+      if (cacheHit) {
+        const extractedInfo = this.extractInfo?.({
+          type: "cacheHit",
+          context,
+        });
+        this.metrics.next({ ...extractedInfo, type: "cacheHit", ...baseMetrics });
+      } else {
+        this.runningRequests.set(traceId, baseMetrics);
+      }
+    }
+  };
+
+  request(operation: Operation, forward?: NextLink) {
+    invariant(forward, "MetricsLink cannot be the final link.");
+
+    const traceId = getTraceId(operation.getContext());
+
+    const request = forward(operation);
+    let initialResponse: FetchResult | undefined;
+
+    const emitRequestMetric = (dynamicData: Omit<RequestMetrics, "endedAt">) => {
+      const metrics = this.runningRequests.get(traceId);
+      if (!metrics) return;
+      this.runningRequests.delete(traceId);
+      const extractedInfo = this.extractInfo?.({
+        type: "request",
+        context: operation.getContext(),
+        operation,
+      });
+      this.metrics.next({
+        ...extractedInfo,
+        type: "request",
+        ...metrics,
+        endedAt: Date.now(),
+        ...dynamicData,
+      });
+    };
+
+    return new Observable<FetchResult>((observer) => {
+      request.subscribe({
+        next: (result) => {
+          initialResponse ??= result;
+          observer.next(result);
+        },
+        error: (error) => {
+          const { response } = operation.getContext();
+          emitRequestMetric({
+            responseCode: response?.status ?? null,
+            finishedAs: "error",
+            errors: [error],
+          });
+          observer.error(error);
+        },
+        complete: () => {
+          const { response } = operation.getContext();
+          emitRequestMetric({
+            responseCode: response.status,
+            finishedAs: "success",
+            errors: initialResponse!.errors,
+          });
+          observer.complete();
+        },
+      });
+    });
+  }
+}
+
+function getTraceId(context: DefaultContext | undefined) {
+  const traceId = context?.[traceIdSymbol];
+  invariant(traceId, "traceId missing in context");
+  return traceId;
+}

src/utilities/common/Subject.ts

@@ -0,0 +1,19 @@
+import { Observable, SubscriptionObserver } from "zen-observable-ts";
+import { fixObservableSubclass } from "../observables/subclassing";
+
+export class Subject<T> extends Observable<T> {
+  private subscribers = new Set<SubscriptionObserver<T>>();
+  constructor() {
+    super((s) => {
+      this.subscribers.add(s);
+      return () => this.subscribers.delete(s);
+    });
+  }
+  next(value: T) {
+    this.subscribers.forEach((s) => s.next(value));
+  }
+}
+
+// Necessary because the Subject constructor has a different
+// signature than the Observable constructor.
+fixObservableSubclass(Subject);

src/utilities/index.ts

@@ -96,5 +96,6 @@ export * from './common/compact';
 export * from './common/makeUniqueId';
 export * from './common/stringifyForDisplay';
 export * from './common/mergeOptions';
+export * from './common/Subject';
 
 export * from './types/IsStrictlyAny';