src/agent/core/meter/MeterSender.ts

/*!
 *
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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
 *
 *     http://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 config from '../../../config/AgentConfig';
import * as grpc from '@grpc/grpc-js';
import { createLogger, throttled } from '../../../logging';
import { MeterReportServiceClient } from '../../../proto/language-agent/Meter_grpc_pb';
import BootService from '../boot/BootService';
import ServiceManager from '../boot/ServiceManager';
import RuntimeMetricsCollector from './RuntimeMetricsCollector';
import { RuntimeSnapshot } from './RuntimeSampler';
import GRPCChannelManager from '../remote/GRPCChannelManager';
import { GRPCChannelListener } from '../remote/GRPCChannelListener';
import { GRPCChannelStatus } from '../remote/GRPCChannelStatus';
import {
  coalesceReport,
  flushCoalesced,
  FLUSH_WAIT_MS,
  ReportCoalesceState,
  runCollectStream,
} from '../remote/coalesceReport';

const logger = createLogger(__filename);
const logReportError = throttled(logger, 'warn', 30000);

/** Reports Node.js runtime metrics via gRPC MeterReportService (Go/Python-compatible pipeline). */
export default class MeterSender implements BootService, GRPCChannelListener {
  private closed = false;
  private channelManager?: GRPCChannelManager;
  private status = GRPCChannelStatus.DISCONNECT;
  private reporterClient?: MeterReportServiceClient;
  /** Latest gauge snapshot only — stale samples have no value after reconnect. */
  private latestSnapshot?: RuntimeSnapshot;
  private timer?: NodeJS.Timeout;
  private readonly reportState: ReportCoalesceState = {};

  private collector!: RuntimeMetricsCollector;

  prepare(): void {
    this.collector = new RuntimeMetricsCollector();
    this.channelManager = ServiceManager.INSTANCE.findService(GRPCChannelManager);
    this.channelManager?.addChannelListener(this);
  }

  boot(): void {
    if (this.timer) {
      logger.warn('MeterSender timer already scheduled; skipping duplicate boot.');
      return;
    }

    this.startTimer();
  }

  onComplete(): void {}

  priority(): number {
    return 0;
  }

  statusChanged(status: GRPCChannelStatus): void {
    this.status = status;
    this.reporterClient = status === GRPCChannelStatus.CONNECTED ? this.createReporterClient() : undefined;
  }

  private createReporterClient(): MeterReportServiceClient | undefined {
    if (!this.channelManager) {
      return undefined;
    }
    return this.channelManager.createClient(MeterReportServiceClient);
  }

  private startTimer(): void {
    this.timer = setInterval(() => {
      if (this.closed) {
        return;
      }
      this.collectSample();
      void this.reportBufferedMetrics();
    }, config.runtimeMetricsReportPeriod || 20000) as NodeJS.Timeout;
    this.timer.unref();
  }

  private collectSample(): void {
    // Always sample so RuntimeSampler CPU deltas stay on the report period (not the outage length).
    this.latestSnapshot = this.collector.sample();
  }

  private reportBufferedMetrics(): Promise<void> {
    return coalesceReport(
      this.reportState,
      () => this.doReportBufferedMetrics(),
      () => this.closed,
    );
  }

  private doReportBufferedMetrics(): Promise<void> {
    if (this.closed) {
      return Promise.resolve();
    }

    if (!config.serviceName || !config.serviceInstance) {
      return Promise.resolve();
    }

    // Check connectivity before consuming the snapshot so disconnect-window samples are kept.
    if (this.status !== GRPCChannelStatus.CONNECTED || !this.reporterClient) {
      return Promise.resolve();
    }

    const snapshot = this.latestSnapshot;
    if (!snapshot) {
      return Promise.resolve();
    }
    this.latestSnapshot = undefined;

    const client = this.reporterClient;
    const serviceName = config.serviceName;
    const serviceInstance = config.serviceInstance;
    return runCollectStream({
      open: (onStatus) =>
        client.collect(new grpc.Metadata(), { deadline: Date.now() + (config.traceTimeout || 10000) }, onStatus),
      writeAll: (stream) => {
        let metadataWritten = false;
        for (const meterData of this.collector.toMeterData(snapshot)) {
          // Meter.proto: service / instance / timestamp on the first stream element only.
          if (!metadataWritten) {
            meterData.setService(serviceName).setServiceinstance(serviceInstance).setTimestamp(snapshot.collectedAt);
            metadataWritten = true;
          }
          stream.write(meterData);
        }
      },
      onFailure: (reason, error) => {
        logReportError(reason, error);
        this.reportGrpcError(error);
      },
      openFailureReason: 'Failed to report runtime meter data',
      endFailureReason: 'Failed to end meter collect stream',
    });
  }

  private reportGrpcError(error: unknown): void {
    if (this.closed) {
      return;
    }

    this.channelManager?.reportError(error);
  }

  /**
   * Best-effort: one shared FLUSH_WAIT_MS budget for forceReport of latest snapshot, then in-flight wait.
   */
  flush(): Promise<any> | null {
    if (this.closed) {
      return null;
    }
    this.collectSample();
    return flushCoalesced(
      this.reportState,
      () => this.doReportBufferedMetrics(),
      () => this.closed,
      () => this.latestSnapshot != null,
      FLUSH_WAIT_MS,
    );
  }

  shutdown(): void {
    this.closed = true;
    if (this.timer) {
      clearInterval(this.timer);
      this.timer = undefined;
    }
    this.reportState.reporting = undefined;
    this.reporterClient = undefined;
    this.latestSnapshot = undefined;
    this.collector.destroy();
    this.channelManager = undefined;
    logger.info('MeterSender destroyed and resources cleaned up');
  }
}