feat: Transfer Manager Metrics

src/file.ts

@@ -61,6 +61,7 @@ import {
   ApiError,
   Duplexify,
   DuplexifyConstructor,
+  GCCL_GCS_CMD_KEY,
 } from './nodejs-common/util';
 // eslint-disable-next-line @typescript-eslint/no-var-requires
 const duplexify: DuplexifyConstructor = require('duplexify');
@@ -221,6 +222,7 @@ type PublicResumableUploadOptions =
 export interface CreateResumableUploadOptions
   extends Pick<resumableUpload.UploadConfig, PublicResumableUploadOptions> {
   preconditionOpts?: PreconditionOptions;
+  [GCCL_GCS_CMD_KEY]?: resumableUpload.UploadConfig[typeof GCCL_GCS_CMD_KEY];
 }
 
 export type CreateResumableUploadResponse = [string];
@@ -371,6 +373,7 @@ export interface CreateReadStreamOptions {
   start?: number;
   end?: number;
   decompress?: boolean;
+  [GCCL_GCS_CMD_KEY]?: string;
 }
 
 export interface SaveOptions extends CreateWriteStreamOptions {
@@ -1580,12 +1583,16 @@ class File extends ServiceObject<File, FileMetadata> {
         headers.Range = `bytes=${tailRequest ? end : `${start}-${end}`}`;
       }
 
-      const reqOpts = {
+      const reqOpts: DecorateRequestOptions = {
         uri: '',
         headers,
         qs: query,
       };
 
+      if (options[GCCL_GCS_CMD_KEY]) {
+        reqOpts[GCCL_GCS_CMD_KEY] = options[GCCL_GCS_CMD_KEY];
+      }
+
       this.requestStream(reqOpts)
         .on('error', err => {
           throughStream.destroy(err);

src/nodejs-common/service.ts

@@ -21,6 +21,7 @@ import {Interceptor} from './service-object';
 import {
   BodyResponseCallback,
   DecorateRequestOptions,
+  GCCL_GCS_CMD_KEY,
   MakeAuthenticatedRequest,
   PackageJson,
   util,
@@ -253,6 +254,12 @@ export class Service {
       } gccl-invocation-id/${uuid.v4()}`,
     };
 
+    if (reqOpts[GCCL_GCS_CMD_KEY]) {
+      reqOpts.headers[
+        'x-goog-api-client'
+      ] += ` gccl-gcs-cmd/${reqOpts[GCCL_GCS_CMD_KEY]}`;
+    }
+
     if (reqOpts.shouldReturnStream) {
       return this.makeAuthenticatedRequest(reqOpts) as {} as r.Request;
     } else {

src/nodejs-common/util.ts

@@ -36,6 +36,14 @@ import {getRuntimeTrackingString} from '../util';
 
 const packageJson = require('../../../package.json');
 
+/**
+ * A unique symbol for providing a `gccl-gcs-cmd` value
+ * for the `X-Goog-API-Client` header.
+ *
+ * E.g the `V` in `X-Goog-API-Client: gccl-gcs-cmd/V`
+ **/
+export const GCCL_GCS_CMD_KEY = Symbol('GCCL_GCS_CMD');
+
 // eslint-disable-next-line @typescript-eslint/no-var-requires
 const duplexify: DuplexifyConstructor = require('duplexify');
 
@@ -233,6 +241,7 @@ export interface DecorateRequestOptions extends r.CoreOptions {
   interceptors_?: Interceptor[];
   shouldReturnStream?: boolean;
   projectId?: string;
+  [GCCL_GCS_CMD_KEY]?: string;
 }
 
 export interface ParsedHttpResponseBody {
@@ -530,7 +539,9 @@ export class Util {
           body: writeStream,
         },
       ],
-    } as {} as r.OptionsWithUri;
+    } as {} as r.OptionsWithUri & {
+      [GCCL_GCS_CMD_KEY]?: string;
+    };
 
     options.makeAuthenticatedRequest(reqOpts, {
       onAuthenticated(err, authenticatedReqOpts) {
@@ -539,7 +550,9 @@ export class Util {
           return;
         }
 
-        requestDefaults.headers = util._getDefaultHeaders();
+        requestDefaults.headers = util._getDefaultHeaders(
+          reqOpts[GCCL_GCS_CMD_KEY]
+        );
         const request = teenyRequest.defaults(requestDefaults);
         request(authenticatedReqOpts!, (err, resp, body) => {
           util.handleResp(err, resp, body, (err, data) => {
@@ -1014,13 +1027,19 @@ export class Util {
       : [optionsOrCallback as T, cb as C];
   }
 
-  _getDefaultHeaders() {
-    return {
+  _getDefaultHeaders(gcclGcsCmd?: string) {
+    const headers = {
       'User-Agent': util.getUserAgentFromPackageJson(packageJson),
       'x-goog-api-client': `${getRuntimeTrackingString()} gccl/${
         packageJson.version
       } gccl-invocation-id/${uuid.v4()}`,
     };
+
+    if (gcclGcsCmd) {
+      headers['x-goog-api-client'] += ` gccl-gcs-cmd/${gcclGcsCmd}`;
+    }
+
+    return headers;
   }
 }
 

src/resumable-upload.ts

@@ -27,6 +27,7 @@ import retry = require('async-retry');
 import {RetryOptions, PreconditionOptions} from './storage';
 import * as uuid from 'uuid';
 import {getRuntimeTrackingString} from './util';
+import {GCCL_GCS_CMD_KEY} from './nodejs-common/util';
 
 const NOT_FOUND_STATUS_CODE = 404;
 const RESUMABLE_INCOMPLETE_STATUS_CODE = 308;
@@ -193,6 +194,8 @@ export interface UploadConfig extends Pick<WritableOptions, 'highWaterMark'> {
    * Configuration options for retrying retryable errors.
    */
   retryOptions: RetryOptions;
+
+  [GCCL_GCS_CMD_KEY]?: string;
 }
 
 export interface ConfigMetadata {
@@ -274,6 +277,7 @@ export class Upload extends Writable {
   private localWriteCache: Buffer[] = [];
   private localWriteCacheByteLength = 0;
   private upstreamEnded = false;
+  #gcclGcsCmd?: string;
 
   constructor(cfg: UploadConfig) {
     super(cfg);
@@ -347,6 +351,8 @@ export class Upload extends Writable {
       : NaN;
     this.contentLength = isNaN(contentLength) ? '*' : contentLength;
 
+    this.#gcclGcsCmd = cfg[GCCL_GCS_CMD_KEY];
+
     this.once('writing', () => {
       if (this.uri) {
         this.continueUploading();
@@ -585,6 +591,14 @@ export class Upload extends Writable {
       delete metadata.contentType;
     }
 
+    let googAPIClient = `${getRuntimeTrackingString()} gccl/${
+      packageJson.version
+    } gccl-invocation-id/${this.currentInvocationId.uri}`;
+
+    if (this.#gcclGcsCmd) {
+      googAPIClient += ` gccl-gcs-cmd/${this.#gcclGcsCmd}`;
+    }
+
     // Check if headers already exist before creating new ones
     const reqOpts: GaxiosOptions = {
       method: 'POST',
@@ -598,9 +612,7 @@ export class Upload extends Writable {
       ),
       data: metadata,
       headers: {
-        'x-goog-api-client': `${getRuntimeTrackingString()} gccl/${
-          packageJson.version
-        } gccl-invocation-id/${this.currentInvocationId.uri}`,
+        'x-goog-api-client': googAPIClient,
         ...headers,
       },
     };
@@ -766,10 +778,16 @@ export class Upload extends Writable {
       },
     });
 
+    let googAPIClient = `${getRuntimeTrackingString()} gccl/${
+      packageJson.version
+    } gccl-invocation-id/${this.currentInvocationId.chunk}`;
+
+    if (this.#gcclGcsCmd) {
+      googAPIClient += ` gccl-gcs-cmd/${this.#gcclGcsCmd}`;
+    }
+
     const headers: GaxiosOptions['headers'] = {
-      'x-goog-api-client': `${getRuntimeTrackingString()} gccl/${
-        packageJson.version
-      } gccl-invocation-id/${this.currentInvocationId.chunk}`,
+      'x-goog-api-client': googAPIClient,
     };
 
     // If using multiple chunk upload, set appropriate header
@@ -904,15 +922,21 @@ export class Upload extends Writable {
   }
 
   private async getAndSetOffset() {
+    let googAPIClient = `${getRuntimeTrackingString()} gccl/${
+      packageJson.version
+    } gccl-invocation-id/${this.currentInvocationId.offset}`;
+
+    if (this.#gcclGcsCmd) {
+      googAPIClient += ` gccl-gcs-cmd/${this.#gcclGcsCmd}`;
+    }
+
     const opts: GaxiosOptions = {
       method: 'PUT',
       url: this.uri!,
       headers: {
         'Content-Length': 0,
         'Content-Range': 'bytes */*',
-        'x-goog-api-client': `${getRuntimeTrackingString()} gccl/${
-          packageJson.version
-        } gccl-invocation-id/${this.currentInvocationId.offset}`,
+        'x-goog-api-client': googAPIClient,
       },
     };
     try {

src/transfer-manager.ts

@@ -26,6 +26,7 @@ import * as retry from 'async-retry';
 import {ApiError} from './nodejs-common';
 import {GaxiosResponse, Headers} from 'gaxios';
 import {createHash} from 'crypto';
+import {GCCL_GCS_CMD_KEY} from './nodejs-common/util';
 
 /**
  * Default number of concurrently executing promises to use when calling uploadManyFiles.
@@ -64,6 +65,21 @@ const UPLOAD_IN_CHUNKS_DEFAULT_CHUNK_SIZE = 32 * 1024 * 1024;
 const DEFAULT_PARALLEL_CHUNKED_UPLOAD_LIMIT = 2;
 
 const EMPTY_REGEX = '(?:)';
+
+/**
+ * The `gccl-gcs-cmd` value for the `X-Goog-API-Client` header.
+ * Example: `gccl-gcs-cmd/tm.upload_many`
+ *
+ * @see {@link GCCL_GCS_CMD}.
+ * @see {@link GCCL_GCS_CMD_KEY}.
+ */
+const GCCL_GCS_CMD_FEATURE = {
+  UPLOAD_MANY: 'tm.upload_many',
+  DOWNLOAD_MANY: 'tm.download_many',
+  UPLOAD_SHARDED: 'tm.upload_sharded',
+  DOWNLOAD_SHARDED: 'tm.download_sharded',
+};
+
 export interface UploadManyFilesOptions {
   concurrencyLimit?: number;
   skipIfExists?: boolean;
@@ -168,7 +184,9 @@ class XMLMultiPartUploadHelper implements MultiPartUploadHelper {
     this.bucket = bucket;
     this.fileName = fileName;
     // eslint-disable-next-line prettier/prettier
-    this.baseUrl = `https://${bucket.name}.${new URL(this.bucket.storage.apiEndpoint).hostname}/${fileName}`;
+    this.baseUrl = `https://${bucket.name}.${
+      new URL(this.bucket.storage.apiEndpoint).hostname
+    }/${fileName}`;
     this.xmlBuilder = new XMLBuilder({arrayNodeName: 'Part'});
     this.xmlParser = new XMLParser();
     this.partsMap = partsMap || new Map<number, string>();
@@ -189,7 +207,22 @@ class XMLMultiPartUploadHelper implements MultiPartUploadHelper {
     const url = `${this.baseUrl}?uploads`;
     return retry(async bail => {
       try {
+        const headers = await this.authClient.getRequestHeaders();
+
+        for (const [key, value] of Object.entries(headers)) {
+          if (key.toLocaleLowerCase().trim() === 'x-goog-api-client') {
+            // Prepend command feature to value, if not already there
+            if (!value.includes(GCCL_GCS_CMD_FEATURE.UPLOAD_SHARDED)) {
+              headers[
+                key
+              ] = `${value} gccl-gcs-cmd/${GCCL_GCS_CMD_FEATURE.UPLOAD_SHARDED}`;
+            }
+            break;
+          }
+        }
+
         const res = await this.authClient.request({
+          headers,
           method: 'POST',
           url,
         });
@@ -314,6 +347,10 @@ export class TransferManager {
     this.bucket = bucket;
   }
 
+  #setMethodHeader() {
+    // .
+  }
+
   /**
    * @typedef {object} UploadManyFilesOptions
    * @property {number} [concurrencyLimit] The number of concurrently executing promises
@@ -394,6 +431,7 @@ export class TransferManager {
 
       const passThroughOptionsCopy: UploadOptions = {
         ...options.passthroughOptions,
+        [GCCL_GCS_CMD_KEY]: GCCL_GCS_CMD_FEATURE.UPLOAD_MANY,
       };
 
       passThroughOptionsCopy.destination = filePath;
@@ -403,6 +441,7 @@ export class TransferManager {
           passThroughOptionsCopy.destination
         );
       }
+
       promises.push(
         limit(() =>
           this.bucket.upload(filePath, passThroughOptionsCopy as UploadOptions)
@@ -487,6 +526,7 @@ export class TransferManager {
     for (const file of files) {
       const passThroughOptionsCopy = {
         ...options.passthroughOptions,
+        [GCCL_GCS_CMD_KEY]: GCCL_GCS_CMD_FEATURE.DOWNLOAD_MANY,
       };
 
       if (options.prefix) {
@@ -499,6 +539,7 @@ export class TransferManager {
       if (options.stripPrefix) {
         passThroughOptionsCopy.destination = file.name.replace(regex, '');
       }
+
       promises.push(limit(() => file.download(passThroughOptionsCopy)));
     }
 
@@ -569,9 +610,15 @@ export class TransferManager {
       chunkEnd = chunkEnd > size ? size : chunkEnd;
       promises.push(
         limit(() =>
-          file.download({start: chunkStart, end: chunkEnd}).then(resp => {
-            return fileToWrite.write(resp[0], 0, resp[0].length, chunkStart);
-          })
+          file
+            .download({
+              start: chunkStart,
+              end: chunkEnd,
+              [GCCL_GCS_CMD_KEY]: GCCL_GCS_CMD_FEATURE.DOWNLOAD_SHARDED,
+            })
+            .then(resp => {
+              return fileToWrite.write(resp[0], 0, resp[0].length, chunkStart);
+            })
         )
       );