From 60a8f384f61e5671c4284cdedf1cadf99108cc07 Mon Sep 17 00:00:00 2001 From: Daniel Truong Date: Sun, 6 Sep 2026 07:22:19 +0000 Subject: [PATCH 1/2] fix(ingest): log the real cause when a DCR upload fails AggregateLogsUploadError from @azure/monitor-ingestion carries the message "undefined\n}"; the RestError sits in errors[].cause. Every drop and flush log line printed the placeholder, so a 403 on the rule looked like a blank failure. Unwrap the causes (status code and service message, no row payloads) and stop retrying when every cause is 401 or 403. MIGRATION.md gains the Monitoring Metrics Publisher grant the history import needs: Owner does not cover data actions on the rule. --- docs/MIGRATION.md | 16 ++++++ scripts/import-penguin-history.js | 4 +- src/ingest/dcr-writer.js | 62 +++++++++++++++++++++-- test/dcr-writer.test.js | 82 ++++++++++++++++++++++++++++++- 4 files changed, 157 insertions(+), 7 deletions(-) diff --git a/docs/MIGRATION.md b/docs/MIGRATION.md index 7e9e847..266137b 100644 --- a/docs/MIGRATION.md +++ b/docs/MIGRATION.md @@ -77,6 +77,22 @@ az storage blob upload --auth-mode login --account-name \ --container-name history --name 2026-09/daily_events_summary.csv --file daily_events_summary.csv ``` +The import posts from your own login, not the Function's identity, so that login needs +`Monitoring Metrics Publisher` on `analytics-dcr-`. Owner does not cover it: sending data to a +data collection rule is a data action, and the template grants the role only to the principals in +`publisherPrincipalIds`. Grant it to yourself, once per environment: + +```bash +DCR=$(az resource show -g -n analytics-dcr-test \ + --resource-type Microsoft.Insights/dataCollectionRules --query id -o tsv) +az role assignment create --role "Monitoring Metrics Publisher" \ + --assignee-object-id "$(az ad signed-in-user show --query id -o tsv)" \ + --assignee-principal-type User --scope "$DCR" +``` + +Give it a few minutes to propagate. Until it does, the upload fails with 403 and the script prints +the status code and the service's message on one `[import]` line. + Then import. Check the counts first, post second: ```bash diff --git a/scripts/import-penguin-history.js b/scripts/import-penguin-history.js index 81dc23f..7db3597 100755 --- a/scripts/import-penguin-history.js +++ b/scripts/import-penguin-history.js @@ -17,7 +17,7 @@ const fs = require('fs'); -const { DAILY_STREAM } = require('../src/ingest/dcr-writer'); +const { DAILY_STREAM, describeUploadError } = require('../src/ingest/dcr-writer'); const { logger } = require('../src/utils/logger'); const ENVIRONMENTS = ['dev', 'test', 'prod']; @@ -265,7 +265,7 @@ if (require.main === module) { main(process.argv.slice(2)) .then((code) => { process.exitCode = code; }) .catch((err) => { - logger.error(`[import] ${err.message}`); + logger.error(`[import] ${describeUploadError(err)}`); process.exitCode = 1; }); } diff --git a/src/ingest/dcr-writer.js b/src/ingest/dcr-writer.js index 41931f1..e862772 100644 --- a/src/ingest/dcr-writer.js +++ b/src/ingest/dcr-writer.js @@ -62,13 +62,13 @@ function enqueue(stream, row) { // .catch on both call sites: batches() stringifies outside sendWithRetry's try, so a row that // cannot be serialised rejects flush() with nothing attached, which is an unhandled rejection. if (buffer.length >= config.maxBatch) { - flush().catch((err) => logger.error(`[analytics] flush failed: ${err.message}`)); + flush().catch((err) => logger.error(`[analytics] flush failed: ${describeUploadError(err)}`)); return; } if (!flushTimer) { flushTimer = setTimeout(() => { - flush().catch((err) => logger.error(`[analytics] flush failed: ${err.message}`)); + flush().catch((err) => logger.error(`[analytics] flush failed: ${describeUploadError(err)}`)); }, config.flushMs); // Never hold the worker open for a pending flush; the shutdown hook in index.js drains instead. if (flushTimer.unref) flushTimer.unref(); @@ -142,6 +142,58 @@ async function flush() { while (inFlight.size > 0) await Promise.all([...inFlight]); } +// How many causes a drop line reports. Every row in a batch goes to one DCR, so the causes repeat; +// three distinct ones is already enough to tell a bad role assignment from a throttled workspace. +const MAX_REPORTED_CAUSES = 3; + +/** + * @azure/monitor-ingestion rejects with an AggregateLogsUploadError whose own `.message` is the + * literal `undefined\n}` — its constructor interpolates an argument the SDK never passes. The + * status code and text are on `errors[i].cause`, a RestError. Read the predicate lazily, like the + * rest of the Azure SDK in this file, and never let the lookup itself throw: this runs from a catch. + */ +let aggregatePredicate; +function isAggregateUploadError(err) { + if (!err) return false; + if (aggregatePredicate === undefined) { + try { + aggregatePredicate = require('@azure/monitor-ingestion').isAggregateLogsUploadError; + } catch (_err) { + aggregatePredicate = null; + } + } + return Boolean(aggregatePredicate && aggregatePredicate(err)) || Array.isArray(err.errors); +} + +function causeStatusCodes(err) { + if (isAggregateUploadError(err)) return err.errors.map((entry) => entry.cause && entry.cause.statusCode); + return [err && err.statusCode]; +} + +/** A short cause for a log line. Never the failed rows: callers log this into the application log. */ +function describeUploadError(err) { + if (!isAggregateUploadError(err)) return err && err.message; + + const causes = []; + for (const { cause } of err.errors) { + const status = cause && cause.statusCode ? `${cause.statusCode} ` : ''; + const text = `${status}${(cause && cause.message) || 'no cause reported'}`; + if (!causes.includes(text)) causes.push(text); + if (causes.length === MAX_REPORTED_CAUSES) break; + } + return causes.join('; '); +} + +/** + * 401 and 403 mean the identity is not a Monitoring Metrics Publisher on the DCR (Owner does not + * carry that data action). Waiting does not grant a role, so two more calls only cost latency on + * the shutdown path and bury the one cause under three identical log lines. + */ +function isPermissionFailure(err) { + const codes = causeStatusCodes(err); + return codes.length > 0 && codes.every((code) => code === 401 || code === 403); +} + async function sendWithRetry(stream, batch) { for (let attempt = 0; attempt < 3; attempt++) { try { @@ -149,12 +201,13 @@ async function sendWithRetry(stream, batch) { logger.debug(`[analytics] sent ${batch.length} row(s) to ${stream}`); return; } catch (err) { - if (attempt === 2) { + if (attempt === 2 || isPermissionFailure(err)) { + const tries = attempt + 1; // Count and cause only, never the rows: SourceIp is masked by the DCR transform, and the // application-log workspace is read by more people than the analytics one. // analytics-drop- alerts on this string. logger.error( - `[analytics] dropped ${batch.length} row(s) for ${stream} after 3 attempts: ${err.message}` + `[analytics] dropped ${batch.length} row(s) for ${stream} after ${tries} attempt${tries === 1 ? '' : 's'}: ${describeUploadError(err)}` ); return; } @@ -166,6 +219,7 @@ async function sendWithRetry(stream, batch) { module.exports = { enqueue, flush, + describeUploadError, EVENTS_STREAM, AUDIT_STREAM, DAILY_STREAM, diff --git a/test/dcr-writer.test.js b/test/dcr-writer.test.js index 4aa6d77..09d1fc5 100644 --- a/test/dcr-writer.test.js +++ b/test/dcr-writer.test.js @@ -8,6 +8,8 @@ process.env.ANALYTICS_MAX_BATCH = '3'; const assert = require('node:assert'); const { test } = require('node:test'); +const { AggregateLogsUploadError } = require('@azure/monitor-ingestion'); + const writer = require('../src/ingest/dcr-writer'); const { logger } = require('../src/utils/logger'); @@ -17,6 +19,12 @@ function row(name) { return { TimeGenerated: new Date().toISOString(), EventName: name, Env: 'test' }; } +// The real SDK error, not a stand-in: its own .message is the literal "undefined\n}", which is the +// whole reason describeUploadError exists. Causes are RestError-shaped, as the SDK builds them. +function uploadError(...causes) { + return new AggregateLogsUploadError(causes.map((cause) => ({ cause, failedLogs: [row('page_view')] }))); +} + test('flush sends each buffer to the stream it was enqueued on', async (t) => { const sent = []; writer._setTransport(async (stream, rows) => { sent.push({ stream, rows }); }); @@ -63,7 +71,7 @@ test('a batch over the byte ceiling is split, keeping every row', () => { test('a batch the transport keeps rejecting is dropped after three attempts, and says so', async (t) => { let attempts = 0; - writer._setTransport(async () => { attempts += 1; throw new Error('403 forbidden'); }); + writer._setTransport(async () => { attempts += 1; throw new Error('socket hang up'); }); t.after(() => writer._resetTransport()); const errors = []; t.mock.method(logger, 'error', (message) => errors.push(message)); @@ -103,3 +111,75 @@ test('a flush waits for a send an earlier flush already started', async (t) => { await shutdown; assert.deepStrictEqual(finished, [3]); }); + +test('a drop line carries the cause the SDK buries, not its own placeholder message', async (t) => { + writer._setTransport(async () => { + throw uploadError({ statusCode: 403, message: 'Operation returned an invalid status code Forbidden' }); + }); + t.after(() => writer._resetTransport()); + const errors = []; + t.mock.method(logger, 'error', (message) => errors.push(message)); + + writer.enqueue(EVENTS_STREAM, row('page_view')); + await writer.flush(); + + // AggregateLogsUploadError.message is `undefined\n}`; an operator needs the status and the text. + assert.match(errors[0], /^\[analytics\] dropped 1 row\(s\) for Custom-EagleEvents_CL/); + assert.match(errors[0], /403 Operation returned an invalid status code Forbidden/); + assert.doesNotMatch(errors[0], /undefined/); +}); + +test('a 403 is dropped on the first attempt: a missing role assignment is not transient', async (t) => { + let attempts = 0; + writer._setTransport(async () => { + attempts += 1; + throw uploadError({ statusCode: 403, message: 'Forbidden' }); + }); + t.after(() => writer._resetTransport()); + t.mock.method(logger, 'error', () => {}); + + writer.enqueue(EVENTS_STREAM, row('page_view')); + await writer.flush(); + + assert.strictEqual(attempts, 1); +}); + +test('a 503 keeps all three attempts', async (t) => { + let attempts = 0; + writer._setTransport(async () => { + attempts += 1; + throw uploadError({ statusCode: 503, message: 'Service Unavailable' }); + }); + t.after(() => writer._resetTransport()); + t.mock.method(logger, 'error', () => {}); + + writer.enqueue(EVENTS_STREAM, row('page_view')); + await writer.flush(); + + assert.strictEqual(attempts, 3); +}); + +test('describeUploadError returns a plain error by its own message', () => { + assert.strictEqual(writer.describeUploadError(new Error('socket hang up')), 'socket hang up'); +}); + +test('describeUploadError collapses causes that repeat across a batch', () => { + const described = writer.describeUploadError(uploadError( + { statusCode: 403, message: 'Forbidden' }, + { statusCode: 403, message: 'Forbidden' }, + { statusCode: 429, message: 'Too many requests' } + )); + + assert.strictEqual(described, '403 Forbidden; 429 Too many requests'); +}); + +test('describeUploadError reports at most three distinct causes', () => { + const described = writer.describeUploadError(uploadError( + { statusCode: 400, message: 'one' }, + { statusCode: 400, message: 'two' }, + { statusCode: 400, message: 'three' }, + { statusCode: 400, message: 'four' } + )); + + assert.strictEqual(described, '400 one; 400 two; 400 three'); +}); From 651252aac5622a329acba3cd00e43b9fddc74de8 Mon Sep 17 00:00:00 2001 From: Daniel Truong Date: Sun, 6 Sep 2026 07:30:19 +0000 Subject: [PATCH 2/2] test(ingest): pin permission-failure rule and drop-line tail Mixed 403/503 batches keep three attempts, 401 drops on the first, and the drop line's attempt count is asserted in full. The aggregate error check is Array.isArray(err.errors) alone: the SDK predicate also matches name-only objects with no errors array. --- src/ingest/dcr-writer.js | 26 +++++++---------- test/dcr-writer.test.js | 60 ++++++++++++++++++++++++++++++++++++++-- 2 files changed, 68 insertions(+), 18 deletions(-) diff --git a/src/ingest/dcr-writer.js b/src/ingest/dcr-writer.js index e862772..4a1d43a 100644 --- a/src/ingest/dcr-writer.js +++ b/src/ingest/dcr-writer.js @@ -149,24 +149,16 @@ const MAX_REPORTED_CAUSES = 3; /** * @azure/monitor-ingestion rejects with an AggregateLogsUploadError whose own `.message` is the * literal `undefined\n}` — its constructor interpolates an argument the SDK never passes. The - * status code and text are on `errors[i].cause`, a RestError. Read the predicate lazily, like the - * rest of the Azure SDK in this file, and never let the lookup itself throw: this runs from a catch. + * status code and text are on `errors[i].cause`, a RestError. The array is the whole test: the SDK + * constructor always assigns `.errors`, while its isAggregateLogsUploadError also accepts a + * name-only object with no `.errors` to read, which would throw here — inside a catch handler. */ -let aggregatePredicate; function isAggregateUploadError(err) { - if (!err) return false; - if (aggregatePredicate === undefined) { - try { - aggregatePredicate = require('@azure/monitor-ingestion').isAggregateLogsUploadError; - } catch (_err) { - aggregatePredicate = null; - } - } - return Boolean(aggregatePredicate && aggregatePredicate(err)) || Array.isArray(err.errors); + return Boolean(err) && Array.isArray(err.errors); } function causeStatusCodes(err) { - if (isAggregateUploadError(err)) return err.errors.map((entry) => entry.cause && entry.cause.statusCode); + if (isAggregateUploadError(err)) return err.errors.map((entry) => entry && entry.cause && entry.cause.statusCode); return [err && err.statusCode]; } @@ -175,9 +167,11 @@ function describeUploadError(err) { if (!isAggregateUploadError(err)) return err && err.message; const causes = []; - for (const { cause } of err.errors) { - const status = cause && cause.statusCode ? `${cause.statusCode} ` : ''; - const text = `${status}${(cause && cause.message) || 'no cause reported'}`; + for (const entry of err.errors) { + const cause = entry && entry.cause; + if (!cause) continue; + const status = cause.statusCode ? `${cause.statusCode} ` : ''; + const text = `${status}${cause.message || 'no cause reported'}`; if (!causes.includes(text)) causes.push(text); if (causes.length === MAX_REPORTED_CAUSES) break; } diff --git a/test/dcr-writer.test.js b/test/dcr-writer.test.js index 09d1fc5..1bab243 100644 --- a/test/dcr-writer.test.js +++ b/test/dcr-writer.test.js @@ -124,8 +124,12 @@ test('a drop line carries the cause the SDK buries, not its own placeholder mess await writer.flush(); // AggregateLogsUploadError.message is `undefined\n}`; an operator needs the status and the text. - assert.match(errors[0], /^\[analytics\] dropped 1 row\(s\) for Custom-EagleEvents_CL/); - assert.match(errors[0], /403 Operation returned an invalid status code Forbidden/); + // The tail says how many calls it took, which separates a missing role from a transient failure. + assert.strictEqual( + errors[0], + '[analytics] dropped 1 row(s) for Custom-EagleEvents_CL after 1 attempt: ' + + '403 Operation returned an invalid status code Forbidden' + ); assert.doesNotMatch(errors[0], /undefined/); }); @@ -151,6 +155,31 @@ test('a 503 keeps all three attempts', async (t) => { throw uploadError({ statusCode: 503, message: 'Service Unavailable' }); }); t.after(() => writer._resetTransport()); + const errors = []; + t.mock.method(logger, 'error', (message) => errors.push(message)); + + writer.enqueue(EVENTS_STREAM, row('page_view')); + await writer.flush(); + + assert.strictEqual(attempts, 3); + assert.strictEqual( + errors[0], + '[analytics] dropped 1 row(s) for Custom-EagleEvents_CL after 3 attempts: 503 Service Unavailable' + ); +}); + +// The SDK splits a batch into chunks and aggregates every chunk's failure, so one 403 next to a +// retryable cause must not cut the retries short: only an all-permission batch is hopeless. +test('a batch with one 403 and one 503 keeps all three attempts', async (t) => { + let attempts = 0; + writer._setTransport(async () => { + attempts += 1; + throw uploadError( + { statusCode: 403, message: 'Forbidden' }, + { statusCode: 503, message: 'Service Unavailable' } + ); + }); + t.after(() => writer._resetTransport()); t.mock.method(logger, 'error', () => {}); writer.enqueue(EVENTS_STREAM, row('page_view')); @@ -159,6 +188,24 @@ test('a 503 keeps all three attempts', async (t) => { assert.strictEqual(attempts, 3); }); +test('a 401 is dropped on the first attempt, like a 403', async (t) => { + let attempts = 0; + writer._setTransport(async () => { + attempts += 1; + throw uploadError( + { statusCode: 401, message: 'Unauthorized' }, + { statusCode: 401, message: 'Unauthorized' } + ); + }); + t.after(() => writer._resetTransport()); + t.mock.method(logger, 'error', () => {}); + + writer.enqueue(EVENTS_STREAM, row('page_view')); + await writer.flush(); + + assert.strictEqual(attempts, 1); +}); + test('describeUploadError returns a plain error by its own message', () => { assert.strictEqual(writer.describeUploadError(new Error('socket hang up')), 'socket hang up'); }); @@ -173,6 +220,15 @@ test('describeUploadError collapses causes that repeat across a batch', () => { assert.strictEqual(described, '403 Forbidden; 429 Too many requests'); }); +test('describeUploadError skips an errors entry that carries no cause', () => { + const described = writer.describeUploadError(uploadError( + undefined, + { statusCode: 429, message: 'Too many requests' } + )); + + assert.strictEqual(described, '429 Too many requests'); +}); + test('describeUploadError reports at most three distinct causes', () => { const described = writer.describeUploadError(uploadError( { statusCode: 400, message: 'one' },