17497369294

Committed 05 Sep 2025 03:21PM UTC coverage: 81.286% (-0.07%) from 81.353%

Build # 17497369294

Build Type

Pull #521

github

Committed by

web-flow

Commit Message

Merge 544cd62ab into 14176badb

Pull Request Pull Request #521: fix: don't wait for telemetry events MCP-179

Run Details

956 of 1271 branches covered (75.22%)

Branch coverage included in aggregate %.

72 of 91 new or added lines in 6 files covered. (79.12%)

1 existing line in 1 file now uncovered.

4769 of 5772 relevant lines covered (82.62%)

45.27 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

85.71

/src/telemetry/telemetry.ts

import type { Session } from "../common/session.js";
import type { BaseEvent, CommonProperties } from "./types.js";
import type { UserConfig } from "../common/config.js";
import { LogId } from "../common/logger.js";
import type { ApiClient } from "../common/atlas/apiClient.js";
import { MACHINE_METADATA } from "./constants.js";
import { EventCache } from "./eventCache.js";
import { detectContainerEnv } from "../helpers/container.js";
import type { DeviceId } from "../helpers/deviceId.js";
import { EventEmitter } from "events";

type EventResult = {
    success: boolean;
    error?: Error;
};

export interface TelemetryEvents {
    "events-emitted": [];
    "events-send-failed": [];
    "events-skipped": [];
}

export class Telemetry {
    private isBufferingEvents: boolean = true;
    /** Resolves when the setup is complete or a timeout occurs */
    public setupPromise: Promise<[string, boolean]> | undefined;
    public readonly events: EventEmitter<TelemetryEvents> = new EventEmitter();

    private eventCache: EventCache;
    private deviceId: DeviceId;

    private constructor(
        private readonly session: Session,
        private readonly userConfig: UserConfig,
        private readonly commonProperties: CommonProperties,
        { eventCache, deviceId }: { eventCache: EventCache; deviceId: DeviceId }
    ) {
        this.eventCache = eventCache;
        this.deviceId = deviceId;
    }

    static create(
        session: Session,
        userConfig: UserConfig,
        deviceId: DeviceId,
        {
            commonProperties = {},
            eventCache = EventCache.getInstance(),
        }: {
            commonProperties?: Partial<CommonProperties>;
            eventCache?: EventCache;
        } = {}
    ): Telemetry {
        const mergedProperties = {
            ...MACHINE_METADATA,
            ...commonProperties,
        };
        const instance = new Telemetry(session, userConfig, mergedProperties, {
            eventCache,
            deviceId,
        });

        void instance.setup();
        return instance;
    }

    private async setup(): Promise<void> {
        if (!this.isTelemetryEnabled()) {
            this.session.logger.info({
                id: LogId.telemetryEmitFailure,
                context: "telemetry",
                message: "Telemetry is disabled.",
                noRedaction: true,
            });
            return;
        }

        this.setupPromise = Promise.all([this.deviceId.get(), detectContainerEnv()]);
        const [deviceIdValue, containerEnv] = await this.setupPromise;

        this.commonProperties.device_id = deviceIdValue;
        this.commonProperties.is_container_env = containerEnv;

        this.isBufferingEvents = false;
    }

    public async close(): Promise<void> {
        this.isBufferingEvents = false;

        this.session.logger.debug({
            id: LogId.telemetryClose,
            message: `Closing telemetry and flushing ${this.eventCache.size} events`,
            context: "telemetry",
        });

        // Wait up to 5 seconds for events to be sent before closing, but don't throw if it times out
        const flushMaxWaitTime = 5000;
        let flushTimeout: NodeJS.Timeout | undefined;
        await Promise.race([
            new Promise<void>((resolve) => {
                flushTimeout = setTimeout(() => {
                    this.session.logger.debug({
                        id: LogId.telemetryClose,
                        message: `Failed to flush remaining events within ${flushMaxWaitTime}ms timeout`,
                        context: "telemetry",
                    });
                    resolve();
                }, flushMaxWaitTime);
                flushTimeout.unref();
            }),
            this.emit([]),
        ]);

        clearTimeout(flushTimeout);
    }

    /**
     * Emits events through the telemetry pipeline
     * @param events - The events to emit
     */
    public emitEvents(events: BaseEvent[]): void {
        if (!this.isTelemetryEnabled()) {
            this.events.emit("events-skipped");
            return;
        }

        // Don't wait for events to be sent - we should not block regular server
        // operations on telemetry
        void this.emit(events);
    }

    /**
     * Gets the common properties for events
     * @returns Object containing common properties for all events
     */
    public getCommonProperties(): CommonProperties {
        return {
            ...this.commonProperties,
            transport: this.userConfig.transport,
            mcp_client_version: this.session.mcpClient?.version,
            mcp_client_name: this.session.mcpClient?.name,
            session_id: this.session.sessionId,
            config_atlas_auth: this.session.apiClient.hasCredentials() ? "true" : "false",
            config_connection_string: this.userConfig.connectionString ? "true" : "false",
        };
    }

    /**
     * Checks if telemetry is currently enabled
     * This is a method rather than a constant to capture runtime config changes
     *
     * Follows the Console Do Not Track standard (https://consoledonottrack.com/)
     * by respecting the DO_NOT_TRACK environment variable
     */
    public isTelemetryEnabled(): boolean {
        // Check if telemetry is explicitly disabled in config
        if (this.userConfig.telemetry === "disabled") {
            return false;
        }

        const doNotTrack = "DO_NOT_TRACK" in process.env;
        return !doNotTrack;
    }

    /**
     * Attempts to emit events through authenticated and unauthenticated clients
     * Falls back to caching if both attempts fail
     */
    private async emit(events: BaseEvent[]): Promise<void> {
        if (this.isBufferingEvents) {
            this.eventCache.appendEvents(events);
            return;
        }

        try {
            const cachedEvents = this.eventCache.getEvents();
            const allEvents = [...cachedEvents.map((e) => e.event), ...events];

            this.session.logger.debug({
                id: LogId.telemetryEmitStart,
                context: "telemetry",
                message: `Attempting to send ${allEvents.length} events (${cachedEvents.length} cached)`,
            });

            const result = await this.sendEvents(this.session.apiClient, allEvents);
            if (result.success) {
                this.eventCache.removeEvents(cachedEvents.map((e) => e.id));
                this.session.logger.debug({
                    id: LogId.telemetryEmitSuccess,
                    context: "telemetry",
                    message: `Sent ${allEvents.length} events successfully: ${JSON.stringify(allEvents)}`,
                });
                this.events.emit("events-emitted");
                return;
            }

            this.session.logger.debug({
                id: LogId.telemetryEmitFailure,
                context: "telemetry",
                message: `Error sending event to client: ${result.error instanceof Error ? result.error.message : String(result.error)}`,
            });
            this.eventCache.appendEvents(events);
            this.events.emit("events-send-failed");
        } catch (error) {
            this.session.logger.debug({
                id: LogId.telemetryEmitFailure,
                context: "telemetry",
                message: `Error emitting telemetry events: ${error instanceof Error ? error.message : String(error)}`,
                noRedaction: true,
            });
            this.events.emit("events-send-failed");
        }
    }

    /**
     * Attempts to send events through the provided API client
     */
    private async sendEvents(client: ApiClient, events: BaseEvent[]): Promise<EventResult> {
        try {
            await client.sendEvents(
                events.map((event) => ({
                    ...event,
                    properties: { ...this.getCommonProperties(), ...event.properties },
                }))
            );
            return { success: true };
        } catch (error) {
            return {
                success: false,
                error: error instanceof Error ? error : new Error(String(error)),
            };
        }
    }
}

1	import type { Session } from "../common/session.js";
2	import type { BaseEvent, CommonProperties } from "./types.js";
3	import type { UserConfig } from "../common/config.js";
4	import { LogId } from "../common/logger.js";	2✔
5	import type { ApiClient } from "../common/atlas/apiClient.js";
6	import { MACHINE_METADATA } from "./constants.js";	2✔
7	import { EventCache } from "./eventCache.js";	2✔
8	import { detectContainerEnv } from "../helpers/container.js";	2✔
9	import type { DeviceId } from "../helpers/deviceId.js";
10	import { EventEmitter } from "events";	2✔
11
12	type EventResult = {
13	success: boolean;
14	error?: Error;
15	};
16
17	export interface TelemetryEvents {
18	"events-emitted": [];
19	"events-send-failed": [];
20	"events-skipped": [];
21	}
22
23	export class Telemetry {	2✔
24	private isBufferingEvents: boolean = true;	2✔
25	/** Resolves when the setup is complete or a timeout occurs */
26	public setupPromise: Promise<[string, boolean]> \| undefined;
27	public readonly events: EventEmitter<TelemetryEvents> = new EventEmitter();	2✔
28
29	private eventCache: EventCache;
30	private deviceId: DeviceId;
31
32	private constructor(	2✔
33	private readonly session: Session,	78✔
34	private readonly userConfig: UserConfig,	78✔
35	private readonly commonProperties: CommonProperties,	78✔
36	{ eventCache, deviceId }: { eventCache: EventCache; deviceId: DeviceId }	78✔
37	) {	78✔
38	this.eventCache = eventCache;	78✔
39	this.deviceId = deviceId;	78✔
40	}	78✔
41
42	static create(	2✔
43	session: Session,	78✔
44	userConfig: UserConfig,	78✔
45	deviceId: DeviceId,	78✔
46	{	78✔
47	commonProperties = {},	78✔
48	eventCache = EventCache.getInstance(),	78✔
49	}: {	78✔
50	commonProperties?: Partial<CommonProperties>;
51	eventCache?: EventCache;
52	} = {}	78✔
53	): Telemetry {	78✔
54	const mergedProperties = {	78✔
55	...MACHINE_METADATA,	78✔
56	...commonProperties,	78✔
57	};	78✔
58	const instance = new Telemetry(session, userConfig, mergedProperties, {	78✔
59	eventCache,	78✔
60	deviceId,	78✔
61	});	78✔
62
63	void instance.setup();	78✔
64	return instance;	78✔
65	}	78✔
66
67	private async setup(): Promise<void> {	2✔
68	if (!this.isTelemetryEnabled()) {	78✔
69	this.session.logger.info({	60✔
70	id: LogId.telemetryEmitFailure,	60✔
71	context: "telemetry",	60✔
72	message: "Telemetry is disabled.",	60✔
73	noRedaction: true,	60✔
74	});	60✔
75	return;	60✔
76	}	60!
77
78	this.setupPromise = Promise.all([this.deviceId.get(), detectContainerEnv()]);	18✔
79	const [deviceIdValue, containerEnv] = await this.setupPromise;	18✔
80
81	this.commonProperties.device_id = deviceIdValue;	18✔
82	this.commonProperties.is_container_env = containerEnv;	18✔
83
84	this.isBufferingEvents = false;	18✔
85	}	78✔
86
87	public async close(): Promise<void> {	2✔
88	this.isBufferingEvents = false;	61✔
89
90	this.session.logger.debug({	61✔
91	id: LogId.telemetryClose,	61✔
92	message: `Closing telemetry and flushing ${this.eventCache.size} events`,	61✔
93	context: "telemetry",	61✔
94	});	61✔
95
96	// Wait up to 5 seconds for events to be sent before closing, but don't throw if it times out
97	const flushMaxWaitTime = 5000;	61✔
98	let flushTimeout: NodeJS.Timeout \| undefined;	61✔
99	await Promise.race([	61✔
100	new Promise<void>((resolve) => {	61✔
101	flushTimeout = setTimeout(() => {	61✔
NEW 102	this.session.logger.debug({	×
NEW 103	id: LogId.telemetryClose,	×
NEW 104	message: `Failed to flush remaining events within ${flushMaxWaitTime}ms timeout`,	×
NEW 105	context: "telemetry",	×
NEW 106	});	×
NEW 107	resolve();	×
108	}, flushMaxWaitTime);	61✔
109	flushTimeout.unref();	61✔
110	}),	61✔
111	this.emit([]),	61✔
112	]);	61✔
113
114	clearTimeout(flushTimeout);	61✔
115	}	61✔
116
117	/**
118	* Emits events through the telemetry pipeline
119	* @param events - The events to emit
120	*/
121	public emitEvents(events: BaseEvent[]): void {	2✔
122	if (!this.isTelemetryEnabled()) {	128✔
123	this.events.emit("events-skipped");	120✔
124	return;	120✔
125	}	120!
126
127	// Don't wait for events to be sent - we should not block regular server
128	// operations on telemetry
129	void this.emit(events);	8✔
130	}	128✔
131
132	/**
133	* Gets the common properties for events
134	* @returns Object containing common properties for all events
135	*/
136	public getCommonProperties(): CommonProperties {	2✔
137	return {	23✔
138	...this.commonProperties,	23✔
139	transport: this.userConfig.transport,	23✔
140	mcp_client_version: this.session.mcpClient?.version,	23✔
141	mcp_client_name: this.session.mcpClient?.name,	23✔
142	session_id: this.session.sessionId,	23✔
143	config_atlas_auth: this.session.apiClient.hasCredentials() ? "true" : "false",	23✔
144	config_connection_string: this.userConfig.connectionString ? "true" : "false",	23!
145	};	23✔
146	}	23✔
147
148	/**
149	* Checks if telemetry is currently enabled
150	* This is a method rather than a constant to capture runtime config changes
151	*
152	* Follows the Console Do Not Track standard (https://consoledonottrack.com/)
153	* by respecting the DO_NOT_TRACK environment variable
154	*/
155	public isTelemetryEnabled(): boolean {	2✔
156	// Check if telemetry is explicitly disabled in config
157	if (this.userConfig.telemetry === "disabled") {	554✔
158	return false;	527✔
159	}	527!
160
161	const doNotTrack = "DO_NOT_TRACK" in process.env;	27✔
162	return !doNotTrack;	27✔
163	}	554✔
164
165	/**
166	* Attempts to emit events through authenticated and unauthenticated clients
167	* Falls back to caching if both attempts fail
168	*/
169	private async emit(events: BaseEvent[]): Promise<void> {	2✔
170	if (this.isBufferingEvents) {	69!
171	this.eventCache.appendEvents(events);	×
172	return;	×
173	}	×
174
175	try {	69✔
176	const cachedEvents = this.eventCache.getEvents();	69✔
177	const allEvents = [...cachedEvents.map((e) => e.event), ...events];	69✔
178
179	this.session.logger.debug({	69✔
180	id: LogId.telemetryEmitStart,	69✔
181	context: "telemetry",	69✔
182	message: `Attempting to send ${allEvents.length} events (${cachedEvents.length} cached)`,	69✔
183	});	69✔
184
185	const result = await this.sendEvents(this.session.apiClient, allEvents);	69✔
186	if (result.success) {	69✔
187	this.eventCache.removeEvents(cachedEvents.map((e) => e.id));	68✔
188	this.session.logger.debug({	68✔
189	id: LogId.telemetryEmitSuccess,	68✔
190	context: "telemetry",	68✔
191	message: `Sent ${allEvents.length} events successfully: ${JSON.stringify(allEvents)}`,	68✔
192	});	68✔
193	this.events.emit("events-emitted");	68✔
194	return;	68✔
195	}	68!
196
197	this.session.logger.debug({	1✔
198	id: LogId.telemetryEmitFailure,	1✔
199	context: "telemetry",	1✔
200	message: `Error sending event to client: ${result.error instanceof Error ? result.error.message : String(result.error)}`,	69!
201	});	69✔
202	this.eventCache.appendEvents(events);	69✔
203	this.events.emit("events-send-failed");	69✔
204	} catch (error) {	69!
NEW 205	this.session.logger.debug({	×
NEW 206	id: LogId.telemetryEmitFailure,	×
NEW 207	context: "telemetry",	×
NEW 208	message: `Error emitting telemetry events: ${error instanceof Error ? error.message : String(error)}`,	×
NEW 209	noRedaction: true,	×
NEW 210	});	×
NEW 211	this.events.emit("events-send-failed");	×
UNCOV 212	}	×
213	}	69✔
214
215	/**
216	* Attempts to send events through the provided API client
217	*/
218	private async sendEvents(client: ApiClient, events: BaseEvent[]): Promise<EventResult> {	2✔
219	try {	69✔
220	await client.sendEvents(	69✔
221	events.map((event) => ({	69✔
222	...event,	9✔
223	properties: { ...this.getCommonProperties(), ...event.properties },	9✔
224	}))	69✔
225	);	69✔
226	return { success: true };	68✔
227	} catch (error) {	69!
228	return {	1✔
229	success: false,	1✔
230	error: error instanceof Error ? error : new Error(String(error)),	1!
231	};	1✔
232	}	1✔
233	}	69✔
234	}	2✔

mongodb-js / mongodb-mcp-server / 17497369294

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous