Add metrics support for scaler and poller events

Including required role for metrics writing to scaler
and poller service accounts.

Implements flushing so that if the cloud function is terminated, the counters are sent to monitoring.

src/poller/poller-core/counters.js

@@ -0,0 +1,165 @@
+/* Copyright 2023 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
+ *
+ *     https://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
+ */
+
+/*
+ * Autoscaler Counters package
+ *
+ * Publishes Counters to Cloud Monitoring
+ *
+ */
+const {MeterProvider, PeriodicExportingMetricReader} =
+    require('@opentelemetry/sdk-metrics');
+const {Resource} = require('@opentelemetry/resources');
+const {MetricExporter} =
+    require('@google-cloud/opentelemetry-cloud-monitoring-exporter');
+const {GcpDetectorSync} = require('@google-cloud/opentelemetry-resource-util');
+const {SemanticResourceAttributes} =
+    require('@opentelemetry/semantic-conventions');
+
+const AUTOSCALER_RESOURCE_ATTRIBUTES = {
+  [SemanticResourceAttributes.SERVICE_NAMESPACE]: 'cloud-spanner-autoscaler',
+  [SemanticResourceAttributes.SERVICE_NAME]: 'poller',
+  [SemanticResourceAttributes.SERVICE_VERSION]: '1.0',
+};
+const COUNTERS_PREFIX =
+  'spanner-autoscaler/' +
+   AUTOSCALER_RESOURCE_ATTRIBUTES[SemanticResourceAttributes.SERVICE_NAME] +
+   '/';
+
+const counters = {};
+/** @type {MeterProvider} */
+let meterProvider;
+
+let lastForceFlushTime = 0;
+const MIN_FORCE_FLUSH_INTERVAL=5_000;
+const FORCE_FLUSH_WAIT_MILLIS=10_000;
+
+/**
+ * Initialize metrics with cloud monitoring
+ */
+function initMetrics() {
+  // Create MeterProvider
+  meterProvider = new MeterProvider({
+    resource: new Resource(AUTOSCALER_RESOURCE_ATTRIBUTES)
+        .merge(new GcpDetectorSync().detect()),
+  });
+
+  meterProvider.addMetricReader(new PeriodicExportingMetricReader({
+    // Export metrics every 30 seconds.
+    // Note: 5 seconds is the smallest sample period allowed by
+    // Cloud Monitoring.
+    exportIntervalMillis: 30_000,
+    exporter: new MetricExporter(),
+  }));
+
+  const meter = meterProvider.getMeter(
+      AUTOSCALER_RESOURCE_ATTRIBUTES[SemanticResourceAttributes.SERVICE_NAME]);
+
+  // Create the counters...
+  counters.pollingSuccessCounter = meter.createCounter(
+      COUNTERS_PREFIX + 'polling-success',
+      {description: 'The number of Spanner polling events that succeeded'});
+  counters.pollingFailedCounter = meter.createCounter(
+      COUNTERS_PREFIX + 'polling-failed',
+      {description: 'The number of Spanner polling events that failed'});
+  counters.requestsSuccess = meter.createCounter(
+      COUNTERS_PREFIX + 'requests-success',
+      {
+        description:
+        'The number of polling request messages handled successfully',
+      });
+  counters.requestsFailed = meter.createCounter(
+      COUNTERS_PREFIX + 'requests-failed',
+      {
+        description:
+        'The number of polling request messages that failed',
+      });
+}
+
+/**
+ * Build an attribute object for the counter
+ *
+ * @param {Object} spanner config object
+ * @return {Object}
+ */
+function getCounterAttributes(spanner) {
+  return {
+    'projectId': spanner.projectId,
+    'instanceId': spanner.instanceId,
+  };
+}
+
+/**
+ * Try to flush any as-yet-unsent counters to cloud montioring
+ *
+ * will only actuall call forceFlush once every 5 seconds.
+ */
+async function tryFlush() {
+  if (lastForceFlushTime + MIN_FORCE_FLUSH_INTERVAL < Date.now()) {
+    lastForceFlushTime = Date.now();
+    try {
+      await meterProvider.forceFlush({timeoutMillis: FORCE_FLUSH_WAIT_MILLIS});
+    } catch (e) {
+      // ignore timeouts
+      if (!e instanceof TimeoutError) {
+        throw e;
+      }
+    }
+  }
+}
+
+/**
+ * Increment polling success counter
+ *
+ * @param {Object} spanner config object
+ */
+function incPollingSuccessCounter(spanner) {
+  counters.pollingSuccessCounter.add(
+      1, getCounterAttributes(spanner));
+}
+
+/**
+ * Increment polling failed counter
+ *
+ * @param {Object} spanner config object
+ */
+function incPollingFailedCounter(spanner) {
+  counters.pollingFailedCounter.add(
+      1, getCounterAttributes(spanner));
+}
+
+/**
+ * Increment messages success counter
+ */
+function incRequestsSuccessCounter() {
+  counters.requestsSuccess.add(1);
+}
+
+/**
+ * Increment messages failed counter
+ */
+function incRequestsFailedCounter() {
+  counters.requestsFailed.add(1);
+}
+
+initMetrics();
+
+module.exports = {
+  incPollingSuccessCounter,
+  incPollingFailedCounter,
+  incRequestsSuccessCounter,
+  incRequestsFailedCounter,
+  tryFlush,
+};

src/poller/poller-core/index.js

@@ -25,6 +25,7 @@ const monitoring = require('@google-cloud/monitoring');
 const {PubSub} = require('@google-cloud/pubsub');
 const {Spanner} = require('@google-cloud/spanner');
 const {log} = require('./utils.js');
+const Counters = require('./counters.js');
 
 // GCP service clients
 const metricsClient = new monitoring.MetricServiceClient();
@@ -505,12 +506,14 @@ async function getMetrics(spanner) {
  * @param {List} spanners config objects
  * @return {Promise}
  */
-forwardMetrics = async (forwarderFunction, spanners) => {
+async function forwardMetrics(forwarderFunction, spanners) {
   for (const spanner of spanners) {
     try {
       const metrics = await getMetrics(spanner);
       await forwarderFunction(spanner, metrics); // Handles exceptions
+      Counters.incPollingSuccessCounter(spanner);
     } catch (err) {
+      Counters.incPollingFailedCounter(spanner);
       log(`Unable to retrieve metrics for ${spanner.projectId}/${
         spanner.instanceId}`,
       {
@@ -529,7 +532,7 @@ forwardMetrics = async (forwarderFunction, spanners) => {
  * @param {List} spanners
  * @return {Promise} aggregatedMetrics
  */
-aggregateMetrics = async (spanners) => {
+async function aggregateMetrics(spanners) {
   const aggregatedMetrics = [];
   for (const spanner of spanners) {
     try {
@@ -563,10 +566,14 @@ exports.checkSpannerScaleMetricsPubSub = async (pubSubEvent, context) => {
     log('Autoscaler poller started (PubSub).',
         {severity: 'DEBUG', payload: spanners});
     await forwardMetrics(postPubSubMessage, spanners);
+    Counters.incRequestsSuccessCounter();
   } catch (err) {
     log(`An error occurred in the Autoscaler poller function (PubSub)`,
         {severity: 'ERROR', payload: err});
     log(`JSON payload`, {severity: 'ERROR', payload: payload});
+    Counters.incPollingFailedCounter();
+  } finally {
+    Counters.tryFlush();
   }
 };
 
@@ -590,11 +597,18 @@ exports.checkSpannerScaleMetricsHTTP = async (req, res) => {
     const spanners = await parseAndEnrichPayload(payload);
     await forwardMetrics(postPubSubMessage, spanners);
     res.status(200).end();
+    Counters.incRequestsSuccessCounter();
   } catch (err) {
     log(`An error occurred in the Autoscaler poller function (HTTP)`,
         {severity: 'ERROR', payload: err});
     log(`JSON payload`, {severity: 'ERROR', payload: payload});
-    res.status(500).end(err.toString());
+    res.writeHead(500, {
+      'Content-Type': 'text/plain',
+    });
+    res.end(err.toString());
+    Counters.incPollingFailedCounter();
+  } finally {
+    Counters.tryFlush();
   }
 };
 
@@ -609,10 +623,14 @@ exports.checkSpannerScaleMetricsJSON = async (payload) => {
     log('Autoscaler poller started (JSON/HTTP).',
         {severity: 'DEBUG', payload: spanners});
     await forwardMetrics(callScalerHTTP, spanners);
+    Counters.incRequestsSuccessCounter();
   } catch (err) {
     log(`An error occurred in the Autoscaler poller function (JSON/HTTP)`,
         {severity: 'ERROR', payload: err});
     log(`JSON payload`, {severity: 'ERROR', payload: payload});
+    Counters.incPollingFailedCounter();
+  } finally {
+    Counters.tryFlush();
   }
 };