Eagerly push metrics when pushInterval is 0

CHANGELOG.md

@@ -2,7 +2,8 @@
 
 ## [unreleased]
 
--
+- (_experimental_) Push metrics to gateway "eagerly" when pushInterval is set to 0
+- Log error when fetch is not defined in push context
 
 ## [v0.6.0] - @autometrics/autometrics - 2023-07-20
 

CONTRIBUTING.md

@@ -69,7 +69,7 @@ your example repo automatically (you won't need to re-run `npm install`)
 Open the `express` app with your editor, e.g.: VSCode - `code examples/express/`
 
 This is now your "lab" environment for testing out how the library or the plugin
-work and feel like.
+work and feel.
 
 #### Debugging TypeScript plugin
 

examples/faas-experimental/README.md

@@ -0,0 +1,5 @@
+# Running Example
+
+```ts
+npx tsx index.ts
+```

examples/faas-experimental/fetch-polyfill.js

@@ -0,0 +1,20 @@
+// fetch-polyfill.js
+import fetch, {
+  Blob,
+  blobFrom,
+  blobFromSync,
+  File,
+  fileFrom,
+  fileFromSync,
+  FormData,
+  Headers,
+  Request,
+  Response,
+} from "node-fetch";
+
+if (!globalThis.fetch) {
+  globalThis.fetch = fetch;
+  globalThis.Headers = Headers;
+  globalThis.Request = Request;
+  globalThis.Response = Response;
+}

examples/faas-experimental/index.ts

@@ -0,0 +1,37 @@
+/**
+ * This example should illustrate how you would use autometrics in a FaaS environment.
+ *
+ * If you need an aggregation gateway locally, try:
+ *
+    docker pull ghcr.io/zapier/prom-aggregation-gateway:v0.7.1
+    docker run --platform linux/amd64 \
+      -dit --name aggregation-gateway \
+      -p 9092:80 \
+      ghcr.io/zapier/prom-aggregation-gateway:v0.7.1
+ *
+ */
+
+// NOTE - For now we need a fetch polyfill in node.
+//        (Fetch will already be defined in the browser and in Deno.)
+import "./fetch-polyfill";
+import { init as initAutometrics, autometrics } from "@autometrics/autometrics";
+
+initAutometrics({
+  // NOTE - The current default exporter does not play nicely with Prometheus Push Gateway,
+  //        You'll end up with the error: 
+  //          "pushed metrics are invalid or inconsistent with existing metrics: pushed metrics must not have timestamps"
+  //
+  //        However, everything works fine with aggregation gateways
+  //
+  pushGateway: "http://localhost:9092/metrics",
+  pushInterval: 0,
+});
+
+// Create a getCheese function that returns "gouda"
+const getCheese = autometrics(async function getCheese() {
+  await new Promise((resolve) => setTimeout(resolve, 120));
+  return "gouda";
+});
+
+// This should produce proper metrics
+getCheese();

examples/faas-experimental/package.json

@@ -0,0 +1,22 @@
+{
+  "name": "faas-experimental",
+  "version": "0.0.1",
+  "type": "module",
+  "description": "Example of experimental usage of autometrics-ts in a faas environment",
+  "scripts": {
+    "start": "tsx index.ts"
+  },
+  "author": "",
+  "license": "ISC",
+  "devDependencies": {
+    "@autometrics/typescript-plugin": "file:../../packages/typescript-plugin",
+    "tsx": "^3.12.7",
+    "typescript": "^4.9.4"
+  },
+  "dependencies": {
+    "@opentelemetry/exporter-prometheus": "^0.35.1",
+    "@opentelemetry/sdk-metrics": "^1.9.1",
+    "autometrics": "file:../../packages/autometrics",
+    "node-fetch": "^3.3.1"
+  }
+}

examples/faas-experimental/tsconfig.json

@@ -0,0 +1,19 @@
+{
+  "compilerOptions": {
+    "module": "NodeNext",
+    "experimentalDecorators": true,
+    "moduleResolution": "NodeNext",
+    "esModuleInterop": true,
+    "noImplicitAny": true,
+    "noImplicitThis": true,
+    "removeComments": true,
+    "resolveJsonModule": true,
+    "sourceMap": true,
+    "outDir": "dist",
+    "target": "ESNext",
+    "strictNullChecks": true,
+    "plugins": [{ "name": "@autometrics/typescript-plugin" }]
+  },
+  "include": ["index.ts"],
+  "exclude": ["node_modules", "dist", "**/*.spec.ts"]
+}

packages/lib/src/instrumentation.ts

@@ -4,16 +4,18 @@ import {
 } from "@opentelemetry/exporter-prometheus";
 import {
   AggregationTemporality,
+  ExplicitBucketHistogramAggregation,
   InMemoryMetricExporter,
   MeterProvider,
   MetricReader,
   PeriodicExportingMetricReader,
-  ExplicitBucketHistogramAggregation,
   View,
 } from "@opentelemetry/sdk-metrics";
-import { buildInfo, BuildInfo, recordBuildInfo } from "./buildInfo";
+import { BuildInfo, buildInfo, recordBuildInfo } from "./buildInfo";
 import { HISTOGRAM_NAME } from "./constants";
 
+let globalShouldEagerlyPush = false;
+let pushMetrics = () => {};
 let autometricsMeterProvider: MeterProvider;
 let exporter: MetricReader;
 
@@ -55,6 +57,7 @@ export function init(options: initOptions) {
   exporter = options.exporter;
   // if a pushGateway is added we overwrite the exporter
   if (options.pushGateway) {
+    // INVESTIGATE - Do we want a periodic exporter when we're pushing metrics eagerly?
     exporter = new PeriodicExportingMetricReader({
       // 0 - using delta aggregation temporality setting
       // to ensure data submitted to the gateway is accurate
@@ -63,10 +66,17 @@ export function init(options: initOptions) {
     // Make sure the provider is initialized and exporter is registered
     getMetricsProvider();
 
-    setInterval(
-      () => pushToGateway(options.pushGateway),
-      options.pushInterval ?? 5000,
-    );
+    const interval = options.pushInterval ?? 5000;
+    pushMetrics = () => pushToGateway(options.pushGateway);
+
+    if (interval > 0) {
+      setInterval(pushMetrics, interval);
+    } else if (interval === 0) {
+      logger("Configuring autometrics to push metrics eagerly");
+      globalShouldEagerlyPush = true;
+    } else {
+      console.error("Invalid pushInterval, metrics will not be pushed");
+    }
   }
 
   // buildInfo is added to init function only for client-side applications
@@ -80,19 +90,86 @@ export function init(options: initOptions) {
   }
 }
 
+export function eagerlyPushMetricsIfConfigured() {
+  if (!globalShouldEagerlyPush) {
+    return;
+  }
+
+  if (exporter instanceof PeriodicExportingMetricReader) {
+    logger("Pushing metrics to gateway");
+    pushMetrics();
+  }
+}
+
+// TODO - add a way to stop the push interval
+// TODO - improve error logging
+// TODO - allow custom fetch function to be passed in
+// TODO - allow configuration of timeout for fetch
 async function pushToGateway(gateway: string) {
-  const exporterResponse = await exporter.collect();
+  if (typeof fetch === "undefined") {
+    console.error(
+      "Fetch is undefined, cannot push metrics to gateway. Consider adding a global polyfill.",
+    );
+    return;
+  }
+
+  // Collect metrics
+  // We return early if there was an error
+  const exporterResponse = await safeCollect();
+  if (exporterResponse === null) {
+    return;
+  }
+
   const serialized = new PrometheusSerializer().serialize(
     exporterResponse.resourceMetrics,
   );
 
-  await fetch(gateway, {
-    method: "POST",
-    mode: "cors",
-    body: serialized,
-  });
-  // we flush the metrics at the end of the submission to ensure the data is not repeated
-  await exporter.forceFlush();
+  try {
+    const response = await fetch(gateway, {
+      method: "POST",
+      mode: "cors",
+      body: serialized,
+    });
+    if (!response.ok) {
+      console.error(`Error pushing metrics to gateway: ${response.statusText}`);
+      // NOTE - Uncomment to log the response body
+      // console.error(JSON.stringify(await response.text(), null, 2));
+    }
+  } catch (fetchError) {
+    console.error(
+      `Error pushing metrics to gateway: ${
+        fetchError?.message ?? "<no error message found>"
+      }`,
+    );
+  }
+
+  await safeFlush();
+}
+
+async function safeCollect() {
+  try {
+    return await exporter.collect();
+  } catch (error) {
+    console.error(
+      `Error collecting metrics for push: ${
+        error?.message ?? "<no error message found>"
+      }`,
+    );
+    return null;
+  }
+}
+
+async function safeFlush() {
+  try {
+    // we flush the metrics at the end of the submission to ensure the data is not repeated
+    await exporter.forceFlush();
+  } catch (error) {
+    console.error(
+      `Error flushing metrics after push: ${
+        error?.message ?? "<no error message found>"
+      }`,
+    );
+  }
 }
 
 /**
@@ -120,6 +197,7 @@ export function getMetricsProvider() {
         }),
       ],
     });
+
     autometricsMeterProvider.addMetricReader(exporter);
   }
 

packages/lib/src/wrappers.ts

@@ -8,7 +8,7 @@ import {
   HISTOGRAM_DESCRIPTION,
   HISTOGRAM_NAME,
 } from "./constants";
-import { getMeter } from "./instrumentation";
+import { eagerlyPushMetricsIfConfigured, getMeter } from "./instrumentation";
 import type { Objective } from "./objectives";
 import {
   ALSInstance,
@@ -100,7 +100,6 @@ export type AutometricsOptions<F extends FunctionSig> = {
    * should be considered a success (regardless if it threw an error). This
    * may be most useful when you want to ignore certain errors that are thrown
    * by the function.
-   *
    */
   recordSuccessIf?: ReportSuccessCondition;
 };
@@ -300,6 +299,9 @@ export function autometrics<F extends FunctionSig>(
           module: moduleName,
         });
       }
+
+      // HACK - Experimental "eager pushing" support
+      eagerlyPushMetricsIfConfigured();
     };
 
     const onError = () => {
@@ -327,6 +329,9 @@ export function autometrics<F extends FunctionSig>(
           caller,
         });
       }
+
+      // HACK - Experimental "eager pushing" support
+      eagerlyPushMetricsIfConfigured();
     };
 
     const recordSuccess = (returnValue: Awaited<ReturnType<F>>) => {