mirror of
https://github.com/openobserve/openobserve.git
synced 2026-11-06 08:09:24 +00:00
366 lines
12 KiB
JavaScript
366 lines
12 KiB
JavaScript
/**
|
|
* Trace ingestion utility for E2E tests
|
|
* Generates and ingests distributed traces for testing purposes
|
|
*/
|
|
|
|
const crypto = require('crypto');
|
|
const testLogger = require('./test-logger.js');
|
|
const { getAuthHeaders, getOrgIdentifier } = require('./cloud-auth.js');
|
|
const { ingestBase } = require('./ingest-url.js');
|
|
|
|
/**
|
|
* Generate a random hex string of specified byte length
|
|
* @param {number} bytes - Number of bytes
|
|
* @returns {string} Hex string
|
|
*/
|
|
function generateHexId(bytes) {
|
|
return crypto.randomBytes(bytes).toString('hex');
|
|
}
|
|
|
|
/**
|
|
* Get current timestamp in nanoseconds
|
|
* @returns {string} Timestamp in nanoseconds
|
|
*/
|
|
function getTimestampNs() {
|
|
return String(Date.now() * 1000000);
|
|
}
|
|
|
|
/**
|
|
* Generate a single distributed trace
|
|
* @param {number} iteration - Trace iteration number
|
|
* @param {Object} [options]
|
|
* @param {'error'|'success'} [options.forceScenario] - Deterministically pick an
|
|
* error (spanStatus=2) or success (spanStatus=1) scenario instead of a random
|
|
* one. Used by tests that must guarantee error traces exist (e.g. the traces
|
|
* error-only toggle), removing the data-dependent flake where a random draw
|
|
* produced no error spans.
|
|
* @param {number} [options.ageMs] - Stamp the trace this many ms before now.
|
|
* @returns {Object} OTLP trace data
|
|
*/
|
|
function generateTrace(iteration, options = {}) {
|
|
// Generate IDs
|
|
const traceId = generateHexId(16);
|
|
const rootSpanId = generateHexId(8);
|
|
const clientSpanId = generateHexId(8);
|
|
const serverSpanId = generateHexId(8);
|
|
|
|
const startTime = String((Date.now() - (options.ageMs || 0)) * 1000000);
|
|
|
|
// Randomly choose scenario
|
|
const scenarios = [
|
|
{
|
|
scenario: "success",
|
|
httpStatus: 200,
|
|
spanStatus: 1, // OK
|
|
errorMessage: "",
|
|
operationName: "get_orders",
|
|
frontendService: "api-gateway",
|
|
backendService: "order-service",
|
|
duration: 150000000 // 150ms
|
|
},
|
|
{
|
|
scenario: "success",
|
|
httpStatus: 200,
|
|
spanStatus: 1,
|
|
errorMessage: "",
|
|
operationName: "process_payment",
|
|
frontendService: "api-gateway",
|
|
backendService: "payment-service",
|
|
duration: 800000000 // 800ms
|
|
},
|
|
{
|
|
scenario: "database_error",
|
|
httpStatus: 500,
|
|
spanStatus: 2, // ERROR
|
|
errorMessage: "Database connection failed",
|
|
operationName: "database_query",
|
|
frontendService: "api-gateway",
|
|
backendService: "user-service",
|
|
duration: 5000000000 // 5 seconds
|
|
},
|
|
{
|
|
scenario: "not_found",
|
|
httpStatus: 404,
|
|
spanStatus: 2,
|
|
errorMessage: "Resource not found",
|
|
operationName: "get_user_profile",
|
|
frontendService: "api-gateway",
|
|
backendService: "profile-service",
|
|
duration: 200000000 // 200ms
|
|
},
|
|
{
|
|
scenario: "auth_error",
|
|
httpStatus: 401,
|
|
spanStatus: 2,
|
|
errorMessage: "Invalid authentication token",
|
|
operationName: "authenticate_user",
|
|
frontendService: "api-gateway",
|
|
backendService: "auth-service",
|
|
duration: 500000000 // 500ms
|
|
}
|
|
];
|
|
|
|
// Scenario selection. `forceScenario` makes the error/success mix deterministic
|
|
// (round-robin within the matching subset) so callers can guarantee at least one
|
|
// error span; without it we keep the original random draw.
|
|
let selected;
|
|
if (options.forceScenario === 'error') {
|
|
const errorScenarios = scenarios.filter((s) => s.spanStatus === 2);
|
|
selected = errorScenarios[iteration % errorScenarios.length];
|
|
} else if (options.forceScenario === 'success') {
|
|
const successScenarios = scenarios.filter((s) => s.spanStatus === 1);
|
|
selected = successScenarios[iteration % successScenarios.length];
|
|
} else {
|
|
selected = scenarios[Math.floor(Math.random() * scenarios.length)];
|
|
}
|
|
|
|
// Calculate timestamps for distributed trace spans
|
|
const clientStart = Number(startTime) + 1000000;
|
|
const serverStart = clientStart + 2000000;
|
|
const serverEnd = serverStart + selected.duration;
|
|
const clientEnd = serverEnd + 1000000;
|
|
const rootEnd = clientEnd + 500000;
|
|
|
|
// Build error events if it's an error scenario
|
|
const errorEvents = selected.spanStatus === 2 ? [
|
|
{
|
|
timeUnixNano: String(serverStart + 1000000),
|
|
name: "exception",
|
|
attributes: [
|
|
{
|
|
key: "exception.type",
|
|
value: { stringValue: selected.scenario }
|
|
},
|
|
{
|
|
key: "exception.message",
|
|
value: { stringValue: selected.errorMessage }
|
|
}
|
|
]
|
|
}
|
|
] : [];
|
|
|
|
// Build status for errors
|
|
const status = selected.spanStatus === 2
|
|
? { code: selected.spanStatus, message: selected.errorMessage }
|
|
: { code: selected.spanStatus };
|
|
|
|
// Create distributed trace payload
|
|
const traceData = {
|
|
resourceSpans: [
|
|
{
|
|
resource: {
|
|
attributes: [
|
|
{ key: "service.name", value: { stringValue: selected.frontendService } },
|
|
{ key: "service.version", value: { stringValue: "1.0.0" } },
|
|
{ key: "environment", value: { stringValue: "test" } },
|
|
{ key: "test.iteration", value: { intValue: iteration } }
|
|
]
|
|
},
|
|
scopeSpans: [
|
|
{
|
|
scope: {
|
|
name: "opentelemetry-instrumentation",
|
|
version: "1.0.0"
|
|
},
|
|
spans: [
|
|
{
|
|
traceId: traceId,
|
|
spanId: rootSpanId,
|
|
name: `HTTP POST /${selected.operationName}`,
|
|
kind: 2, // SERVER
|
|
startTimeUnixNano: String(startTime),
|
|
endTimeUnixNano: String(rootEnd),
|
|
attributes: [
|
|
{ key: "http.method", value: { stringValue: "POST" } },
|
|
{ key: "http.url", value: { stringValue: `https://api.example.com/v1/${selected.operationName}` } },
|
|
{ key: "http.status_code", value: { intValue: selected.httpStatus } },
|
|
{ key: "user.id", value: { stringValue: `user_${Math.floor(Math.random() * 100)}` } },
|
|
{ key: "request.id", value: { stringValue: `req_${generateHexId(8)}` } }
|
|
],
|
|
status: status
|
|
},
|
|
{
|
|
traceId: traceId,
|
|
spanId: clientSpanId,
|
|
parentSpanId: rootSpanId,
|
|
name: `HTTP POST ${selected.backendService}/${selected.operationName}`,
|
|
kind: 3, // CLIENT
|
|
startTimeUnixNano: String(clientStart),
|
|
endTimeUnixNano: String(clientEnd),
|
|
attributes: [
|
|
{ key: "http.method", value: { stringValue: "POST" } },
|
|
{ key: "http.url", value: { stringValue: `http://${selected.backendService}:8080/v1/${selected.operationName}` } },
|
|
{ key: "peer.service", value: { stringValue: selected.backendService } },
|
|
{ key: "http.status_code", value: { intValue: selected.httpStatus } }
|
|
],
|
|
status: status
|
|
}
|
|
]
|
|
}
|
|
]
|
|
},
|
|
{
|
|
resource: {
|
|
attributes: [
|
|
{ key: "service.name", value: { stringValue: selected.backendService } },
|
|
{ key: "service.version", value: { stringValue: "2.0.0" } },
|
|
{ key: "environment", value: { stringValue: "test" } }
|
|
]
|
|
},
|
|
scopeSpans: [
|
|
{
|
|
scope: {
|
|
name: "opentelemetry-instrumentation",
|
|
version: "1.0.0"
|
|
},
|
|
spans: [
|
|
{
|
|
traceId: traceId,
|
|
spanId: serverSpanId,
|
|
parentSpanId: clientSpanId,
|
|
name: `POST /${selected.operationName}`,
|
|
kind: 2, // SERVER
|
|
startTimeUnixNano: String(serverStart),
|
|
endTimeUnixNano: String(serverEnd),
|
|
attributes: [
|
|
{ key: "http.method", value: { stringValue: "POST" } },
|
|
{ key: "http.target", value: { stringValue: `/v1/${selected.operationName}` } },
|
|
{ key: "http.status_code", value: { intValue: selected.httpStatus } },
|
|
{ key: "user.id", value: { stringValue: `user_${Math.floor(Math.random() * 100)}` } }
|
|
],
|
|
events: errorEvents,
|
|
status: status
|
|
}
|
|
]
|
|
}
|
|
]
|
|
}
|
|
]
|
|
};
|
|
|
|
testLogger.debug(`Generated trace ${iteration}`, {
|
|
traceId,
|
|
scenario: selected.scenario,
|
|
services: `${selected.frontendService} -> ${selected.backendService}`,
|
|
status: selected.httpStatus
|
|
});
|
|
|
|
return traceData;
|
|
}
|
|
|
|
/**
|
|
* Ingest traces into OpenObserve
|
|
* @param {import('@playwright/test').Page} page - Playwright page object
|
|
* @param {number} traceCount - Number of traces to generate and ingest
|
|
* @returns {Promise<Object>} Ingestion result
|
|
*/
|
|
async function ingestTraces(page, traceCount = 10, options = {}) {
|
|
testLogger.info(`Starting trace ingestion`, { traceCount, forceScenario: options.forceScenario, ageMs: options.ageMs });
|
|
|
|
const orgId = getOrgIdentifier() || "default";
|
|
const headers = getAuthHeaders();
|
|
const baseUrl = (process.env.ZO_BASE_URL || process.env.INGESTION_URL).replace(/\/$/, '');
|
|
|
|
const results = [];
|
|
const traceIds = [];
|
|
|
|
for (let i = 1; i <= traceCount; i++) {
|
|
const traceData = generateTrace(i, options);
|
|
traceIds.push(traceData.resourceSpans[0].scopeSpans[0].spans[0].traceId);
|
|
|
|
try {
|
|
const response = await page.request.post(`${ingestBase(baseUrl)}/api/${orgId}/v1/traces`, {
|
|
headers,
|
|
data: traceData
|
|
});
|
|
|
|
const status = response.status();
|
|
let data = null;
|
|
try {
|
|
data = await response.json();
|
|
} catch {
|
|
data = { text: await response.text().catch(() => '') };
|
|
}
|
|
|
|
results.push({ status, data });
|
|
|
|
if (status !== 200) {
|
|
testLogger.warn(`Trace ${i} ingestion returned non-200 status`, {
|
|
status,
|
|
response: data
|
|
});
|
|
}
|
|
} catch (error) {
|
|
testLogger.error(`Failed to ingest trace ${i}`, { error: error.message });
|
|
results.push({ status: 'error', error: error.message });
|
|
}
|
|
|
|
// Small delay between traces
|
|
await new Promise(resolve => setTimeout(resolve, 100));
|
|
}
|
|
|
|
const successCount = results.filter(r => r.status === 200).length;
|
|
const failureCount = results.length - successCount;
|
|
|
|
testLogger.info('Trace ingestion completed', {
|
|
total: traceCount,
|
|
successful: successCount,
|
|
failed: failureCount,
|
|
sampleTraceIds: traceIds.slice(0, 5)
|
|
});
|
|
|
|
return {
|
|
total: traceCount,
|
|
successful: successCount,
|
|
failed: failureCount,
|
|
traceIds: traceIds,
|
|
results: results
|
|
};
|
|
}
|
|
|
|
/**
|
|
* Ingest `count` single-span ERROR traces named err-op-000.. (newest first) into `streamName`,
|
|
* each followed by an OK span, so a span_status filter and page boundaries are both checkable.
|
|
* @returns {Promise<{status: number}>}
|
|
*/
|
|
async function ingestSequencedErrorSpans(page, streamName, count) {
|
|
const orgId = getOrgIdentifier() || "default";
|
|
const baseUrl = (process.env.ZO_BASE_URL || process.env.INGESTION_URL).replace(/\/$/, '');
|
|
const now = BigInt(Date.now()) * 1000000n;
|
|
const span = (name, startNs, durationMs, statusCode) => ({
|
|
traceId: generateHexId(16),
|
|
spanId: generateHexId(8),
|
|
name,
|
|
kind: 2,
|
|
startTimeUnixNano: String(startNs),
|
|
endTimeUnixNano: String(startNs + BigInt(durationMs) * 1000000n),
|
|
attributes: [],
|
|
status: { code: statusCode },
|
|
});
|
|
const spans = [];
|
|
for (let i = 0; i < count; i++) {
|
|
const start = now - BigInt(i + 1) * 4000000000n;
|
|
const seq = String(i).padStart(3, '0');
|
|
spans.push(span(`err-op-${seq}`, start, i + 1, 2));
|
|
spans.push(span(`ok-op-${seq}`, start + 2000000000n, 1, 1));
|
|
}
|
|
const response = await page.request.post(`${ingestBase(baseUrl)}/api/${orgId}/v1/traces`, {
|
|
headers: { ...getAuthHeaders(), 'stream-name': streamName },
|
|
data: {
|
|
resourceSpans: [{
|
|
resource: { attributes: [{ key: 'service.name', value: { stringValue: 'e2e-sequenced-errors' } }] },
|
|
scopeSpans: [{ scope: { name: 'e2e' }, spans }],
|
|
}],
|
|
},
|
|
});
|
|
testLogger.info('Sequenced error spans ingested', { streamName, count, status: response.status() });
|
|
return { status: response.status() };
|
|
}
|
|
|
|
module.exports = {
|
|
generateTrace,
|
|
ingestTraces,
|
|
ingestSequencedErrorSpans,
|
|
generateHexId,
|
|
getTimestampNs
|
|
}; |