Add Initial Version of Fleet Awareness Plugin

federation-integration-testsuite-js/src/snapshotSerializers/readableSpanArraySerializer.ts

@@ -1,5 +1,5 @@
 import { Plugin } from 'pretty-format';
-import { ReadableSpan } from '@opentelemetry/tracing'
+import { ReadableSpan } from '@opentelemetry/sdk-trace-base';
 
 export default {
     test(value: any) {

gateway-js/package.json

@@ -36,12 +36,22 @@
     "@apollo/utils.keyvaluecache": "^2.1.0",
     "@apollo/utils.logger": "^2.0.0",
     "@josephg/resolvable": "^1.0.1",
-    "@opentelemetry/api": "^1.0.1",
+    "@opentelemetry/api": "^1.9.0",
+    "@opentelemetry/core": "^1.30.1",
+    "@opentelemetry/exporter-metrics-otlp-http": "^0.57.2",
+    "@opentelemetry/resources": "^1.30.1",
+    "@opentelemetry/resource-detector-alibaba-cloud": "0.30.1",
+    "@opentelemetry/resource-detector-aws": "1.12.0",
+    "@opentelemetry/resource-detector-azure": "0.6.1",
+    "@opentelemetry/resource-detector-gcp": "0.33.1",
+    "@opentelemetry/sdk-metrics": "^1.30.1",
+    "@opentelemetry/semantic-conventions": "^1.28.0",
     "@types/node-fetch": "^2.6.2",
     "async-retry": "^1.3.3",
     "loglevel": "^1.6.1",
     "make-fetch-happen": "^11.0.0",
     "node-abort-controller": "^3.0.1",
+    "node-cpu-count": "^0.1.1",
     "node-fetch": "^2.6.7"
   },
   "peerDependencies": {

gateway-js/src/__tests__/gateway/opentelemetry.test.ts

@@ -4,12 +4,12 @@ import {
   fixtures,
   spanSerializer,
 } from 'apollo-federation-integration-testsuite';
+import { buildSubgraphSchema } from '@apollo/subgraph';
 import {
   InMemorySpanExporter,
   SimpleSpanProcessor,
-} from '@opentelemetry/tracing';
-import { NodeTracerProvider } from '@opentelemetry/node';
-import { buildSubgraphSchema } from '@apollo/subgraph';
+} from '@opentelemetry/sdk-trace-base';
+import { NodeTracerProvider } from '@opentelemetry/sdk-trace-node';
 
 expect.addSnapshotSerializer(spanSerializer);
 

gateway-js/src/datasources/RemoteGraphQLDataSource.ts

@@ -8,17 +8,22 @@ import { Headers as NodeFetchHeaders, Request as NodeFetchRequest } from 'node-f
 import { Fetcher, FetcherRequestInit, FetcherResponse } from '@apollo/utils.fetcher';
 import { GraphQLError, GraphQLErrorExtensions } from 'graphql';
 import { GatewayCacheHint, GatewayCachePolicy, GatewayGraphQLRequest, GatewayGraphQLRequestContext, GatewayGraphQLResponse } from '@apollo/server-gateway-interface';
+import { MeterProvider } from '@opentelemetry/sdk-metrics';
+import { Counter } from '@opentelemetry/api';
 
 export class RemoteGraphQLDataSource<
   TContext extends Record<string, any> = Record<string, any>,
 > implements GraphQLDataSource<TContext>
 {
   fetcher: Fetcher;
+  fetch_request_size_counter: (Counter | undefined);
+  fetch_response_size_counter: (Counter | undefined);
 
   constructor(
     config?: Partial<RemoteGraphQLDataSource<TContext>> &
       object &
       ThisType<RemoteGraphQLDataSource<TContext>>,
+    meterProvider?: MeterProvider,
   ) {
     this.fetcher = fetcher.defaults({
       // Allow an arbitrary number of sockets per subgraph. This is the default
@@ -31,12 +36,22 @@ export class RemoteGraphQLDataSource<
       // intact.
       retry: false,
     });
+    if (meterProvider) {
+      const meter = meterProvider.getMeter('apollo/gateway');
+      this.fetch_request_size_counter = meter.createCounter("apollo.gateway.operations.fetch.request_size", {
+        unit: 'bytes',
+      });
+      this.fetch_response_size_counter = meter.createCounter("apollo.gateway.operations.fetch.response_size", {
+        unit: 'bytes',
+      });
+    }
     if (config) {
       return Object.assign(this, config);
     }
   }
 
   url!: string;
+  name!: string;
 
   /**
    * Whether the downstream request should be made with automated persisted
@@ -149,7 +164,7 @@ export class RemoteGraphQLDataSource<
           request: requestWithoutQuery,
           context,
           overallCachePolicy,
-          pathInIncomingRequest
+          pathInIncomingRequest,
         });
       }
     }
@@ -167,7 +182,7 @@ export class RemoteGraphQLDataSource<
       request: requestWithQuery,
       context,
       overallCachePolicy,
-      pathInIncomingRequest
+      pathInIncomingRequest,
     });
   }
 
@@ -201,6 +216,9 @@ export class RemoteGraphQLDataSource<
     let fetchResponse: FetcherResponse | undefined;
 
     try {
+      this.fetch_request_size_counter?.add(Buffer.byteLength(stringifiedRequestWithoutHttp), {
+        "subgraph.name": this.name
+      })
       // Use our local `fetcher` to allow for fetch injection
       // Use the fetcher's `Request` implementation for compatibility
       fetchResponse = await this.fetcher(http.url, requestInit);
@@ -215,12 +233,22 @@ export class RemoteGraphQLDataSource<
         throw new Error(`Expected JSON response body, but received: ${body}`);
       }
 
+      this.fetch_response_size_counter?.add(Buffer.byteLength(body.toString()), {
+        "subgraph.name": this.name
+      })
+
       return {
         ...body,
         http: fetchResponse,
       };
     } catch (error) {
-      this.didEncounterError(error, fetchRequest, fetchResponse, context, request);
+      this.didEncounterError(
+        error,
+        fetchRequest,
+        fetchResponse,
+        context,
+        request,
+      );
       throw error;
     }
   }
@@ -234,17 +262,22 @@ export class RemoteGraphQLDataSource<
     request,
     context,
     overallCachePolicy,
-    pathInIncomingRequest
+    pathInIncomingRequest,
   }: {
     response: GatewayGraphQLResponse;
     request: GatewayGraphQLRequest;
     context: TContext;
     overallCachePolicy: GatewayCachePolicy | null;
-    pathInIncomingRequest?: ResponsePath
+    pathInIncomingRequest?: ResponsePath;
   }): Promise<GatewayGraphQLResponse> {
     const processedResponse =
       typeof this.didReceiveResponse === 'function'
-        ? await this.didReceiveResponse({ response, request, context, pathInIncomingRequest })
+        ? await this.didReceiveResponse({
+            response,
+            request,
+            context,
+            pathInIncomingRequest,
+          })
         : response;
 
     if (overallCachePolicy) {
@@ -275,8 +308,11 @@ export class RemoteGraphQLDataSource<
 
   public didReceiveResponse?(
     requestContext: Required<
-      Pick<GatewayGraphQLRequestContext<TContext>, 'request' | 'response' | 'context'>
-    > & { pathInIncomingRequest?: ResponsePath }
+      Pick<
+        GatewayGraphQLRequestContext<TContext>,
+        'request' | 'response' | 'context'
+      >
+    > & { pathInIncomingRequest?: ResponsePath },
   ): GatewayGraphQLResponse | Promise<GatewayGraphQLResponse>;
 
   public didEncounterError(