Improve error handling and messaging for cloud-logging

js/plugins/firebase/jest.config.ts

@@ -36,7 +36,7 @@ const config: Config = {
 
   // A map from regular expressions to paths to transformers
   transform: {
-    '^.+\\.[jt]s$': 'ts-jest',
+    '^.+\\.ts$': 'ts-jest',
   },
 
   // An array of regexp pattern strings that are matched against all source file paths, matched files will skip transformation

js/plugins/google-cloud/jest.config.ts

@@ -0,0 +1,48 @@
+/**
+ * Copyright 2024 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * For a detailed explanation regarding each configuration property, visit:
+ * https://jestjs.io/docs/configuration
+ */
+
+import type { Config } from 'jest';
+
+const config: Config = {
+  // Automatically clear mock calls, instances, contexts and results before every test
+  clearMocks: true,
+
+  // A preset that is used as a base for Jest's configuration
+  preset: 'ts-jest',
+
+  // The glob patterns Jest uses to detect test files
+  testMatch: ['**/tests/**/*_test.ts'],
+
+  // An array of regexp pattern strings that are matched against all test paths, matched tests are skipped
+  testPathIgnorePatterns: ['/node_modules/'],
+
+  // A map from regular expressions to paths to transformers
+  transform: {}, // disabled for ESM
+
+  // An array of regexp pattern strings that are matched against all source file paths, matched files will skip transformation
+  transformIgnorePatterns: ['/node_modules/'],
+
+  moduleNameMapper: {
+    '^(\\.{1,2}/.*)\\.js$': '$1',
+  },
+};
+
+export default config;

js/plugins/google-cloud/package.json

@@ -21,7 +21,7 @@
     "build:clean": "rimraf ./lib",
     "build": "npm-run-all build:clean check compile",
     "build:watch": "tsup-node --watch",
-    "test": "tsx --test --test-concurrency=1 ./tests/*_test.ts"
+    "test": "node --experimental-vm-modules node_modules/jest/bin/jest --runInBand --verbose"
   },
   "repository": {
     "type": "git",
@@ -47,17 +47,19 @@
     "@opentelemetry/sdk-trace-base": "^1.25.0",
     "google-auth-library": "^9.6.3",
     "node-fetch": "^3.3.2",
-    "prettier-plugin-organize-imports": "^3.2.4",
     "winston": "^3.12.0"
   },
   "peerDependencies": {
-    "genkit": "workspace:*",
-    "@genkit-ai/core": "workspace:*"
+    "@genkit-ai/core": "workspace:*",
+    "genkit": "workspace:*"
   },
   "devDependencies": {
+    "@jest/globals": "^29.7.0",
     "@types/node": "^20.11.16",
+    "jest": "^29.7.0",
     "npm-run-all": "^4.1.5",
     "rimraf": "^6.0.1",
+    "ts-jest": "^29.1.2",
     "tsup": "^8.0.2",
     "tsx": "^4.7.0",
     "typescript": "^4.9.0"

js/plugins/google-cloud/src/auth.ts

@@ -14,12 +14,22 @@
  * limitations under the License.
  */
 import { logger } from 'genkit/logging';
-import { GoogleAuth } from 'google-auth-library';
-import { GcpTelemetryConfig } from './types';
+import { auth, GoogleAuth } from 'google-auth-library';
+import { GcpPrincipal, GcpTelemetryConfig } from './types';
 
 /**
- * Allow customers to pass in cloud credentials from environment variables
- * following: https://github.com/googleapis/google-auth-library-nodejs?tab=readme-ov-file#loading-credentials-from-environment-variables
+ * Allows Google Cloud credentials to be to passed in "raw" as an environment
+ * variable. This is helpful in environments where the developer has limited
+ * ability to configure their compute environment, but does have the ablilty to
+ * set environment variables.
+ *
+ * This is different from the GOOGLE_APPLICATION_CREDENTIALS used by ADC, which
+ * represents a path to a credential file on disk. In *most* cases, even for
+ * 3rd party cloud providers, developers *should* attempt to use ADC, which
+ * searches for credential files in standard locations, before using this
+ * method.
+ *
+ * See also: https://github.com/googleapis/google-auth-library-nodejs?tab=readme-ov-file#loading-credentials-from-environment-variables
  */
 export async function credentialsFromEnvironment(): Promise<
   Partial<GcpTelemetryConfig>
@@ -47,3 +57,27 @@ export async function credentialsFromEnvironment(): Promise<
   }
   return options;
 }
+
+/**
+ * Resolve the currently configured principal, either from the Genkit specific
+ * GCLOUD_SERVICE_ACCOUNT_CREDS environment variable, or from ADC.
+ *
+ * Since the Google Cloud Telemetry Exporter will discover credentials on its
+ * own, we don't immediately have access to the current principal. This method
+ * can be handy to get access to the current credential for logging debugging
+ * information or other purposes.
+ **/
+export async function resolveCurrentPrincipal(): Promise<GcpPrincipal> {
+  const envCredentials = await credentialsFromEnvironment();
+  const adcCredentials = await auth.getCredentials();
+
+  // TODO(michaeldoyle): How to look up if the user provided credentials in the
+  // plugin config (i.e. GcpTelemetryOptions)
+  let serviceAccountEmail =
+    envCredentials.credentials?.client_email ?? adcCredentials.client_email;
+
+  return {
+    projectId: envCredentials.projectId,
+    serviceAccountEmail,
+  };
+}

js/plugins/google-cloud/src/gcpLogger.ts

@@ -15,8 +15,10 @@
  */
 
 import { LoggingWinston } from '@google-cloud/logging-winston';
+import { logger } from 'genkit/logging';
 import { Writable } from 'stream';
 import { GcpTelemetryConfig } from './types';
+import { loggingDenied, loggingDeniedHelpText } from './utils';
 
 /**
  * Additional streams for writing log data to. Useful for unit testing.
@@ -54,6 +56,7 @@ export class GcpLogger {
             prefix: 'genkit',
             logName: 'genkit_log',
             credentials: this.config.credentials,
+            defaultCallback: await this.getErrorHandler(),
           })
         : new winston.transports.Console()
     );
@@ -68,6 +71,28 @@ export class GcpLogger {
     });
   }
 
+  private async getErrorHandler(): Promise<(err: Error | null) => void> {
+    // only log the first time
+    let instructionsLogged = false;
+    let helpInstructions = await loggingDeniedHelpText();
+
+    return (err: Error | null) => {
+      // Use the defaultLogger so that logs don't get swallowed by
+      // the open telemetry exporter
+      const defaultLogger = logger.defaultLogger;
+      if (err && loggingDenied(err)) {
+        if (!instructionsLogged) {
+          instructionsLogged = true;
+          defaultLogger.error(
+            `Unable to send logs to Google Cloud: ${err.message}\n\n${helpInstructions}\n`
+          );
+        }
+      } else if (err) {
+        defaultLogger.error(`Unable to send logs to Google Cloud: ${err}`);
+      }
+    };
+  }
+
   private shouldExport(env?: string) {
     return this.config.export;
   }

js/plugins/google-cloud/src/gcpOpenTelemetry.ts

@@ -14,7 +14,11 @@
  * limitations under the License.
  */
 
-import { MetricExporter } from '@google-cloud/opentelemetry-cloud-monitoring-exporter';
+import { logger } from '@genkit-ai/core/logging';
+import {
+  ExporterOptions,
+  MetricExporter,
+} from '@google-cloud/opentelemetry-cloud-monitoring-exporter';
 import { TraceExporter } from '@google-cloud/opentelemetry-cloud-trace-exporter';
 import { GcpDetectorSync } from '@google-cloud/opentelemetry-resource-util';
 import { Span, SpanStatusCode, TraceFlags } from '@opentelemetry/api';
@@ -36,6 +40,7 @@ import {
   InstrumentType,
   PeriodicExportingMetricReader,
   PushMetricExporter,
+  ResourceMetrics,
 } from '@opentelemetry/sdk-metrics';
 import { NodeSDKConfiguration } from '@opentelemetry/sdk-node';
 import {
@@ -52,7 +57,13 @@ import { featuresTelemetry } from './telemetry/feature.js';
 import { generateTelemetry } from './telemetry/generate.js';
 import { pathsTelemetry } from './telemetry/path.js';
 import { GcpTelemetryConfig } from './types';
-import { extractErrorName } from './utils';
+import {
+  extractErrorName,
+  metricsDenied,
+  metricsDeniedHelpText,
+  tracingDenied,
+  tracingDeniedHelpText,
+} from './utils';
 
 let metricExporter: PushMetricExporter;
 let spanProcessor: BatchSpanProcessor;
@@ -88,35 +99,43 @@ export class GcpOpenTelemetry {
     record['logging.googleapis.com/spanId'] ??= spanContext.spanId;
   };
 
-  getConfig(): Partial<NodeSDKConfiguration> {
-    spanProcessor = new BatchSpanProcessor(this.createSpanExporter());
+  async getConfig(): Promise<Partial<NodeSDKConfiguration>> {
+    spanProcessor = new BatchSpanProcessor(await this.createSpanExporter());
     return {
       resource: this.resource,
       spanProcessor: spanProcessor,
       sampler: this.config.sampler,
       instrumentations: this.getInstrumentations(),
-      metricReader: this.createMetricReader(),
+      metricReader: await this.createMetricReader(),
     };
   }
 
-  private createSpanExporter(): SpanExporter {
+  private async createSpanExporter(): Promise<SpanExporter> {
     spanExporter = new AdjustingTraceExporter(
       this.shouldExportTraces()
         ? new TraceExporter({
+            // Creds for non-GCP environments; otherwise credentials will be
+            // automatically detected via ADC
             credentials: this.config.credentials,
           })
         : new InMemorySpanExporter(),
       this.config.exportIO,
-      this.config.projectId
+      this.config.projectId,
+      getErrorHandler(
+        (err) => {
+          return tracingDenied(err);
+        },
+        await tracingDeniedHelpText()
+      )
     );
     return spanExporter;
   }
 
   /**
    * Creates a {MetricReader} for pushing metrics out to GCP via OpenTelemetry.
    */
-  private createMetricReader(): PeriodicExportingMetricReader {
-    metricExporter = this.buildMetricExporter();
+  private async createMetricReader(): Promise<PeriodicExportingMetricReader> {
+    metricExporter = await this.buildMetricExporter();
     return new PeriodicExportingMetricReader({
       exportIntervalMillis: this.config.metricExportIntervalMillis,
       exportTimeoutMillis: this.config.metricExportTimeoutMillis,
@@ -150,15 +169,25 @@ export class GcpOpenTelemetry {
     ];
   }
 
-  private buildMetricExporter(): PushMetricExporter {
+  private async buildMetricExporter(): Promise<PushMetricExporter> {
     const exporter: PushMetricExporter = this.shouldExportMetrics()
-      ? new MetricExporter({
-          userAgent: {
-            product: 'genkit',
-            version: GENKIT_VERSION,
+      ? new MetricExporterWrapper(
+          {
+            userAgent: {
+              product: 'genkit',
+              version: GENKIT_VERSION,
+            },
+            // Creds for non-GCP environments; otherwise credentials will be
+            // automatically detected via ADC
+            credentials: this.config.credentials,
           },
-          credentials: this.config.credentials,
-        })
+          getErrorHandler(
+            (err) => {
+              return metricsDenied(err);
+            },
+            await metricsDeniedHelpText()
+          )
+        )
       : new InMemoryMetricExporter(AggregationTemporality.DELTA);
     exporter.selectAggregation = (instrumentType: InstrumentType) => {
       if (instrumentType === InstrumentType.HISTOGRAM) {
@@ -175,6 +204,31 @@ export class GcpOpenTelemetry {
   }
 }
 
+/**
+ * Rewrites the export method to include an error handler which logs
+ * helpful information about how to set up metrics/telemetry in GCP.
+ */
+class MetricExporterWrapper extends MetricExporter {
+  constructor(
+    private options?: ExporterOptions,
+    private errorHandler?: (error: Error) => void
+  ) {
+    super(options);
+  }
+
+  export(
+    metrics: ResourceMetrics,
+    resultCallback: (result: ExportResult) => void
+  ): void {
+    super.export(metrics, (result) => {
+      if (this.errorHandler && result.error) {
+        this.errorHandler(result.error);
+      }
+      resultCallback(result);
+    });
+  }
+}
+
 /**
  * Adjusts spans before exporting to GCP. Redacts model input
  * and output content, and augments span attributes before sending to GCP.
@@ -183,14 +237,20 @@ class AdjustingTraceExporter implements SpanExporter {
   constructor(
     private exporter: SpanExporter,
     private logIO: boolean,
-    private projectId?: string
+    private projectId?: string,
+    private errorHandler?: (error: Error) => void
   ) {}
 
   export(
     spans: ReadableSpan[],
     resultCallback: (result: ExportResult) => void
   ): void {
-    this.exporter?.export(this.adjust(spans), resultCallback);
+    this.exporter?.export(this.adjust(spans), (result) => {
+      if (this.errorHandler && result.error) {
+        this.errorHandler(result.error);
+      }
+      resultCallback(result);
+    });
   }
 
   shutdown(): Promise<void> {
@@ -362,6 +422,30 @@ class AdjustingTraceExporter implements SpanExporter {
   }
 }
 
+function getErrorHandler(
+  shouldLogFn: (err: Error) => boolean,
+  helpText: string
+): (err: Error) => void {
+  // only log the first time
+  let instructionsLogged = false;
+
+  return (err) => {
+    // Use the defaultLogger so that logs don't get swallowed by the open
+    // telemetry exporter
+    const defaultLogger = logger.defaultLogger;
+    if (err && shouldLogFn(err)) {
+      if (!instructionsLogged) {
+        instructionsLogged = true;
+        defaultLogger.error(
+          `Unable to send telemetry to Google Cloud: ${err.message}\n\n${helpText}\n`
+        );
+      }
+    } else if (err) {
+      defaultLogger.error(`Unable to send telemetry to Google Cloud: ${err}`);
+    }
+  };
+}
+
 export function __getMetricExporterForTesting(): InMemoryMetricExporter {
   return metricExporter as InMemoryMetricExporter;
 }