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..4a1d43a 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,52 @@ 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. 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. + */ +function isAggregateUploadError(err) { + return Boolean(err) && Array.isArray(err.errors); +} + +function causeStatusCodes(err) { + if (isAggregateUploadError(err)) return err.errors.map((entry) => 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 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; + } + 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 +195,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 +213,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..1bab243 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,131 @@ 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. + // 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/); +}); + +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()); + 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')); + await writer.flush(); + + 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'); +}); + +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 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' }, + { statusCode: 400, message: 'two' }, + { statusCode: 400, message: 'three' }, + { statusCode: 400, message: 'four' } + )); + + assert.strictEqual(described, '400 one; 400 two; 400 three'); +});