cloudrun-malware-scanner/metrics.ts

/*
 * Copyright 2021 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.
 */
import * as process from 'node:process';

import {
  MeterProvider,
  PeriodicExportingMetricReader,
} from '@opentelemetry/sdk-metrics';
import {Resource} from '@opentelemetry/resources';
import {MetricExporter as GcpMetricExporter} from '@google-cloud/opentelemetry-cloud-monitoring-exporter';
import {GcpDetectorSync} from '@google-cloud/opentelemetry-resource-util';
import * as Semconv from '@opentelemetry/semantic-conventions';
import {version as packageVersion} from './package.json';

import {logger} from './logger.js';
import * as OpenTelemetryApi from '@opentelemetry/api';

const METRIC_TYPE_ROOT = 'malware-scanning/';

interface CounterAttributes {
  [x: string]: string;
}

const RESOURCE_ATTRIBUTES: CounterAttributes = {
  [Semconv.SEMRESATTRS_SERVICE_NAMESPACE]: 'googlecloudplatform',
  [Semconv.SEMRESATTRS_SERVICE_NAME]: 'gcs-malware-scanning',
  [Semconv.SEMRESATTRS_SERVICE_VERSION]: packageVersion,
};

const COUNTERS_PREFIX =
  RESOURCE_ATTRIBUTES[Semconv.SEMRESATTRS_SERVICE_NAMESPACE] +
  '/' +
  RESOURCE_ATTRIBUTES[Semconv.SEMRESATTRS_SERVICE_NAME] +
  '/';

enum COUNTER_NAMES {
  cleanFiles = 'clean-files',
  infectedFiles = 'infected-files',
  ignoredFiles = 'ignored-files',
  scansFailed = 'scans-failed',
  bytesScanned = 'bytes-scanned',
  scanDuration = 'scan-duration',
  cvdUpdates = 'cvd-mirror-updates',
}

const COUNTER_ATTRIBUTE_NAMES = {
  sourceBucket: 'source_bucket',
  destinationBucket: 'destination_bucket',
  clamVersion: 'clam_version',
  cloudRunRevision: 'cloud_run_revision',
  cvdUpdateStatus: 'cvd_update_status',
  ignoredReason: 'ignored_reason',
  ignoredRegex: 'ignored_regex',
};

interface Counter {
  cumulative?: OpenTelemetryApi.Counter;
  histogram?: OpenTelemetryApi.Histogram;
}

const COUNTERS: Map<COUNTER_NAMES, Counter> = new Map();

const METRIC_EXPORT_INTERVAL = parseInt(
  process.env.EXPORT_INTERVAL || '20000',
  10,
);

class DiagToPinoLogger implements OpenTelemetryApi.DiagLogger {
  suppressErrors: boolean;

  constructor() {
    // In some cases where errors may be expected, we want to be able to supress
    // them.
    this.suppressErrors = false;
  }
  /* eslint-disable @typescript-eslint/no-explicit-any */
  verbose(message: string, ...args: any[]) {
    logger.trace('otel: ' + message, args);
  }

  debug(message: string, ...args: any[]) {
    logger.debug('otel: ' + message, args);
  }

  info(message: string, ...args: any[]) {
    logger.info('otel: ' + message, args);
  }

  warn(message: string, ...args: any[]) {
    logger.warn('otel: ' + message, args);
  }

  error(message: string, ...args: any[]) {
    if (!this.suppressErrors) {
      logger.error('otel: ' + message, args);
    }
  }
  /* eslint-enable */
}

OpenTelemetryApi.default.diag.setLogger(new DiagToPinoLogger(), {
  logLevel: OpenTelemetryApi.DiagLogLevel.INFO,
  suppressOverrideMessage: true,
});

function writeScanFailedMetric(sourceBucket?: string) {
  const attrs: CounterAttributes = {
    [COUNTER_ATTRIBUTE_NAMES.cloudRunRevision]:
      process.env.K_REVISION || 'no-revision',
  };
  if (sourceBucket) {
    attrs[COUNTER_ATTRIBUTE_NAMES.sourceBucket] = sourceBucket;
    attrs[COUNTER_ATTRIBUTE_NAMES.destinationBucket] = sourceBucket;
  }
  COUNTERS.get(COUNTER_NAMES.scansFailed)?.cumulative?.add(1, attrs);
}

function writeScanCleanMetric(
  sourceBucket: string,
  destinationBucket: string,
  fileSize: number,
  scanDuration: number,
  clamVersion: string,
) {
  writeScanCompletedMetric_(
    COUNTER_NAMES.cleanFiles,
    sourceBucket,
    destinationBucket,
    fileSize,
    scanDuration,
    clamVersion,
  );
}

function writeScanIgnoredMetric(
  sourceBucket: string,
  destinationBucket: string,
  fileSize: number,
  ignoredReason: string,
  ignoredRegex?: string,
) {
  const additionalAttrs: CounterAttributes = {};
  if (ignoredReason) {
    additionalAttrs[COUNTER_ATTRIBUTE_NAMES.ignoredReason] = ignoredReason;
  }
  if (ignoredRegex) {
    additionalAttrs[COUNTER_ATTRIBUTE_NAMES.ignoredRegex] = ignoredRegex;
  }
  writeScanCompletedMetric_(
    COUNTER_NAMES.ignoredFiles,
    sourceBucket,
    destinationBucket,
    fileSize,
    0,
    null,
    additionalAttrs,
  );
}

function writeScanInfectedMetric(
  sourceBucket: string,
  destinationBucket: string,
  fileSize: number,
  scanDuration: number,
  clamVersion: string,
) {
  writeScanCompletedMetric_(
    COUNTER_NAMES.infectedFiles,
    sourceBucket,
    destinationBucket,
    fileSize,
    scanDuration,
    clamVersion,
  );
}

function writeScanCompletedMetric_(
  counterName: COUNTER_NAMES,
  sourceBucket: string,
  destinationBucket: string,
  fileSize: number,
  scanDuration: number | null,
  clamVersion: string | null,
  additionalAttrs: CounterAttributes = {},
) {
  const attrs: CounterAttributes = {
    ...additionalAttrs,
    [COUNTER_ATTRIBUTE_NAMES.sourceBucket]: sourceBucket,
    [COUNTER_ATTRIBUTE_NAMES.destinationBucket]: destinationBucket,
    [COUNTER_ATTRIBUTE_NAMES.cloudRunRevision]:
      process.env.K_REVISION || 'no-revision',
  };

  if (clamVersion) {
    attrs[COUNTER_ATTRIBUTE_NAMES.clamVersion] = clamVersion;
  }

  const counter = COUNTERS.get(counterName);
  if (!counter?.cumulative) {
    throw new Error('Unknown counter: ' + counterName);
  }
  counter.cumulative.add(1, attrs);

  COUNTERS.get(COUNTER_NAMES.bytesScanned)?.cumulative?.add(fileSize, attrs);
  if (scanDuration) {
    COUNTERS.get(COUNTER_NAMES.scanDuration)?.histogram?.record(
      scanDuration,
      attrs,
    );
  }
}

function writeCvdMirrorUpdatedMetric(success: boolean, isUpdated: boolean) {
  COUNTERS.get(COUNTER_NAMES.cvdUpdates)?.cumulative?.add(1, {
    [COUNTER_ATTRIBUTE_NAMES.cloudRunRevision]:
      process.env.K_REVISION || 'no-revision',
    [COUNTER_ATTRIBUTE_NAMES.cvdUpdateStatus]: success
      ? isUpdated
        ? 'SUCCESS_UPDATED'
        : 'SUCCESS_NO_UPDATES'
      : 'FAILURE',
  });
}

function initMetrics(projectId: string) {
  if (!projectId) {
    throw Error('Unable to proceed without a Project ID');
  }

  logger.debug('initializing metrics');

  const resources = new GcpDetectorSync()
    .detect()
    .merge(new Resource(RESOURCE_ATTRIBUTES));

  const meterProvider = new MeterProvider({
    resource: resources,
    readers: [
      new PeriodicExportingMetricReader({
        exportIntervalMillis: METRIC_EXPORT_INTERVAL,
        exportTimeoutMillis: METRIC_EXPORT_INTERVAL,
        exporter: new GcpMetricExporter({prefix: 'workload.googleapis.com'}),
      }),
    ],
  });

  const meter = meterProvider.getMeter(COUNTERS_PREFIX);

  // Create cumulative counters
  [
    {
      name: COUNTER_NAMES.cleanFiles,
      description:
        'Number of files scanned that were found to be clean of malware at the time of scan',
    },
    {
      name: COUNTER_NAMES.ignoredFiles,
      description: 'Number of files that were ignored and not scanned',
    },
    {
      name: COUNTER_NAMES.infectedFiles,
      description:
        'Number of files scanned that were found to contain malware at the time of scan',
    },
    {
      name: COUNTER_NAMES.scansFailed,
      description: 'Number of malware scan requests which failed',
    },
    {
      name: COUNTER_NAMES.cvdUpdates,
      description:
        'Number of CVD mirror update checks performed with their status',
    },
    {
      name: COUNTER_NAMES.bytesScanned,
      description: 'Total number of bytes scanned',
      unit: 'By',
    },
  ].forEach((counter) =>
    COUNTERS.set(counter.name, {
      cumulative: meter.createCounter(COUNTERS_PREFIX + counter.name, {
        description: counter.description,
        unit: counter.unit,
      }),
    }),
  );

  COUNTERS.set(COUNTER_NAMES.scanDuration, {
    histogram: meter.createHistogram(
      COUNTERS_PREFIX + COUNTER_NAMES.scanDuration,
      {
        description: 'Duration spent scanning files',
        unit: 'ms',
        advice: {
          explicitBucketBoundaries: [
            0, 50, 100, 200, 500, 1000, 2000, 5000, 10000, 20000, 50000, 100000,
            200000,
          ],
        },
      },
    ),
  });

  // Sanity check on COUNTERS length
  if (COUNTERS.size !== Object.keys(COUNTER_NAMES).length) {
    throw new Error('Code Error: not all counters initialized');
  }

  logger.info(
    `Metrics initialized for ${METRIC_TYPE_ROOT} on project ${projectId}`,
  );
}

export {
  writeScanFailedMetric as writeScanFailed,
  writeScanCleanMetric as writeScanClean,
  writeScanIgnoredMetric as writeScanIgnored,
  writeScanInfectedMetric as writeScanInfected,
  writeCvdMirrorUpdatedMetric as writeCvdMirrorUpdated,
  initMetrics as init,
};