fix(ingest): log the real cause when a DCR upload fails

docs/MIGRATION.md

@@ -77,6 +77,22 @@ az storage blob upload --auth-mode login --account-name <analyticsfc…> \
   --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-<env>`. 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 <rg> -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

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;
     });
 }

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,19 +142,66 @@ 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 {
       await sendBatch(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-<env> 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,

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');
+});