bigquery-telemetry

1// Copyright 2026 The "Anthropic on Google Cloud" Authors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     https://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15// A Claude Code mod that copies each OpenTelemetry event record into
16// BigQuery, one table per event type, in the layout a Cloud Logging sink
17// writes. README.md describes the options, the row format and the limits.
18
19import {
20  MAX_BATCH_BYTES,
21  MAX_BATCH_ROWS,
22  MAX_ROW_BYTES,
23  MAX_VALUE_BYTES,
24  backoffMs,
25  fitToTable,
26  insertAllBody,
27  jsonBytes,
28  outcomeOf,
29  resourceAttributesOf,
30  rowOf,
31  sessionFieldsOf,
32  tableNameOf,
33  takeBatch,
34  truncated,
35  wantsHostname,
36} from './rows.mjs'
37
38const LOG_PREFIX = 'bigquery-telemetry: '
39const METADATA_TOKEN_URL =
40  'http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/token'
41const METADATA_EMAIL_URL =
42  'http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/email'
43const TOKENINFO_URL = 'https://oauth2.googleapis.com/tokeninfo'
44// How long a failed lookup of the credentials' email address is kept before
45// it is tried again.
46const EMAIL_RETRY_MS = 10 * 60 * 1000
47// With on_auth_failure "block", how long a prompt waits for a token before
48// it is refused.
49const AUTH_CHECK_TIMEOUT_MS = 25_000
50// While credentials are the problem, retries come at least this often, so the
51// warning clears soon after the developer signs in.
52const AUTH_RETRY_MAX_MS = 30_000
53// gcloud doesn't say when its token expires and may print one it cached
54// earlier, so reuse it for 10 minutes at most, and get a new one on a 401.
55const GCLOUD_TOKEN_REUSE_MS = 10 * 60 * 1000
56// How long a table's columns, read with tables.get, are trusted before
57// they are read again (setup.sh may have added some).
58const TABLE_FIELDS_REUSE_MS = 10 * 60 * 1000
59// Batches sent at once, usually each for its own table.
60const MAX_PARALLEL_SENDS = 8
61// Cap on rows held in memory while BigQuery can't be reached.
62const MAX_QUEUE_BYTES = 16 * 1024 * 1024
63// Cap on the rows one session keeps in the store. All sessions share the
64// mod's store, which holds 4 MiB in total.
65const MAX_SPOOL_BYTES = 1024 * 1024
66const SPOOL_PREFIX = 'unsent:'
67// A session that just ended may still be sending the rows it saved, so
68// other sessions leave saved rows alone until they are this old. A running
69// session waiting to retry can own older ones; taking those can duplicate rows.
70const SPOOL_MIN_AGE_MS = 30_000
71
72let config
73const state = {
74  queue: [], // items { table, insertId, json, bytes } waiting to be sent
75  queuedBytes: 0,
76  inFlight: [], // the rows being sent now
77  context: undefined, // promise of what every row needs from the session
78  token: undefined, // { value, expiresAt }
79  tokenRequest: undefined, // promise of the token being fetched now, shared by every caller
80  email: undefined, // { value, readAt }: the credentials' email address, '' if unknown
81  emailRequest: undefined, // promise of the lookup running now, shared by every caller
82  tables: new Map(), // table name -> { fields, readAt }: its columns, from tables.get
83  tableReads: new Map(), // table name -> promise of the tables.get running now
84  prefetched: new Set(), // tables whose columns enqueue() has already asked for
85  failures: 0, // failed attempts in a row
86  retryAt: 0, // no attempt before this time (ms since the epoch)
87  flushing: undefined, // promise of the running drain
88  timerStarted: false, // whether the flush timer runs; it outlives a session id
89  disabled: '', // why the mod stopped sending, when it did
90  ended: false, // the process is exiting
91  spoolKey: SPOOL_PREFIX + crypto.randomUUID(), // this process's key in the store
92  spooled: false, // whether the store holds rows under spoolKey
93  spoolWrites: Promise.resolve(true), // the latest write of saved rows; they run one at a time
94  taken: [], // [{ key, ids }]: other sessions' keys to delete once these rows are sent
95  warned: new Set(),
96  // Why rows can't be written: { kind: 'credentials' | 'permission', reason },
97  // shown above the prompt until a send succeeds.
98  authProblem: undefined,
99  toasted: new Set(), // kinds of authProblem already announced with a toast
100}
101
102export function register(on, options) {
103  config = configOf(options)
104  state.disabled = problemOf(config)
105
106  on('session.start', async ($, e, next) => {
107    if (state.disabled) await warnOnce($, 'disabled', 'no rows will be sent to BigQuery: ' + state.disabled)
108    try {
109      await contextOf($)
110    } catch (error) {
111      await warnOnce($, 'start', "could not read the session's settings; trying again with the next record: " + messageOf(error))
112    }
113    if (!state.disabled) {
114      // session.start can fire again in the same process (after /clear or a
115      // resume), and the first timer still runs then.
116      if (!state.timerStarted) {
117        state.timerStarted = true
118        $.clock.every(config.intervalMs, () => {
119          void flush($)
120        })
121      }
122      // Get a token now, so that one is at hand at exit, and the email
123      // address it belongs to.
124      void tokenOf($)
125        .then(() => emailOf($))
126        .catch(() => {})
127      await replaySpool($)
128    }
129    return next(e)
130  })
131
132  // Always pass the record on with next(e), and first: refusing or holding
133  // it would also affect the OpenTelemetry export the organization may run.
134  on('telemetry.log', { to: 'collector' }, async ($, e, next) => {
135    const result = await next(e)
136    // If a later hook refused the record, the export doesn't get it, so neither does BigQuery.
137    if (result?.deny !== undefined) return result
138    try {
139      await enqueue($, e)
140    } catch (error) {
141      await warnOnce($, 'enqueue', 'could not queue a record, so it will not be sent: ' + messageOf(error))
142    }
143    return result
144  })
145
146  // While rows can't be written for lack of credentials or permission, a
147  // line above the prompt says why and how to fix it.
148  on('ui.render', { component: 'AbovePrompt' }, ($, e, next) => {
149    if (!state.authProblem || e.hasSurvey) return next(e)
150    const { Box, Text } = $.ui.resolve(e)
151    return authBand(Box, Text)
152  })
153
154  // With on_auth_failure "block", a prompt is refused while there are no
155  // working Google credentials, so that no session runs unrecorded. A crash
156  // here refuses the prompt too.
157  on('prompt.submit', async ($, e, next) => {
158    // A prompt while rows can't be written is a good moment to try again: the
159    // developer may have just signed in.
160    if (state.authProblem && !state.disabled) {
161      state.retryAt = 0
162      void flush($)
163    }
164    if (config.onAuthFailure !== 'block') return next(e)
165    const problem = await blockingProblem($)
166    if (problem) return { drop: problem }
167    return next(e)
168  }).catch(($, e, next) =>
169    next.called ? next(e) : { drop: 'bigquery-telemetry could not check the Google credentials, so the prompt was not sent.' },
170  )
171
172  on('session.end', async ($, e, next) => {
173    try {
174      if (e.reason === 'clear' || e.reason === 'resume') {
175        // The process goes on under a new session id, and the timer keeps
176        // sending; this just sends what is queued now.
177        void flush($)
178      } else {
179        await finish($, next.budget)
180      }
181    } catch (error) {
182      console.warn(LOG_PREFIX + 'could not save or send rows at exit: ' + messageOf(error))
183    }
184    return next(e)
185  })
186}
187
188function configOf(options) {
189  const o = options ?? {}
190  const text = (value, fallback) => {
191    const s = typeof value === 'string' ? value.trim() : ''
192    return s === '' ? fallback : s
193  }
194  const project = text(o.project, '')
195  const seconds = Number(o.flush_interval_seconds)
196  return {
197    project,
198    dataset: text(o.dataset, 'claude_code_telemetry'),
199    tablePrefix: text(o.table_prefix, ''),
200    hostnameTaskId: o.hostname_task_id !== false && o.hostname_task_id !== 'false',
201    emailFromCredentials: o.email_from_credentials !== false && o.email_from_credentials !== 'false',
202    auth: text(o.auth, 'adc'),
203    onAuthFailure: text(o.on_auth_failure, 'warn').toLowerCase(),
204    quotaProject: text(o.quota_project, ''),
205    intervalMs: (Number.isFinite(seconds) && seconds >= 1 ? Math.min(seconds, 300) : 5) * 1000,
206    apiBase: text(o.api_base, 'https://bigquery.googleapis.com').replace(/\/+$/, ''),
207  }
208}
209
210// Why the options can't work, or '' if they can. Checked before anything is
211// sent, so that a token never goes anywhere but Google.
212function problemOf(c) {
213  if (!c.project) {
214    return 'the "project" option is not set. Set it in pluginConfigs in your user or managed ' +
215      'settings; Claude Code does not read plugin options from project settings.'
216  }
217  if (!['warn', 'block'].includes(c.onAuthFailure)) {
218    return 'the "on_auth_failure" option is "' + c.onAuthFailure + '", but it must be warn or block.'
219  }
220  if (!['adc', 'gcloud', 'metadata', 'none'].includes(c.auth)) {
221    return 'the "auth" option is "' + c.auth + '", but it must be adc, gcloud, metadata or none.'
222  }
223  if (c.auth !== 'none' && !isGoogleApi(c.apiBase)) {
224    return 'the "api_base" option must be an https://*.googleapis.com URL unless "auth" is none, ' +
225      'so that a Google token is never sent anywhere else.'
226  }
227  if (!isUrl(c.apiBase)) return 'the "api_base" option is not an http or https URL.'
228  if (!/^[A-Za-z0-9_]*$/.test(c.tablePrefix)) {
229    return 'the "table_prefix" option may contain only letters, digits and underscores, as in setup.sh.'
230  }
231  return ''
232}
233
234// What every row needs from the session, read once.
235function contextOf($) {
236  if (!state.context) {
237    state.context = loadContext($).catch((error) => {
238      state.context = undefined
239      throw error
240    })
241  }
242  return state.context
243}
244
245async function loadContext($) {
246  const resourceAttributes = await $.env.get('OTEL_RESOURCE_ATTRIBUTES')
247  const serviceName = await $.env.get('OTEL_SERVICE_NAME')
248  const nonessentialOff = await $.env.get('CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC')
249
250  if (nonessentialOff && !state.disabled) {
251    await disable(
252      $,
253      'CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC is set, which makes Claude Code refuse every ' +
254        "network request from a mod. Unset it to send rows (DISABLE_TELEMETRY doesn't block the mod).",
255    )
256  }
257  const attributes = resourceAttributesOf(resourceAttributes, serviceName)
258  const hostname = config.hostnameTaskId && !state.disabled && wantsHostname(attributes) ? await hostnameOf($) : ''
259  return sessionFieldsOf(attributes, config.project, hostname)
260}
261
262// The machine's hostname, for the resource's task_id, or '' if it can't be found.
263async function hostnameOf($) {
264  try {
265    const run = await $.process.run(['hostname'], { timeoutMs: 5_000 })
266    const name = run.exitCode === 0 ? run.stdout.trim() : ''
267    if (name) return name
268  } catch {
269    // Fall back to the environment below.
270  }
271  return ((await $.env.get('HOSTNAME')) || (await $.env.get('COMPUTERNAME')) || '').trim()
272}
273
274async function enqueue($, e) {
275  if (state.disabled) return
276  const context = await contextOf($)
277  if (state.disabled) return
278  const { table, json } = rowOf(e, context, crypto.randomUUID(), new Date().toISOString(), config.tablePrefix)
279  const bytes = jsonBytes(json)
280  if (bytes > MAX_ROW_BYTES) {
281    await warnOnce(
282      $,
283      'too-big:' + e.event,
284      'a record of type ' + e.event + ' (about ' + Math.ceil(bytes / 1024) + ' KiB) was not sent: the mod drops records over ' +
285        MAX_ROW_BYTES / 1024 + " KiB, Cloud Logging's limit for one log entry. Records this large usually come from " +
286        'OTEL_LOG_USER_PROMPTS, OTEL_LOG_ASSISTANT_RESPONSES, OTEL_LOG_TOOL_DETAILS or OTEL_LOG_RAW_API_BODIES; ' +
287        "unset them if that content doesn't need to reach BigQuery.",
288    )
289    return
290  }
291  // Only this process's rows get its credentials' email address, and only
292  // when the record carries none of its own.
293  const wantsEmail = config.emailFromCredentials && !json.labels.user_email
294  pushBack([{ table, insertId: json.insertId, json, bytes, wantsEmail }])
295  // Read a new table's columns now, once, so that sending at exit needn't wait for it.
296  if (!state.prefetched.has(table)) {
297    state.prefetched.add(table)
298    void fieldsOf($, table).catch(() => {})
299  }
300  while (state.queuedBytes > MAX_QUEUE_BYTES && state.queue.length > 1) {
301    const oldest = state.queue.shift()
302    state.queuedBytes -= oldest.bytes
303    await warnOnce(
304      $,
305      'full',
306      "over 16 MiB of rows is waiting because BigQuery hasn't accepted any for a while, so the oldest rows are being dropped.",
307    )
308  }
309  if (state.ended) {
310    // A record that arrives after session.end: save it with the rest.
311    await spool($)
312  } else if (state.queue.length >= MAX_BATCH_ROWS || state.queuedBytes >= MAX_BATCH_BYTES) {
313    void flush($)
314  }
315}
316
317function pushBack(items) {
318  for (const item of items) {
319    state.queue.push(item)
320    state.queuedBytes += item.bytes
321  }
322}
323
324function pushFront(items) {
325  state.queue.unshift(...items)
326  for (const item of items) state.queuedBytes += item.bytes
327}
328
329// Starts sending what is queued, unless a send is already running. The
330// returned promise never rejects.
331function flush($) {
332  if (!state.flushing) {
333    state.flushing = drain($)
334      .catch((error) =>
335        console.warn(LOG_PREFIX + 'sending stopped on an unexpected error and resumes at the next interval: ' + messageOf(error)),
336      )
337      .finally(() => {
338        state.flushing = undefined
339      })
340  }
341  return state.flushing
342}
343
344async function drain($) {
345  while (state.queue.length > 0 && !state.disabled && (await $.clock.now()) >= state.retryAt) {
346    // Each kind of event has its own table, so batches go out together, up
347    // to 8 at once: at exit, one round trip covers that many tables.
348    const batches = []
349    while (state.queue.length > 0 && batches.length < MAX_PARALLEL_SENDS) batches.push(takeBatch(state.queue))
350    for (const item of batches.flat()) state.queuedBytes -= item.bytes
351    state.inFlight = batches.flat()
352    const outcomes = await Promise.all(
353      batches.map(async (batch) => {
354        const outcome = await sendOrFail($, batch)
355        // Settle each batch as soon as it is answered, so that the saved copies
356        // stay in step with what is still unsent even if the process exits now.
357        state.inFlight = state.inFlight.filter((item) => !batch.includes(item))
358        pushFront(outcome.retry)
359        if (state.ended || state.spooled) await spool($)
360        if (state.taken.length > 0) {
361          try {
362            await releaseTaken($, [...outcome.sent, ...outcome.dropped])
363          } catch (error) {
364            // The other session's key stays; its rows may be sent again later.
365            console.warn(LOG_PREFIX + 'could not remove rows another session saved: ' + messageOf(error))
366          }
367        }
368        return outcome
369      }),
370    )
371    const egress = outcomes.find((o) => o.egressRefused)
372    if (egress) {
373      await disable($, 'Claude Code refused the request to BigQuery: ' + egress.reason + '.')
374      return
375    }
376    const failed = outcomes.some(isFailure)
377    let waitMs = failed ? backoffMs(state.failures + 1) : 0
378    if (state.authProblem?.kind === 'credentials') waitMs = Math.min(waitMs, AUTH_RETRY_MAX_MS)
379    for (const outcome of outcomes) await report($, outcome, waitMs)
380    state.failures = failed ? state.failures + 1 : 0
381    state.retryAt = failed ? (await $.clock.now()) + waitMs : 0
382    if (state.failures > 0) return
383  }
384}
385
386// send(), with a failure before BigQuery answered turned into an outcome.
387async function sendOrFail($, batch) {
388  try {
389    return await send($, batch)
390  } catch (error) {
391    const message = messageOf(error)
392    // Claude Code refuses every request from the mod.
393    const egress = message.match(/\$\.http\.fetch: refused: (.*)$/)
394    if (egress) return { sent: [], retry: batch, dropped: [], reason: egress[1], egressRefused: true }
395    // Claude Code refused this one request, for example for its size.
396    if (/\$\.http\.fetch: \S+ refused: /.test(message)) {
397      return { sent: [], retry: [], dropped: batch, reason: message, retryLater: false, refusedByClaudeCode: true }
398    }
399    // The token or the request failed before BigQuery answered.
400    return { sent: [], retry: batch, dropped: [], reason: message, retryLater: true }
401  }
402}
403
404// Whether an outcome calls for a backoff before the next attempt.
405function isFailure(outcome) {
406  return outcome.retryLater || (outcome.sent.length === 0 && outcome.dropped.length === 0)
407}
408
409// Warns about rows an outcome dropped, or will send again after waitMs.
410async function report($, outcome, waitMs) {
411  if (outcome.dropped.length > 0) {
412    await warnOnce(
413      $,
414      'dropped:' + outcome.reason,
415      (outcome.refusedByClaudeCode
416        ? 'Claude Code refused the request for ' + outcome.dropped.length + " row(s), so they won't be sent: "
417        : outcome.missingTable
418          ? outcome.dropped.length + " row(s) won't be sent: "
419          : 'BigQuery refused ' + outcome.dropped.length + " row(s), so they won't be sent: ") + outcome.reason,
420    )
421  }
422  if (isFailure(outcome)) {
423    await warnOnce(
424      $,
425      'retry:' + outcome.reason,
426      'could not send rows to BigQuery; trying again in about ' + Math.ceil(waitMs / 1000) + ' s: ' + outcome.reason,
427    )
428  }
429}
430
431// Sends a batch, all for one table, after removing the labels that table
432// has no column for.
433async function send($, batch) {
434  const table = batch[0].table
435  const fields = await fieldsOf($, table)
436  if (fields === 'missing') {
437    return {
438      sent: [], retry: [], dropped: batch, retryLater: false, missingTable: true,
439      reason: 'the table ' + config.project + ':' + config.dataset + '.' + table + " doesn't exist. Check the " +
440        'project, dataset and table_prefix options. If they are right, this is a kind of event setup.sh made no ' +
441        'table for: add it to event-labels.json and run setup.sh again.',
442    }
443  }
444  if (fields) {
445    for (const item of batch) {
446      const fitted = fitToTable(item.json, fields)
447      item.json = fitted.json
448      for (const name of fitted.dropped) {
449        // Debug log only: this can happen in every session until setup.sh is run again.
450        if (!state.warned.has('column:' + table + '.' + name)) {
451          state.warned.add('column:' + table + '.' + name)
452          console.warn(
453            LOG_PREFIX + 'the table ' + table + ' has no column ' + name +
454              (name.startsWith('labels.') ? ' and no labels.extra_labels' : '') +
455              ", so that value isn't sent. Run setup.sh again to add the missing columns.",
456          )
457        }
458      }
459    }
460  }
461  // The credentials' email address goes only into the request, never into
462  // item.json, so it is never written to the store with unsent rows.
463  const email = batch.some((item) => item.wantsEmail) ? await emailOf($) : ''
464  const toSend = batch.map((item) => withEmail(item, email, fields))
465  const response = await call($, 'POST', tableUrlOf(table) + '/insertAll', insertAllBody(toSend))
466  if (response.status === 403) {
467    await noteAuthProblem($, 'permission', outcomeOf(batch, response.status, response.text).reason)
468  } else if (response.ok) {
469    await clearAuthProblem($)
470  }
471  return outcomeOf(batch, response.status, response.text)
472}
473
474// The item to send: with labels.user_email set to email when the item wants
475// it, has none, and its table has that column (or its columns are unknown).
476function withEmail(item, email, fields) {
477  if (!email || !item.wantsEmail || item.json.labels?.user_email) return item
478  const labelFields = fields?.find?.((f) => f.name === 'labels')?.fields
479  if (labelFields && !labelFields.some((f) => f.name === 'user_email')) return item
480  return { ...item, json: { ...item.json, labels: { ...item.json.labels, user_email: truncated(email, MAX_VALUE_BYTES) } } }
481}
482
483function tableUrlOf(table) {
484  return (
485    config.apiBase +
486    '/bigquery/v2/projects/' + encodeURIComponent(config.project) +
487    '/datasets/' + encodeURIComponent(config.dataset) +
488    '/tables/' + encodeURIComponent(table)
489  )
490}
491
492// The table's columns from tables.get, read at most every 10 minutes:
493// 'missing' if the table doesn't exist, or undefined if BigQuery refused to
494// show them (BigQuery then drops the values that have no column). Throws when
495// the read should be tried again later (401, 408, 429, 5xx, network).
496async function fieldsOf($, table) {
497  const now = await $.clock.now()
498  const known = state.tables.get(table)
499  if (known && now - known.readAt < TABLE_FIELDS_REUSE_MS) return known.fields
500  if (!state.tableReads.has(table)) {
501    state.tableReads.set(table, readFields($, table, now).finally(() => state.tableReads.delete(table)))
502  }
503  return state.tableReads.get(table)
504}
505
506async function readFields($, table, now) {
507  const response = await call($, 'GET', tableUrlOf(table) + '?fields=schema')
508  if (response.status === 404) {
509    state.tables.set(table, { fields: 'missing', readAt: now })
510    return 'missing'
511  }
512  if (response.status === 401 || response.status === 408 || response.status === 429 || response.status >= 500) {
513    throw new Error('could not read the columns of ' + table + ': HTTP ' + response.status)
514  }
515  let fields
516  try {
517    fields = response.ok ? JSON.parse(response.text).schema?.fields : undefined
518  } catch {
519    fields = undefined
520  }
521  if (!Array.isArray(fields)) {
522    state.tables.set(table, { fields: undefined, readAt: now })
523    await warnOnce(
524      $,
525      'fields',
526      'could not read the columns of ' + table + ' (HTTP ' + response.status + '), so values the table has no column for ' +
527        'are dropped by BigQuery without a warning. Writers need bigquery.tables.get as well as bigquery.tables.updateData.',
528    )
529    return undefined
530  }
531  state.tables.set(table, { fields, readAt: now })
532  return fields
533}
534
535// A request to BigQuery with the token, sent once more with a new token after a 401.
536async function call($, method, url, body) {
537  let { response, token } = await fetchWithToken($, method, url, body)
538  if (response.status === 401 && config.auth !== 'none') {
539    // The token may have expired early: get a new one and try once more. Only
540    // the token this request used is thrown away, so that requests answered
541    // 401 at the same time share one new token.
542    if (state.token?.value === token) state.token = undefined
543    ;({ response } = await fetchWithToken($, method, url, body))
544    if (response.status === 401) {
545      await noteAuthProblem($, 'credentials', 'BigQuery rejected the credentials (HTTP 401) even after getting a new token')
546    }
547  }
548  if (response.ok && state.authProblem?.kind === 'credentials') await clearAuthProblem($)
549  return response
550}
551
552async function fetchWithToken($, method, url, body) {
553  const headers = {}
554  if (body !== undefined) headers['Content-Type'] = 'application/json'
555  const token = await tokenOf($)
556  if (token) {
557    // problemOf() already refused such an api_base; this is a second check.
558    if (!isGoogleApi(url)) throw new Error('refusing to send a Google token to ' + new URL(url).origin)
559    headers.Authorization = 'Bearer ' + token
560  }
561  if (config.quotaProject) headers['x-goog-user-project'] = config.quotaProject
562  const response = await $.http.fetch(url, body === undefined ? { method, headers } : { method, headers, body })
563  return { response, token }
564}
565
566// An OAuth access token for BigQuery, or '' when auth is "none". Callers
567// share one fetch while it runs, so gcloud never runs twice at once.
568async function tokenOf($) {
569  if (config.auth === 'none') return ''
570  const now = await $.clock.now()
571  if (state.token && state.token.expiresAt > now) return state.token.value
572  if (!state.tokenRequest) {
573    state.tokenRequest = newToken($, now)
574      .catch(async (error) => {
575        await noteAuthProblem($, 'credentials', messageOf(error))
576        throw error
577      })
578      .finally(() => {
579        state.tokenRequest = undefined
580      })
581  }
582  return state.tokenRequest
583}
584
585async function newToken($, now) {
586  let value
587  let lifetimeMs = GCLOUD_TOKEN_REUSE_MS
588  if (config.auth === 'metadata') {
589    const response = await $.http.fetch(METADATA_TOKEN_URL, { headers: { 'Metadata-Flavor': 'Google' } })
590    if (!response.ok) throw new Error('the metadata server answered HTTP ' + response.status + ' when asked for an access token')
591    const body = JSON.parse(response.text)
592    value = body.access_token
593    const expiresIn = Number(body.expires_in)
594    // Reuse it until a minute before it expires; if the answer doesn't say, for 5 minutes.
595    lifetimeMs = Number.isFinite(expiresIn) ? Math.max(0, (expiresIn - 60) * 1000) : 5 * 60 * 1000
596  } else {
597    const run =
598      config.auth === 'gcloud'
599        ? await $.process.run(['gcloud', 'auth', 'print-access-token'], { timeoutMs: 20_000 })
600        : await $.process.run(['gcloud', 'auth', 'application-default', 'print-access-token'], { timeoutMs: 20_000 })
601    if (run.exitCode !== 0) {
602      throw new Error('gcloud could not print an access token (exit ' + run.exitCode + '): ' + errorLine(run.stderr))
603    }
604    value = run.stdout.trim()
605  }
606  if (!value) throw new Error('the ' + (config.auth === 'metadata' ? 'metadata server' : 'gcloud command') + ' returned no access token')
607  state.token = { value, expiresAt: now + lifetimeMs }
608  return value
609}
610
611// The email address of the account the credentials belong to, or '' when it
612// is off, unknown, or can't be found. Looked up once per process; a failed
613// lookup is tried again after 10 minutes, and one that failed only because
614// there was no token, with the next send. Never throws.
615async function emailOf($) {
616  if (!config.emailFromCredentials || config.auth === 'none') return ''
617  const now = await $.clock.now()
618  if (state.email && (state.email.value || now - state.email.readAt < EMAIL_RETRY_MS)) return state.email.value
619  if (!state.emailRequest) {
620    state.emailRequest = lookUpEmail($)
621      .then((value) => {
622        state.email = { value, readAt: now }
623        return value
624      })
625      .catch(async (error) => {
626        // No token, or Claude Code refuses the mod's requests: send() reports
627        // that already, and the lookup can run again.
628        if (error?.noToken || /\$\.http\.fetch: refused: /.test(messageOf(error))) return ''
629        state.email = { value: '', readAt: now }
630        await warnOnce($, 'email', "could not find the credentials' email address, so rows without a user.email " +
631          'attribute are sent without one: ' + messageOf(error))
632        return ''
633      })
634      .finally(() => {
635        state.emailRequest = undefined
636      })
637  }
638  return state.emailRequest
639}
640
641async function lookUpEmail($) {
642  if (config.auth === 'metadata') {
643    const response = await $.http.fetch(METADATA_EMAIL_URL, { headers: { 'Metadata-Flavor': 'Google' } })
644    if (!response.ok) throw new Error('the metadata server answered HTTP ' + response.status)
645    return response.text.trim()
646  }
647  // Google's tokeninfo endpoint names the account a token was issued to, when
648  // it has the email scope (gcloud's tokens have it by default). The token
649  // goes in the body, not the URL.
650  const token = await tokenOf($).catch((error) => {
651    throw Object.assign(new Error(messageOf(error)), { noToken: true })
652  })
653  if (!isGoogleApi(TOKENINFO_URL)) throw new Error('refusing to send a Google token to ' + new URL(TOKENINFO_URL).origin)
654  const response = await $.http.fetch(TOKENINFO_URL, {
655    method: 'POST',
656    headers: { 'Content-Type': 'application/x-www-form-urlencoded' },
657    body: 'access_token=' + encodeURIComponent(token),
658  })
659  if (!response.ok) throw new Error('tokeninfo answered HTTP ' + response.status)
660  const info = JSON.parse(response.text)
661  if (!info.email) throw new Error("the token has no email scope, so tokeninfo doesn't name its account")
662  if (String(info.email_verified) !== 'true') throw new Error('tokeninfo says the email address is not verified')
663  return String(info.email)
664}
665
666// Called at exit. Saves unsent rows first, because Claude Code may stop
667// the mod at any moment now, then sends them if a token is at hand, taking
668// at most half of what is left of the shared exit budget, and at most 1 s.
669async function finish($, budget) {
670  state.ended = true
671  if (state.disabled) return
672  await spool($)
673  if (state.queue.length === 0 && state.inFlight.length === 0) return
674  const waitMs = Math.min((budget?.remainingMs ?? 0) / 2, 1000)
675  const hasToken = config.auth === 'none' || (state.token && state.token.expiresAt > (await $.clock.now()))
676  if (!hasToken || waitMs < 200) return
677  state.retryAt = 0
678  await Promise.race([flush($), $.clock.sleep(waitMs)])
679}
680
681// Writes the unsent rows (the batches being sent and the queue) under this
682// process's key, or removes the key when nothing is left. Returns true only
683// if every unsent row is now saved. Writes run one at a time, each saving
684// what is unsent when it runs, so the last one always reflects the latest state.
685function spool($) {
686  const write = state.spoolWrites.then(() => writeSpool($))
687  state.spoolWrites = write
688  return write
689}
690
691async function writeSpool($) {
692  const rows = []
693  let bytes = 0
694  let lost = 0
695  for (const item of [...state.inFlight, ...state.queue]) {
696    if (bytes + item.bytes > MAX_SPOOL_BYTES) {
697      lost += 1
698      continue
699    }
700    rows.push({ table: item.table, insertId: item.insertId, json: item.json })
701    bytes += item.bytes
702  }
703  try {
704    if (rows.length === 0) {
705      if (state.spooled) await $.store.delete(state.spoolKey)
706    } else {
707      await $.store.set(state.spoolKey, { writtenAt: await $.clock.now(), rows })
708    }
709    state.spooled = rows.length > 0
710  } catch (error) {
711    console.warn(LOG_PREFIX + 'could not save unsent rows; they are lost if Claude Code exits before sending them: ' + messageOf(error))
712    return false
713  }
714  if (lost > 0) {
715    console.warn(
716      LOG_PREFIX + lost + ' unsent row(s) were not saved because a session may save only 1 MiB; ' +
717        'they are lost if Claude Code exits before sending them',
718    )
719  }
720  return lost === 0
721}
722
723// Takes over rows that other sessions saved, whole keys at a time while
724// they fit in this session's 1 MiB: queues them and saves them under this
725// process's key, then removes the old keys; if that save fails, it removes
726// each old key once its rows are sent. A key can belong to a running session
727// waiting to retry; taking it can duplicate rows, which the spend views remove.
728async function replaySpool($) {
729  try {
730    const now = await $.clock.now()
731    const taken = []
732    const rows = []
733    let bytes = state.queuedBytes
734    for (const key of await $.store.keys()) {
735      if (!key.startsWith(SPOOL_PREFIX) || key === state.spoolKey) continue
736      const saved = await $.store.get(key)
737      if (saved && now - Number(saved.writtenAt ?? 0) < SPOOL_MIN_AGE_MS) continue
738      const keyRows = []
739      for (const r of Array.isArray(saved?.rows) ? saved.rows : []) {
740        // A saved row goes to its event's table under the current table_prefix,
741        // which may differ from the one it was saved with, so r.table is only
742        // checked for being there. Anything else is dropped.
743        const table = typeof r?.table === 'string' && r.json && typeof r.json === 'object' ? savedTableOf(r.json) : ''
744        if (table && typeof r.insertId === 'string') {
745          keyRows.push({ table, insertId: r.insertId, json: r.json, bytes: jsonBytes(r.json) })
746        }
747      }
748      const keyBytes = keyRows.reduce((sum, r) => sum + r.bytes, 0)
749      if (bytes + keyBytes > MAX_SPOOL_BYTES) continue
750      bytes += keyBytes
751      taken.push({ key, ids: new Set(keyRows.map((r) => r.insertId)) })
752      rows.push(...keyRows)
753    }
754    if (taken.length === 0) return
755    pushFront(rows)
756    if (await spool($)) {
757      for (const { key } of taken) await $.store.delete(key)
758    } else {
759      state.taken.push(...taken)
760    }
761  } catch (error) {
762    await warnOnce($, 'replay', 'could not pick up rows that earlier sessions saved; they stay saved for a later session: ' + messageOf(error))
763  }
764}
765
766// The table for a saved row, from the event name at the end of its logName.
767function savedTableOf(json) {
768  const match = typeof json.logName === 'string' ? json.logName.match(/\/logs\/([^/]+)$/) : null
769  if (!match) return ''
770  try {
771    return tableNameOf(decodeURIComponent(match[1]), config.tablePrefix)
772  } catch {
773    return ''
774  }
775}
776
777// Removes other sessions' keys whose rows have all been sent or dropped.
778async function releaseTaken($, done) {
779  for (const item of done) for (const t of state.taken) t.ids.delete(item.insertId)
780  const finished = state.taken.filter((t) => t.ids.size === 0)
781  state.taken = state.taken.filter((t) => t.ids.size > 0)
782  for (const { key } of finished) await $.store.delete(key)
783}
784
785async function disable($, reason) {
786  state.disabled = reason
787  // Saves what is unsent, including rows taken over from other sessions,
788  // for a later session that may be able to send it. Once that is saved,
789  // the other sessions' keys holding the taken rows can go.
790  if ((await spool($)) && state.taken.length > 0) {
791    try {
792      for (const { key } of state.taken) await $.store.delete(key)
793      state.taken = []
794    } catch (error) {
795      console.warn(LOG_PREFIX + 'could not remove rows another session saved: ' + messageOf(error))
796    }
797  }
798  state.queue = []
799  state.queuedBytes = 0
800  state.inFlight = []
801  await warnOnce($, 'disabled', 'no rows will be sent to BigQuery: ' + reason)
802}
803
804// Records why rows can't be written, and redraws the band. The first time
805// each kind of problem appears in this process, a toast says so too.
806async function noteAuthProblem($, kind, reason) {
807  if (config.auth === 'none') return
808  const changed = state.authProblem?.kind !== kind || state.authProblem?.reason !== reason
809  state.authProblem = { kind, reason }
810  if (!changed) return
811  redraw($)
812  if (!state.toasted.has(kind)) {
813    state.toasted.add(kind)
814    try {
815      await $.ui.toast('BigQuery telemetry. ' + problemSummary(), { timeoutMs: 15_000 })
816    } catch {
817      // Not every surface shows toasts; the band and the log still say it.
818    }
819  }
820  await warnOnce($, 'auth:' + kind + ':' + reason, problemSummary() + ' (' + reason + ')')
821}
822
823async function clearAuthProblem($) {
824  if (!state.authProblem) return
825  state.authProblem = undefined
826  redraw($)
827}
828
829function redraw($) {
830  try {
831    $.ui.invalidate('ui.render')
832  } catch {
833    // No drawing surface (claude -p, the SDK).
834  }
835}
836
837// One sentence: what's wrong and how to fix it.
838function problemSummary() {
839  const p = state.authProblem
840  if (!p) return ''
841  if (p.kind === 'permission') {
842    return 'Rows are not being written: missing BigQuery permission. Ask an admin to grant ' +
843      'bigquery.tables.updateData on ' + config.project + ':' + config.dataset + '.'
844  }
845  const fix = {
846    adc: 'run: gcloud auth application-default login',
847    gcloud: 'run: gcloud auth login',
848    metadata: "check this machine's attached service account",
849  }[config.auth]
850  return 'Rows are not being written: no working Google credentials. To fix, ' + fix + '.'
851}
852
853function authBand(Box, Text) {
854  const waiting = state.queue.length + state.inFlight.length
855  const blocking = config.onAuthFailure === 'block' && state.authProblem.kind === 'credentials'
856  const parts = [
857    Text({ color: blocking ? 'red' : 'yellow', bold: true, children: '⚠ BigQuery telemetry: ' }),
858    Text({ children: problemSummary().replace(/^Rows are not being written: /, '') }),
859  ]
860  if (waiting > 0) parts.push(Text({ dimColor: true, children: '  ' + waiting + ' row(s) waiting' }))
861  if (blocking) parts.push(Text({ color: 'red', children: '  Prompts are blocked until this is fixed.' }))
862  return Box({ flexDirection: 'row', flexWrap: 'wrap', paddingX: 1, children: parts })
863}
864
865// With on_auth_failure "block": why a prompt can't go ahead, or '' if it can.
866async function blockingProblem($) {
867  if (state.disabled) return 'bigquery-telemetry: prompts are blocked because no rows can be sent to BigQuery: ' + state.disabled
868  if (config.auth === 'none') return ''
869  // Neither side of the race rejects, so whichever loses is simply ignored.
870  const outcome = await Promise.race([
871    tokenOf($).then(() => ({ ok: true }), (error) => ({ error })),
872    $.clock.sleep(AUTH_CHECK_TIMEOUT_MS).then(() => ({ error: new Error('timed out getting a Google token') })),
873  ])
874  if (outcome.error) {
875    if (!state.authProblem) await noteAuthProblem($, 'credentials', messageOf(outcome.error))
876    return 'bigquery-telemetry: prompts are blocked until telemetry can be recorded. ' + problemSummary()
877  }
878  // A token was found. If BigQuery rejected the last one, check again before
879  // letting the prompt through: a 401 here means the credentials are still bad.
880  if (state.authProblem?.kind === 'credentials') {
881    try {
882      const response = await call($, 'GET', tableUrlOf(tableNameOf('user_prompt', config.tablePrefix)) + '?fields=schema')
883      if (response.status === 401) {
884        return 'bigquery-telemetry: prompts are blocked until telemetry can be recorded. ' + problemSummary()
885      }
886      await clearAuthProblem($)
887    } catch (error) {
888      // BigQuery rejected these credentials before and couldn't be reached to
889      // check again: fail closed, as the gate is meant to.
890      return 'bigquery-telemetry: prompts are blocked: BigQuery rejected the credentials earlier, and checking ' +
891        'them again failed (' + messageOf(error) + '). ' + problemSummary()
892    }
893  }
894  return ''
895}
896
897async function warn($, text) {
898  console.warn(LOG_PREFIX + text)
899  try {
900    await $.ui.log(text) // Claude Code adds the mod's name
901  } catch {
902    // Some surfaces have no transcript to write to; the debug log has it.
903  }
904}
905
906async function warnOnce($, key, text) {
907  if (state.warned.has(key)) {
908    console.warn(LOG_PREFIX + text)
909    return
910  }
911  state.warned.add(key)
912  await warn($, text)
913}
914
915function isUrl(text) {
916  try {
917    const url = new URL(text)
918    return url.protocol === 'https:' || url.protocol === 'http:'
919  } catch {
920    return false
921  }
922}
923
924function isGoogleApi(text) {
925  try {
926    const url = new URL(text)
927    return url.protocol === 'https:' && (url.hostname === 'googleapis.com' || url.hostname.endsWith('.googleapis.com'))
928  } catch {
929    return false
930  }
931}
932
933// One line from a command's error output, without control characters and at
934// most 200 characters: gcloud's ERROR: line if there is one (it can follow
935// WARNING: lines), else the first line. The rest (instructions, URLs) would
936// garble a one-line warning.
937function errorLine(text) {
938  const lines = String(text ?? '').split(/\r?\n/).map((l) => l.trim()).filter((l) => l)
939  const line = lines.find((l) => l.startsWith('ERROR:')) ?? lines[0] ?? ''
940  const clean = line.replace(/[\u0000-\u001f\u007f-\u009f]/g, ' ')
941  return clean.length > 200 ? clean.slice(0, 199) + '…' : clean
942}
943
944function messageOf(error) {
945  return error instanceof Error ? error.message : String(error)
946}
947

1// Copyright 2026 The "Anthropic on Google Cloud" Authors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     https://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15// Helpers for the mod that make no mods API calls, so tests can call
16// them directly: record to row, batching, and reading BigQuery's answers.
17
18// Each record becomes a row in the layout a Cloud Logging sink writes to
19// BigQuery when Claude Code's OpenTelemetry export goes through Google's
20// Telemetry API: one table per event type, camelCase LogEntry columns, and
21// labels as a RECORD of STRING fields. See "What a row contains" in README.md.
22// https://docs.cloud.google.com/logging/docs/export/bigquery
23// https://docs.cloud.google.com/stackdriver/docs/reference/telemetry/otlp-log-record-to-log-entry
24// https://docs.cloud.google.com/stackdriver/docs/reference/telemetry/otlp-attribute-to-logging-resource
25
26// Cloud Logging's limits for an entry made from an OTLP record: 256 KiB per
27// entry (the mod drops a bigger row), 512 B per attribute key and 64 KiB per value.
28export const MAX_ROW_BYTES = 256 * 1024
29const MAX_KEY_BYTES = 512
30export const MAX_VALUE_BYTES = 64 * 1024
31// A BigQuery field name that a sink writes is at most 128 characters.
32const MAX_FIELD_NAME = 128
33
34// insertAll takes at most 50,000 rows and 10 MB in one request. A batch is at
35// most 500 rows and 3 MiB, or one row of at most 256 KiB, well under both.
36export const MAX_BATCH_ROWS = 500
37export const MAX_BATCH_BYTES = 3 * 1024 * 1024
38
39// Attributes each resource label is read from, first non-empty wins; else ''.
40// The names are those Google's OTLP-to-resource mapping uses for generic_task.
41const RESOURCE_LABEL_KEYS = {
42  location: ['location', 'cloud.availability_zone', 'cloud.region'],
43  namespace: ['namespace', 'service.namespace'],
44  job: ['job', 'service.name'],
45  task_id: ['task_id', 'service.instance.id'],
46}
47
48const encoder = new TextEncoder()
49
50/** Size of a value's JSON text in bytes (UTF-8). */
51export function jsonBytes(value) {
52  return encoder.encode(JSON.stringify(value)).length
53}
54
55/**
56 * The resource attributes Claude Code sends that a mod can know. Later ones
57 * win: service.name "claude-code", then OTEL_RESOURCE_ATTRIBUTES, then
58 * OTEL_SERVICE_NAME.
59 */
60export function resourceAttributesOf(otelResourceAttributes, otelServiceName) {
61  const attributes = Object.create(null) // no prototype, so a __proto__ key is kept
62  attributes['service.name'] = 'claude-code'
63  Object.assign(attributes, envAttributesOf(otelResourceAttributes ?? ''))
64  if (otelServiceName) attributes['service.name'] = otelServiceName
65  return attributes
66}
67
68// OTEL_RESOURCE_ATTRIBUTES ("k1=v1,k2=v2"): pairs without a key are skipped,
69// quotes around a value are dropped, and percent-encoding is decoded where valid.
70function envAttributesOf(text) {
71  const attributes = Object.create(null)
72  for (const pair of text.split(',')) {
73    const at = pair.indexOf('=')
74    const key = at < 0 ? '' : pair.slice(0, at).trim()
75    if (!key) continue
76    const value = pair.slice(at + 1).trim().replace(/^"|"$/g, '')
77    try {
78      attributes[key] = decodeURIComponent(value)
79    } catch {
80      attributes[key] = value
81    }
82  }
83  return attributes
84}
85
86/**
87 * Whether the mod needs the machine's hostname for task_id: no attribute
88 * names the task.
89 */
90export function wantsHostname(attributes) {
91  return !RESOURCE_LABEL_KEYS.task_id.some((key) => attributes[key])
92}
93
94/**
95 * The monitored resource { type, labels }: always generic_task, with
96 * project_id the given project, job the service name (claude-code unless
97 * OTEL_SERVICE_NAME says otherwise), task_id the service.instance.id
98 * attribute or else hostname, and location and namespace from their
99 * attributes. A label with no value is ''. See "The resource" in README.md.
100 */
101export function resourceOf(attributes, project, hostname = '') {
102  const first = (keys) => keys.map((key) => attributes[key]).find((value) => value) ?? ''
103  const labels = { project_id: project }
104  for (const [name, keys] of Object.entries(RESOURCE_LABEL_KEYS)) labels[name] = first(keys)
105  if (!labels.task_id) labels.task_id = hostname
106  return { type: 'generic_task', labels }
107}
108
109/**
110 * A label value: lists (and any other object) as compact JSON text, as
111 * Google documents for lists. Numbers
112 * and booleans as JavaScript writes them ("34", "0.000188", "1e-7", "true"):
113 * not confirmed against a real Telemetry API export.
114 */
115export function labelValueOf(value) {
116  if (typeof value === 'string') return value
117  if (typeof value === 'object' && value !== null) return JSON.stringify(value)
118  return String(value)
119}
120
121/**
122 * text cut to at most maxBytes of UTF-8, ending in "…" when cut, as Cloud
123 * Logging marks a truncated label. Whether the ellipsis counts toward the
124 * limit is not confirmed against a real Telemetry API export.
125 */
126export function truncated(text, maxBytes) {
127  const bytes = encoder.encode(text)
128  if (bytes.length <= maxBytes) return text
129  const kept = new TextDecoder().decode(bytes.subarray(0, maxBytes - 3)) // "…" is 3 bytes
130  return kept.replace(/\uFFFD$/, '') + '…' // drop a character cut in half
131}
132
133/**
134 * The BigQuery field name a sink gives a label key: each character other
135 * than a letter, digit or underscore becomes "_", leading underscores are
136 * removed, and the name is lowercased and cut to 128 characters
137 * ("session.id" becomes session_id). '' when nothing is left.
138 */
139export function fieldNameOf(key) {
140  return key.replace(/[^A-Za-z0-9_]/gu, '_').replace(/^_+/, '').toLowerCase().slice(0, MAX_FIELD_NAME)
141}
142
143/**
144 * The table a sink writes a log to: the prefix, then the log ID with every
145 * character other than a letter, digit or underscore turned into "_" (Google
146 * shows "-", "." and "/" becoming "_").
147 */
148export function tableNameOf(logId, prefix = '') {
149  return prefix + logId.replace(/[^A-Za-z0-9_]/gu, '_')
150}
151
152/**
153 * Fields every row of a session shares: logProject (the project option, used
154 * in logName and resource.labels.project_id) and resource.
155 */
156export function sessionFieldsOf(resourceAttributes, project, hostname = '') {
157  return { logProject: project, resource: resourceOf(resourceAttributes, project, hostname) }
158}
159
160/**
161 * { table, json } for a telemetry.log event { event, attributes, loggedAt,
162 * span? }. The log ID is always the event name. context is what
163 * sessionFieldsOf returns; insertId must stay the same when the row is sent
164 * again. trace, spanId and traceSampled are left out when not set.
165 */
166export function rowOf(entry, context, insertId, receivedAt, tablePrefix = '') {
167  // The record's attributes and nothing else, as strings under their sink
168  // field names. After cutting and renaming, the first of two equal names stays.
169  const labels = Object.create(null)
170  for (const [key, value] of Object.entries(entry.attributes ?? {})) {
171    const name = fieldNameOf(truncated(key, MAX_KEY_BYTES))
172    if (name && !(name in labels)) labels[name] = truncated(labelValueOf(value), MAX_VALUE_BYTES)
173  }
174  const logId = entry.event
175  const span = entry.span
176  const json = {
177    timestamp: entry.loggedAt,
178    receiveTimestamp: receivedAt,
179    logName: 'projects/' + context.logProject + '/logs/' + encodeURIComponent(logId),
180    resource: context.resource,
181    severity: 'DEFAULT',
182    textPayload: 'claude_code.' + entry.event,
183    labels: { ...labels },
184    insertId,
185  }
186  if (span) {
187    // The bare 32-hex-digit trace ID: Google says "the value from the log
188    // record's traceId field", with no example.
189    json.trace = span.traceId
190    json.spanId = span.spanId
191    // Set only when the sampled bit is 1; "otherwise, it remains unset".
192    if ((span.traceFlags & 1) === 1) json.traceSampled = true
193  }
194  return { table: tableNameOf(logId, tablePrefix), json }
195}
196
197// The labels field that holds, as one JSON object, every label the table
198// has no field of its own for. setup.sh adds it to every table as a JSON
199// field; insertAll takes a JSON field's value as JSON text.
200export const EXTRA_LABELS_FIELD = 'extra_labels'
201
202/**
203 * json fitted to the table's columns (fields, from tables.get): labels the
204 * table has no field for go into labels.extra_labels as JSON text, or are
205 * dropped when the table lacks that field too; resource labels it lacks are
206 * dropped. Also returns the names of the fields it dropped.
207 */
208export function fitToTable(json, fields) {
209  const subfields = (record, name) => record?.find((f) => f.name === name)?.fields ?? []
210  const labelNames = new Set(subfields(fields, 'labels').map((f) => f.name))
211  const resourceNames = new Set(subfields(subfields(fields, 'resource'), 'labels').map((f) => f.name))
212  const hasExtra = labelNames.delete(EXTRA_LABELS_FIELD)
213  const dropped = []
214  const labels = {}
215  const extra = {}
216  for (const [name, value] of Object.entries(json.labels ?? {})) {
217    if (labelNames.has(name)) labels[name] = value
218    else if (hasExtra) extra[name] = value
219    else dropped.push('labels.' + name)
220  }
221  if (Object.keys(extra).length > 0) labels[EXTRA_LABELS_FIELD] = JSON.stringify(extra)
222  const resourceLabels = {}
223  for (const [name, value] of Object.entries(json.resource?.labels ?? {})) {
224    if (resourceNames.has(name)) resourceLabels[name] = value
225    else dropped.push('resource.labels.' + name)
226  }
227  return { json: { ...json, labels, resource: { ...json.resource, labels: resourceLabels } }, dropped }
228}
229
230/**
231 * Removes and returns items { table, insertId, json, bytes } from the queue:
232 * those for the first item's table, at most maxRows and maxBytes, but always
233 * at least one.
234 */
235export function takeBatch(queue, maxRows = MAX_BATCH_ROWS, maxBytes = MAX_BATCH_BYTES) {
236  if (queue.length === 0) return []
237  const table = queue[0].table
238  const taken = []
239  let bytes = 0
240  for (let i = 0; i < queue.length && taken.length < maxRows; ) {
241    const item = queue[i]
242    if (item.table !== table) {
243      i++
244      continue
245    }
246    if (bytes + item.bytes > maxBytes && taken.length > 0) break
247    bytes += item.bytes
248    taken.push(item)
249    queue.splice(i, 1)
250  }
251  return taken
252}
253
254/** The insertAll request body for a batch. */
255export function insertAllBody(batch) {
256  return JSON.stringify({
257    kind: 'bigquery#tableDataInsertAllRequest',
258    skipInvalidRows: false,
259    // A backstop: fields the table lacks are dropped rather than refusing
260    // the row. The mod normally removes them first, and says which.
261    ignoreUnknownValues: true,
262    rows: batch.map((item) => ({ insertId: item.insertId, json: item.json })),
263  })
264}
265
266// BigQuery error reasons that mean "try the same request again later".
267// https://cloud.google.com/bigquery/docs/error-messages
268const RETRYABLE_REASONS = new Set([
269  'backendError',
270  'internalError',
271  'rateLimitExceeded',
272  'quotaExceeded',
273  'timeout',
274  'stopped',
275])
276
277/**
278 * Sorts a sent batch by BigQuery's answer into { sent, retry, dropped },
279 * with the reason for any failure. retryLater is true when the retried
280 * rows should wait for a backoff before the next attempt.
281 */
282export function outcomeOf(batch, status, text) {
283  let body
284  try {
285    body = JSON.parse(text)
286  } catch {
287    body = undefined
288  }
289
290  if (status >= 200 && status < 300) {
291    const errors = Array.isArray(body?.insertErrors) ? body.insertErrors : []
292    if (errors.length === 0) return { sent: batch, retry: [], dropped: [], reason: '', retryLater: false }
293    const failed = new Map()
294    for (const error of errors) failed.set(error.index, error.errors ?? [])
295    const sent = []
296    const retry = []
297    const dropped = []
298    const reasons = new Set()
299    batch.forEach((item, index) => {
300      const rowErrors = failed.get(index)
301      if (rowErrors === undefined) sent.push(item)
302      else if (rowErrors.length > 0 && rowErrors.every((e) => RETRYABLE_REASONS.has(e.reason))) retry.push(item)
303      else {
304        dropped.push(item)
305        for (const e of rowErrors) reasons.add(describeError(e))
306      }
307    })
308    return { sent, retry, dropped, reason: [...reasons].join('; '), retryLater: false }
309  }
310
311  const error = body?.error
312  const reasons = (error?.errors ?? []).map((e) => e.reason)
313  const message = error?.message ?? (typeof text === 'string' ? text.slice(0, 200) : '')
314  // 403 too: a missing permission can be granted later, so the rows wait
315  // (with backoff, up to the in-memory limit) instead of being dropped.
316  const isRetryable =
317    status === 401 || status === 403 || status === 408 || status === 429 || status >= 500 ||
318    reasons.some((r) => RETRYABLE_REASONS.has(r))
319  if (isRetryable) {
320    return { sent: [], retry: batch, dropped: [], reason: 'HTTP ' + status + ': ' + message, retryLater: true }
321  }
322  return { sent: [], retry: [], dropped: batch, reason: 'HTTP ' + status + ': ' + message, retryLater: false }
323}
324
325function describeError(error) {
326  const where = error.location ? ' (' + error.location + ')' : ''
327  return (error.reason ?? 'error') + where + ': ' + (error.message ?? '')
328}
329
330/** How long to wait before the next attempt after `failures` failures in a row. */
331export function backoffMs(failures, random = Math.random()) {
332  const base = Math.min(300_000, 2_000 * 2 ** Math.max(0, failures - 1))
333  return Math.round(base * (0.5 + random / 2))
334}
335