lambda-telemetry-checker.ts

// Code generated by serverless-ci e2e-shared sync. DO NOT EDIT.
// Source of truth lives in serverless-ci/e2e/shared -- edit there; local changes are overwritten.

/*
 * Unless explicitly stated otherwise all files in this repository are licensed
 * under the Apache License Version 2.0.
 *
 * This product includes software developed at Datadog (https://www.datadoghq.com/).
 * Copyright 2026 Datadog, Inc.
 */

import {client, v2} from '@datadog/datadog-api-client';

import {RUN_ID_TAG_KEY} from './naming';

// Runner-agnostic telemetry poller. Mirrors the datadog-ci reference
// (cloud-run-telemetry-checker.ts): poll spans + logs on a bounded budget, then assert
// *identity* on the matched records, not mere existence.

const POLL_INTERVAL_SECONDS = 15;
const MAX_ATTEMPTS = 20;

// A single ingested span or log, flattened to the fields we assert on -- mirrors the Go
// shared Event: top-level string attributes plus "key:value" tag strings.
interface ParsedEvent {
  attrs: Record<string, string>;
  tags: string[];
}

interface IdentityTag {
  key: string;
  value: string;
}

// Flatten a raw span/log record into attrs + tags. Reserved fields (service/env/version)
// sit at the top of `attributes`; logs nest their structured attributes one level deeper;
// tags arrive as a "key:value" string array.
const parseEvent = (record: unknown): ParsedEvent => {
  const attrs: Record<string, string> = {};
  const tags: string[] = [];
  const attributes = (record as {attributes?: Record<string, unknown>})?.attributes;
  if (attributes && typeof attributes === 'object') {
    for (const key of ['service', 'env', 'version']) {
      const value = attributes[key];
      if (typeof value === 'string' && value !== '') {
        attrs[key] = value;
      }
    }
    const nested = attributes.attributes;
    if (nested && typeof nested === 'object') {
      for (const [key, value] of Object.entries(nested)) {
        if (typeof value === 'string') {
          attrs[key] = value;
        }
      }
    }
    if (Array.isArray(attributes.tags)) {
      for (const tag of attributes.tags) {
        if (typeof tag === 'string') {
          tags.push(tag);
        }
      }
    }
  }

  return {attrs, tags};
};

// Assert key=value as a structured attribute or a "key:value" tag -- identity, not a
// substring match against the serialized blob.
const has = (event: ParsedEvent, key: string, value: string): boolean =>
  event.attrs[key] === value || event.tags.includes(`${key}:${value}`);

const identityLabel = (identity: IdentityTag[]): string => identity.map(({key, value}) => `${key}:${value}`).join(', ');

// Auth failures (bad/missing DATADOG_API_KEY/DATADOG_APP_KEY) are non-retryable: polling
// won't fix credentials, so surface them immediately instead of burning the full budget.
// The datadog-api-client throws ApiException with a numeric `code` carrying the status.
const authErrorCode = (error: unknown): number | undefined => {
  const code = (error as {code?: unknown})?.code;

  return code === 401 || code === 403 ? code : undefined;
};

const waitFor = (seconds: number): Promise<void> => new Promise((resolve) => setTimeout(resolve, seconds * 1000));

const buildConfiguration = (): client.Configuration => {
  const configuration = client.createConfiguration({
    authMethods: {
      apiKeyAuth: process.env.DATADOG_API_KEY ?? process.env.DD_API_KEY,
      appKeyAuth: process.env.DATADOG_APP_KEY ?? process.env.DD_APP_KEY,
    },
    // node-fetch can fail while decoding compressed search responses. Requesting
    // identity encoding keeps polling reliable without changing request payloads.
    httpConfig: {compress: false},
  });
  const site = process.env.DATADOG_SITE ?? process.env.DD_SITE;
  if (site) {
    configuration.setServerVariables({site});
  }

  return configuration;
};

// Poll until at least one returned record carries every identity marker. We filter
// in-process (rather than trusting the query alone) so a stray record that merely
// matches the service filter can't pass for one stamped with the full identity.
const pollUntilIdentity = async (
  label: string,
  query: () => Promise<unknown[]>,
  identity: IdentityTag[],
): Promise<void> => {
  for (let attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) {
    // eslint-disable-next-line no-console
    console.log(`[${label}] attempt ${attempt}/${MAX_ATTEMPTS}`);
    try {
      const results = await query();
      const matching = results.filter((record) => {
        const event = parseEvent(record);

        return identity.every(({key, value}) => has(event, key, value));
      });
      if (matching.length > 0) {
        // eslint-disable-next-line no-console
        console.log(`[${label}] found ${matching.length} record(s) with identity [${identityLabel(identity)}]`);

        return;
      }
    } catch (error) {
      const code = authErrorCode(error);
      if (code !== undefined) {
        throw new Error(
          `[${label}] authentication failed (HTTP ${code}) -- check DATADOG_API_KEY/DATADOG_APP_KEY ` +
            `and DATADOG_SITE; not retrying`,
        );
      }
      // eslint-disable-next-line no-console
      console.error(`[${label}] query error:`, error);
    }

    if (attempt < MAX_ATTEMPTS) {
      await waitFor(POLL_INTERVAL_SECONDS);
    }
  }
  throw new Error(
    `[${label}] timed out after ${MAX_ATTEMPTS} attempts (${MAX_ATTEMPTS * POLL_INTERVAL_SECONDS}s) ` +
      `waiting for telemetry with identity [${identityLabel(identity)}]`,
  );
};

const recentWindow = (): {from: string; to: string} => {
  const now = new Date();
  const from = new Date(now.getTime() - 15 * 60 * 1000);

  return {from: from.toISOString(), to: now.toISOString()};
};

const querySpans = async (configuration: client.Configuration, serviceName: string): Promise<unknown[]> => {
  const api = new v2.SpansApi(configuration);
  const {from, to} = recentWindow();
  const response = await api.listSpans({
    body: {
      data: {
        attributes: {
          filter: {query: `@service:${serviceName}`, from, to},
          page: {limit: 25},
        },
        type: 'search_request',
      },
    },
  });

  return response.data ?? [];
};

const queryLogs = async (configuration: client.Configuration, serviceName: string): Promise<unknown[]> => {
  const api = new v2.LogsApi(configuration);
  const {from, to} = recentWindow();
  const response = await api.listLogs({
    body: {
      filter: {query: `service:${serviceName}`, from, to},
      page: {limit: 25},
    },
  });

  return response.data ?? [];
};

export interface TelemetryIdentity {
  serviceName: string;
  env: string;
  version: string;
  runId: string;
  // Run-id tag key; defaults to the shared convention. Override only if a repo diverges.
  runIdTagKey?: string;
}

export const checkTelemetryFlowing = async ({
  serviceName,
  env,
  version,
  runId,
  runIdTagKey = RUN_ID_TAG_KEY,
}: TelemetryIdentity): Promise<void> => {
  const configuration = buildConfiguration();
  await Promise.all([
    // Traces carry service + env + version + run-id identity.
    pollUntilIdentity('spans', () => querySpans(configuration, serviceName), [
      {key: 'service', value: serviceName},
      {key: 'env', value: env},
      {key: 'version', value: version},
      {key: runIdTagKey, value: runId},
    ]),
    // Logs carry service + env + run-id identity.
    pollUntilIdentity('logs', () => queryLogs(configuration, serviceName), [
      {key: 'service', value: serviceName},
      {key: 'env', value: env},
      {key: runIdTagKey, value: runId},
    ]),
  ]);
};