Add counters to poller

Adds counters to poller, and 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,155 @@
+/* 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 COUNTERS_PREFIX = 'spanner-autoscaler/';
+const AUTOSCALER_RESOURCE_ATTRIBUTES = {
+  [SemanticResourceAttributes.SERVICE_NAMESPACE]: 'cloud-spanner-autoscaler',
+  [SemanticResourceAttributes.SERVICE_NAME]: 'poller',
+  [SemanticResourceAttributes.SERVICE_VERSION]: '1.0',
+};
+
+const counters = {};
+/** @type {MeterProvider} */
+let meterProvider;
+
+let lastForceFlushTime = 0;
+const MIN_FORCE_FLUSH_INTERVAL=5_000;
+const FORCE_FLUSH_WAIT_MILLIS=1_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();
+    await meterProvider.forceFlush({timeoutMillis: FORCE_FLUSH_WAIT_MILLIS});
+  }
+}
+
+/**
+ * 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,
+};