Shared queue consumer telemetry improvements and perf changes #1619
Conversation
|
|
Warning There were issues while running some tools. Please review the errors and either fix the tool’s configuration or disable the tool if it’s a critical failure. 🔧 eslint
apps/webapp/app/v3/tracer.server.tsOops! Something went wrong! :( ESLint: 8.45.0 ESLint couldn't find the config "custom" to extend from. Please check that the name of the config is correct. The config "custom" was referenced from the config file in "/.eslintrc.js". If you still have problems, please stop by https://eslint.org/chat/help to chat with the team. WalkthroughThis pull request introduces several modifications across different components of the application, primarily focusing on OpenTelemetry tracing and queue message handling. The changes consolidate environment configuration for trace exporter authentication, refactor message processing in queue consumers, and adjust logging behavior. The modifications aim to improve code structure, enhance tracing capabilities, and provide more flexible configuration for trace exporters. Changes
Sequence DiagramsequenceDiagram
participant ENV as Environment Config
participant Tracer as OTLPTraceExporter
participant Logger as Structured Logger
ENV->>Tracer: Provide auth headers as JSON string
Tracer->>Tracer: Parse headers
Logger->>Logger: Check span recording status
alt Span not recording
Logger-->>Logger: Suppress debug logs
end
Possibly related PRs
Poem
Finishing Touches
Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media? 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (2)
packages/core/src/logger.ts (1)
112-115: Consider moving the early return check before structuredLog preparation.While the optimization to skip debug logs when spans aren't recording is good, the check occurs after preparing the
structuredLogobject. Moving it before would avoid unnecessary object creation and JSON operations.#structuredLog( loggerFunction: (message: string, ...args: any[]) => void, message: string, level: string, ...args: Array<Record<string, unknown> | undefined> ) { // Get the current context from trace if it exists const currentSpan = trace.getSpan(context.active()); + + // If the span is not recording, and it's a debug log, don't log it + if (currentSpan && !currentSpan.isRecording() && level === "debug") { + return; + } const structuredLog = { ...structureArgs(safeJsonClone(args) as Record<string, unknown>[], this.#filteredKeys), ...this.#additionalFields(), timestamp: new Date(), name: this.#name, message, level, traceId: currentSpan && currentSpan.isRecording() ? currentSpan?.spanContext().traceId : undefined, parentSpanId: currentSpan && currentSpan.isRecording() ? currentSpan?.spanContext().spanId : undefined, }; - // If the span is not recording, and it's a debug log, don't log it - if (currentSpan && !currentSpan.isRecording() && level === "debug") { - return; - } loggerFunction(JSON.stringify(structuredLog, this.#jsonReplacer)); }apps/webapp/app/env.server.ts (1)
193-193: Document the expected format for INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADERS.The consolidation of auth headers into a single property is good, but needs documentation for the expected JSON format.
Add JSDoc comment above the property:
+ /** JSON string containing auth headers. Example: {"Authorization": "Bearer token", "Custom-Header": "value"} */ INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADERS: z.string().optional(),
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (6)
apps/webapp/app/env.server.ts(1 hunks)apps/webapp/app/models/runtimeEnvironment.server.ts(4 hunks)apps/webapp/app/v3/marqs/devQueueConsumer.server.ts(0 hunks)apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts(14 hunks)apps/webapp/app/v3/tracer.server.ts(1 hunks)packages/core/src/logger.ts(1 hunks)
💤 Files with no reviewable changes (1)
- apps/webapp/app/v3/marqs/devQueueConsumer.server.ts
🧰 Additional context used
🪛 Biome (1.9.4)
apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts
[error] 154-154: This variable is used before its declaration.
The variable is declared here:
(lint/correctness/noInvalidUseBeforeDeclaration)
⏰ Context from checks skipped due to timeout of 90000ms (7)
- GitHub Check: e2e / 🧪 CLI v3 tests (windows-latest - pnpm)
- GitHub Check: e2e / 🧪 CLI v3 tests (windows-latest - npm)
- GitHub Check: e2e / 🧪 CLI v3 tests (buildjet-8vcpu-ubuntu-2204 - pnpm)
- GitHub Check: e2e / 🧪 CLI v3 tests (buildjet-8vcpu-ubuntu-2204 - npm)
- GitHub Check: units / 🧪 Unit Tests
- GitHub Check: typecheck / typecheck
- GitHub Check: Analyze (javascript-typescript)
🔇 Additional comments (2)
apps/webapp/app/models/runtimeEnvironment.server.ts (1)
Line range hint
8-15: Reconsider replacingfindUniquewithfindFirstto ensure correct environment retrievalReplacing
findUniquewithfindFirstmay lead to unintended behavior if multiple records match the query criteria, asfindFirstwill return the first matching record, which may not be unique. This could result in incorrect environments being retrieved.Please ensure that the uniqueness of the queried fields is enforced at the database level to prevent duplicate records. If duplicates are possible, consider handling them appropriately or reverting to
findUnique.Also applies to: 28-35, 48-55, 88-95
apps/webapp/app/env.server.ts (1)
193-193: Verify usage of the new consolidated headers property.Need to ensure all existing configurations using the old header properties are migrated correctly.
✅ Verification successful
Migration to consolidated headers property is complete and correct
The codebase shows a clean implementation of the consolidated INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADERS property, with proper JSON parsing and type-safe usage. No traces of the old header properties were found, indicating a successful migration.
🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash # Check for usage of old header properties in configuration files rg -l "INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADER_(NAME|VALUE)" # Check for header parsing logic in tracer configuration ast-grep --pattern 'INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADERS'Length of output: 126
Script:
#!/bin/bash # Broader search for OTEL-related environment variables rg -i "otel.*header" -A 5 -B 5 # Look for OpenTelemetry configuration files fd -e ts -e js -e json . | grep -i "otel\|telemetry" # Search for environment variable schema definitions ast-grep --pattern 'z.string().optional()'Length of output: 35094
Script:
#!/bin/bash # Look for OTEL configuration files fd -e ts -e js -e json . | grep -i "otel" | xargs cat # Search for any header-related configuration in OTEL files rg -i "header" -A 5 -B 5 $(fd -e ts -e js -e json . | grep -i "otel")Length of output: 80976
apps/webapp/app/v3/tracer.server.ts
Outdated
| const headers = env.INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADERS | ||
| ? (JSON.parse(env.INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADERS) as Record<string, string>) | ||
| : undefined; |
There was a problem hiding this comment.
Handle potential JSON parsing errors of environment variable
Parsing env.INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADERS without error handling may cause runtime crashes if the environment variable contains invalid JSON. It's important to catch and handle any parsing errors to prevent application failure.
Apply this diff to add error handling:
const headers = env.INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADERS
- ? (JSON.parse(env.INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADERS) as Record<string, string>)
+ ? (() => {
+ try {
+ return JSON.parse(env.INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADERS) as Record<string, string>;
+ } catch (error) {
+ console.error('Failed to parse INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADERS:', error);
+ return undefined;
+ }
+ })()
: undefined;📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| const headers = env.INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADERS | |
| ? (JSON.parse(env.INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADERS) as Record<string, string>) | |
| : undefined; | |
| const headers = env.INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADERS | |
| ? (() => { | |
| try { | |
| return JSON.parse(env.INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADERS) as Record<string, string>; | |
| } catch (error) { | |
| console.error('Failed to parse INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADERS:', error); | |
| return undefined; | |
| } | |
| })() | |
| : undefined; |
@trigger.dev/build
trigger.dev
@trigger.dev/react-hooks
@trigger.dev/sdk
@trigger.dev/core
@trigger.dev/rsc
commit: |
There was a problem hiding this comment.
Actionable comments posted: 2
♻️ Duplicate comments (1)
apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts (1)
144-163:⚠️ Potential issueFix variable declaration order to prevent potential issues.
The
_currentSpanvariable is used before its declaration in the#endCurrentSpanmethod. Reorder the private properties to ensure variables are declared before use.Apply this diff to fix the variable declaration order:
export class SharedQueueConsumer { + private _currentSpan: Span | undefined; private _backgroundWorkers: Map<string, BackgroundWorkerWithTasks> = new Map(); private _deprecatedWorkers: Map<string, BackgroundWorkerWithTasks> = new Map(); private _enabled = false; private _options: Required<SharedQueueConsumerOptions>; private _perTraceCountdown: number | undefined; private _traceStartedAt: Date | undefined; private _currentSpanContext: Context | undefined; private _reasonStats: Record<string, number> = {}; private _actionStats: Record<string, number> = {}; - private _currentSpan: Span | undefined; private _endSpanInNextIteration = false; private _tasks = sharedQueueTasks; private _id: string; private _connectedAt: Date; private _iterationsCount = 0; private _totalIterationsCount = 0; private _runningDurationInMs = 0; private _currentMessage: MessagePayload | undefined; private _currentMessageData: SharedQueueMessageBody | undefined;🧰 Tools
🪛 Biome (1.9.4)
[error] 156-156: This variable is used before its declaration.
The variable is declared here:
(lint/correctness/noInvalidUseBeforeDeclaration)
🧹 Nitpick comments (4)
apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts (4)
1-10: Consider using the OpenTelemetry SDK convenience package.Instead of importing individual OpenTelemetry components, consider using the
@opentelemetry/sdk-trace-basepackage which provides a more convenient way to access commonly used tracing components.-import { - Context, - ROOT_CONTEXT, - Span, - SpanKind, - SpanOptions, - SpanStatusCode, - context, - trace, -} from "@opentelemetry/api"; +import { Context, Span, SpanKind, SpanOptions, SpanStatusCode } from "@opentelemetry/sdk-trace-base"; +import { ROOT_CONTEXT, context, trace } from "@opentelemetry/api";
144-163: Consider grouping related properties for better maintainability.The class properties could be better organized by grouping them based on their purpose (e.g., telemetry, state management, configuration).
Consider reorganizing the properties into these groups:
- Configuration properties (
_options,_enabled)- Worker management (
_backgroundWorkers,_deprecatedWorkers)- Telemetry properties (
_currentSpan,_reasonStats, etc.)- State management (
_currentMessage,_currentMessageData)- Metrics (
_iterationsCount,_totalIterationsCount, etc.)🧰 Tools
🪛 Biome (1.9.4)
[error] 156-156: This variable is used before its declaration.
The variable is declared here:
(lint/correctness/noInvalidUseBeforeDeclaration)
497-525: Consider implementing a message handler factory pattern.The
#handleMessagemethod uses a switch statement to delegate to specific handlers. Consider implementing a message handler factory pattern to make it more maintainable and extensible.Example implementation:
private messageHandlers = new Map<SharedQueueMessageBody['type'], MessageHandler>([ ['EXECUTE', this.#handleExecuteMessage], ['RESUME', this.#handleResumeMessage], ['RESUME_AFTER_DURATION', this.#handleResumeAfterDurationMessage], ['FAIL', this.#handleFailMessage], ]); async #handleMessage(message: MessagePayload, data: SharedQueueMessageBody): Promise<HandleMessageResult> { return await this.#startActiveSpan("handleMessage()", async (span) => { const handler = this.messageHandlers.get(data.type); if (!handler) { throw new Error(`No handler found for message type: ${data.type}`); } return handler.call(this, message, data); }); }
1395-1427: Add JSDoc documentation for the#startActiveSpanhelper method.The
#startActiveSpanmethod is a crucial helper for telemetry but lacks documentation explaining its purpose and parameters.Add JSDoc documentation:
/** * Creates and manages an active span for tracing operations. * @param name - The name of the span * @param fn - The function to execute within the span * @param options - Optional span configuration * @returns The result of the executed function * @throws Rethrows any error from the executed function after recording it in the span */ async #startActiveSpan<T>( name: string, fn: (span: Span) => Promise<T>, options?: SpanOptions ): Promise<T> {
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts(14 hunks)
🧰 Additional context used
🪛 Biome (1.9.4)
apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts
[error] 156-156: This variable is used before its declaration.
The variable is declared here:
(lint/correctness/noInvalidUseBeforeDeclaration)
⏰ Context from checks skipped due to timeout of 90000ms (7)
- GitHub Check: e2e / 🧪 CLI v3 tests (windows-latest - pnpm)
- GitHub Check: e2e / 🧪 CLI v3 tests (windows-latest - npm)
- GitHub Check: e2e / 🧪 CLI v3 tests (buildjet-8vcpu-ubuntu-2204 - pnpm)
- GitHub Check: units / 🧪 Unit Tests
- GitHub Check: e2e / 🧪 CLI v3 tests (buildjet-8vcpu-ubuntu-2204 - npm)
- GitHub Check: typecheck / typecheck
- GitHub Check: Analyze (javascript-typescript)
ericallam
force-pushed
the
shared-queue-consumer-telemetry-improvements
branch
from
c669b44 to
7bc2ded
Compare
There was a problem hiding this comment.
Actionable comments posted: 0
♻️ Duplicate comments (3)
apps/webapp/app/v3/tracer.server.ts (1)
133-135:⚠️ Potential issueHandle potential JSON parsing errors of environment variable.
Parsing
env.INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADERSwithout error handling may cause runtime crashes if the environment variable contains invalid JSON.Apply this diff to add error handling:
const headers = env.INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADERS - ? (JSON.parse(env.INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADERS) as Record<string, string>) + ? (() => { + try { + return JSON.parse(env.INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADERS) as Record<string, string>; + } catch (error) { + console.error('Failed to parse INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADERS:', error); + return undefined; + } + })() : undefined;apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts (2)
150-163:⚠️ Potential issueFix variable declaration order to prevent use before declaration.
The variable
_currentSpanis used before its declaration in the#endCurrentSpanmethod. This could lead to runtime issues.Apply this diff to fix the variable declaration order:
private _perTraceCountdown: number | undefined; private _traceStartedAt: Date | undefined; private _currentSpanContext: Context | undefined; + private _currentSpan: Span | undefined; private _reasonStats: Record<string, number> = {}; private _actionStats: Record<string, number> = {}; - private _currentSpan: Span | undefined; private _endSpanInNextIteration = false; private _tasks = sharedQueueTasks; private _id: string; private _connectedAt: Date; private _iterationsCount = 0; private _totalIterationsCount = 0; private _runningDurationInMs = 0; private _currentMessage: MessagePayload | undefined; private _currentMessageData: SharedQueueMessageBody | undefined;🧰 Tools
🪛 Biome (1.9.4)
[error] 156-156: This variable is used before its declaration.
The variable is declared here:
(lint/correctness/noInvalidUseBeforeDeclaration)
527-918: 🛠️ Refactor suggestionBreak down the large
#handleExecuteMessagemethod.The method is 391 lines long and handles multiple concerns. Consider breaking it down into smaller, focused methods for better maintainability and testability.
Suggested breakdown:
- Task run validation
- Worker deployment validation
- Task execution
- Error handling
Example refactor:
async #handleExecuteMessage(message: MessagePayload, data: SharedQueueExecuteMessageBody): Promise<HandleMessageResult> { const existingTaskRun = await this.#validateTaskRun(message); if (!existingTaskRun) { return this.#generateNoTaskRunResponse(message); } const deployment = await this.#validateAndGetDeployment(existingTaskRun); if (!deployment) { return this.#generateNoDeploymentResponse(existingTaskRun); } return this.#executeTask(message, data, existingTaskRun, deployment); }
🧹 Nitpick comments (1)
apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts (1)
920-1294: Extract common error handling patterns.The message handling methods share similar error handling and response generation patterns. Consider extracting these into utility methods to reduce duplication and improve maintainability.
Example:
private generateErrorResponse(action: HandleMessageAction, reason: string, attrs?: Record<string, any>, error?: Error | string): HandleMessageResult { return { action, reason, attrs, error, interval: this._options.nextTickInterval, retryInMs: 5_000, }; } private generateCheckpointResponse(eventId: string): HandleMessageResult { return { action: "noop", reason: "restored_checkpoint", attrs: { checkpoint_event_id: eventId, }, }; }
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (6)
apps/webapp/app/env.server.ts(1 hunks)apps/webapp/app/models/runtimeEnvironment.server.ts(4 hunks)apps/webapp/app/v3/marqs/devQueueConsumer.server.ts(0 hunks)apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts(14 hunks)apps/webapp/app/v3/tracer.server.ts(1 hunks)packages/core/src/logger.ts(1 hunks)
💤 Files with no reviewable changes (1)
- apps/webapp/app/v3/marqs/devQueueConsumer.server.ts
🚧 Files skipped from review as they are similar to previous changes (3)
- apps/webapp/app/env.server.ts
- packages/core/src/logger.ts
- apps/webapp/app/models/runtimeEnvironment.server.ts
🧰 Additional context used
🪛 Biome (1.9.4)
apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts
[error] 156-156: This variable is used before its declaration.
The variable is declared here:
(lint/correctness/noInvalidUseBeforeDeclaration)
⏰ Context from checks skipped due to timeout of 90000ms (7)
- GitHub Check: e2e / 🧪 CLI v3 tests (windows-latest - pnpm)
- GitHub Check: e2e / 🧪 CLI v3 tests (windows-latest - npm)
- GitHub Check: e2e / 🧪 CLI v3 tests (buildjet-8vcpu-ubuntu-2204 - pnpm)
- GitHub Check: units / 🧪 Unit Tests
- GitHub Check: typecheck / typecheck
- GitHub Check: e2e / 🧪 CLI v3 tests (buildjet-8vcpu-ubuntu-2204 - npm)
- GitHub Check: Analyze (javascript-typescript)
🔇 Additional comments (2)
apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts (2)
1373-1381: Add retry logic for message acknowledgment operations.The
#ackand#nackmethods don't handle potential failures in the message queue operations. Consider adding retry logic for better reliability.Example implementation:
async #ack(messageId: string, retries = 3): Promise<void> { for (let i = 0; i < retries; i++) { try { await marqs?.acknowledgeMessage(messageId, "Acking and doing more work in SharedQueueConsumer"); return; } catch (error) { if (i === retries - 1) throw error; await new Promise(resolve => setTimeout(resolve, Math.pow(2, i) * 1000)); } } }
1395-1427: Well-implemented span management!The
#startActiveSpanmethod is well-structured with proper error handling, attribute management, and cleanup. Good job on ensuring spans are properly ended in the finally block.
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (2)
apps/webapp/app/v3/tracer.server.ts (2)
207-216: Add error logging for configuration parsing failures.While the error handling is good, silently failing without logging could make it difficult to diagnose configuration issues in production.
Consider adding error logging:
function parseInternalTraceHeaders(): Record<string, string> | undefined { try { return env.INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADERS ? (JSON.parse(env.INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADERS) as Record<string, string>) : undefined; - } catch { + } catch (error) { + console.error('Failed to parse INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADERS:', error); return; } }
208-216: Consider adding header structure validation.Since these headers are used for authentication, it would be beneficial to validate their structure to ensure they meet expected format and contain required fields.
Consider:
- Adding type validation for the parsed JSON
- Checking for required header fields
- Validating header values against expected formats (if applicable)
Example validation approach:
interface TraceHeaders { // Define expected header structure [key: string]: string; } function validateHeaders(headers: unknown): headers is TraceHeaders { if (!headers || typeof headers !== 'object') return false; return Object.entries(headers).every( ([key, value]) => typeof key === 'string' && typeof value === 'string' ); } function parseInternalTraceHeaders(): Record<string, string> | undefined { try { const parsed = env.INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADERS ? JSON.parse(env.INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADERS) : undefined; if (parsed && !validateHeaders(parsed)) { console.error('Invalid header structure in INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADERS'); return; } return parsed; } catch (error) { console.error('Failed to parse INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADERS:', error); return; } }
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (2)
apps/webapp/app/v3/tracer.server.ts(2 hunks)packages/core/src/logger.ts(1 hunks)
🚧 Files skipped from review as they are similar to previous changes (1)
- packages/core/src/logger.ts
🔇 Additional comments (1)
apps/webapp/app/v3/tracer.server.ts (1)
133-138: LGTM! Good improvements to header configuration.The changes effectively consolidate header configuration into a single environment variable and handle parsing failures gracefully by defaulting to an empty object. This is a good defensive programming practice.
Summary by CodeRabbit
Configuration Changes
Tracing and Logging Improvements
Code Refactoring