mirror of
https://github.com/stablyai/orca.git
synced 2026-10-30 16:47:34 +00:00
feat(relay): log the director instance's TCP counters on each runtime line (#34027)
Log-only, Cloud Run director only. Each runtime line gains per-window deltas of the instance's /proc/net/snmp TCP counters (segments in and out, retransmits, errors, resets), so a stall can be checked against packet re-sends. Unreadable or reset windows are omitted.
This commit is contained in:
@@ -15,6 +15,7 @@ import {
|
||||
} from './cell-flags.js'
|
||||
import { startControlFlagChannel } from './relay-control-flag-channel.js'
|
||||
import { readCloudRunInstanceZone } from './cloud-run-instance-zone.js'
|
||||
import { readSandboxTcpCounters } from './sandbox-tcp-counters.js'
|
||||
import {
|
||||
reconcileCellAdmissionAtStartup,
|
||||
roleOwnsAssignmentMaintenance
|
||||
@@ -81,6 +82,7 @@ const {
|
||||
// Cloud Run sets K_SERVICE; GCE cells and local runs skip the metadata read.
|
||||
if (process.env.K_SERVICE) {
|
||||
void readCloudRunInstanceZone().then((zone) => observability.setInstanceZone(zone))
|
||||
observability.setTcpCounterReader(() => readSandboxTcpCounters())
|
||||
}
|
||||
// Only assignment-maintenance owner reaps credentials; readers already enforce expiry, so cell-wide copies multiplied scans without changing answers.
|
||||
const cleanupTimer = roleOwnsAssignmentMaintenance(config.role)
|
||||
|
||||
@@ -345,6 +345,32 @@ describe('relay observability', () => {
|
||||
expect(entries[1]).toMatchObject({ instanceZone: 'us-central1-a' })
|
||||
})
|
||||
|
||||
it('adds the sandbox TCP counter deltas of each window once a reader is set', () => {
|
||||
const entries: Record<string, unknown>[] = []
|
||||
const observability = new RelayObservability(
|
||||
{ role: 'director', cellId: 'director', region: 'us-central1' },
|
||||
(entry) => entries.push(entry)
|
||||
)
|
||||
const reads = [10, 13, 13, undefined, 20, 2].map((retransSegs) =>
|
||||
retransSegs === undefined
|
||||
? undefined
|
||||
: { inSegs: retransSegs * 100, outSegs: 0, retransSegs, inErrs: 0, outRsts: 0, estabResets: 0 }
|
||||
)
|
||||
observability.flush(counts)
|
||||
observability.setTcpCounterReader(() => reads.shift())
|
||||
for (let window = 0; window < 5; window++) {
|
||||
observability.flush(counts)
|
||||
}
|
||||
|
||||
expect(entries[0]).not.toHaveProperty('tcpRetransSegsDelta')
|
||||
expect(entries[1]).toMatchObject({ tcpRetransSegsDelta: 3, tcpInSegsDelta: 300 })
|
||||
expect(entries[2]).toMatchObject({ tcpRetransSegsDelta: 0 })
|
||||
// An unreadable window, the window after it, and a counter reset carry no delta.
|
||||
expect(entries[3]).not.toHaveProperty('tcpRetransSegsDelta')
|
||||
expect(entries[4]).not.toHaveProperty('tcpRetransSegsDelta')
|
||||
expect(entries[5]).not.toHaveProperty('tcpRetransSegsDelta')
|
||||
})
|
||||
|
||||
it('bounds the per-connection sticky summary', () => {
|
||||
const entries: Array<Record<string, unknown>> = []
|
||||
const observability = new RelayObservability(
|
||||
|
||||
@@ -6,6 +6,7 @@ import type { CellInventoryHoldCounts } from './cell-inventory-hold-samples.js'
|
||||
import type { PostgresPoolPressureCounts } from './postgres-pool-pressure.js'
|
||||
import type { RelayReadinessGraceEvent, RelayReadinessObservation } from './relay-readiness.js'
|
||||
import { RELAY_FIX_LEVEL } from './relay-fix-level.js'
|
||||
import { sandboxTcpCounterDeltas, type SandboxTcpCounters } from './sandbox-tcp-counters.js'
|
||||
import type { StickySlotSample } from './sticky-slot-statements.js'
|
||||
|
||||
export type RelayRuntimeCounts = {
|
||||
@@ -380,6 +381,8 @@ export class RelayObservability implements RelayRuntimeObserver {
|
||||
private timer: ReturnType<typeof setInterval> | null = null
|
||||
private lastFlushAt = 0
|
||||
private instanceZone: string | undefined
|
||||
private readTcpCounters: (() => SandboxTcpCounters | undefined) | undefined
|
||||
private lastTcpCounters: SandboxTcpCounters | undefined
|
||||
private lastFlushedSafety = {
|
||||
sqlFailures: 0,
|
||||
reconnects: 0,
|
||||
@@ -395,6 +398,19 @@ export class RelayObservability implements RelayRuntimeObserver {
|
||||
this.instanceZone = zone
|
||||
}
|
||||
|
||||
// TCP retransmits per window, to tell a lossy network path from a slow server.
|
||||
setTcpCounterReader(read: () => SandboxTcpCounters | undefined): void {
|
||||
this.readTcpCounters = read
|
||||
this.lastTcpCounters = read()
|
||||
}
|
||||
|
||||
private tcpCounterFields(): Record<string, number> {
|
||||
const current = this.readTcpCounters?.()
|
||||
const previous = this.lastTcpCounters
|
||||
this.lastTcpCounters = current
|
||||
return (current && previous && sandboxTcpCounterDeltas(previous, current)) || {}
|
||||
}
|
||||
|
||||
recordAuth(success: boolean): void {
|
||||
if (success) this.deltas.authSuccesses++
|
||||
else this.deltas.authFailures++
|
||||
@@ -687,6 +703,7 @@ export class RelayObservability implements RelayRuntimeObserver {
|
||||
region: this.identity.region,
|
||||
// Which zone this instance runs in; read once at boot, absent until then or off Cloud Run.
|
||||
...(this.instanceZone === undefined ? {} : { instanceZone: this.instanceZone }),
|
||||
...this.tcpCounterFields(),
|
||||
...counts,
|
||||
forwardedBytesDelta: deltas.forwardedBytes,
|
||||
authSuccessesDelta: deltas.authSuccesses,
|
||||
|
||||
@@ -0,0 +1,56 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import {
|
||||
parseProcNetSnmpTcp,
|
||||
readSandboxTcpCounters,
|
||||
sandboxTcpCounterDeltas
|
||||
} from './sandbox-tcp-counters.js'
|
||||
|
||||
// Tcp lines in /proc/net/snmp's layout, with counter values a Cloud Run gen1 (gVisor) sandbox reported on 2026-10-26.
|
||||
const GEN1_SNMP = [
|
||||
'Ip: Forwarding DefaultTTL InReceives',
|
||||
'Ip: 1 64 2000',
|
||||
'Tcp: RtoAlgorithm RtoMin RtoMax MaxConn ActiveOpens PassiveOpens AttemptFails EstabResets CurrEstab InSegs OutSegs RetransSegs InErrs OutRsts InCsumErrors',
|
||||
'Tcp: 1 200 120000 -1 5 3 0 0 2 485 212 1 0 122 0',
|
||||
'Udp: InDatagrams NoPorts',
|
||||
'Udp: 4 0'
|
||||
].join('\n')
|
||||
|
||||
const counters = (retransSegs: number, inSegs = 485) => ({
|
||||
inSegs,
|
||||
outSegs: 212,
|
||||
retransSegs,
|
||||
inErrs: 0,
|
||||
outRsts: 122,
|
||||
estabResets: 0
|
||||
})
|
||||
|
||||
describe('sandbox TCP counters', () => {
|
||||
it('reads the Tcp counters of /proc/net/snmp', () => {
|
||||
expect(parseProcNetSnmpTcp(GEN1_SNMP)).toEqual(counters(1))
|
||||
})
|
||||
|
||||
it('reads nothing from a file without both Tcp lines or with a non-numeric counter', () => {
|
||||
expect(parseProcNetSnmpTcp('Ip: Forwarding\nIp: 1')).toBeUndefined()
|
||||
expect(parseProcNetSnmpTcp(GEN1_SNMP.replace(' 485 ', ' x '))).toBeUndefined()
|
||||
})
|
||||
|
||||
it('reads nothing where the file is absent, as off Linux', () => {
|
||||
expect(
|
||||
readSandboxTcpCounters(() => {
|
||||
throw new Error('ENOENT')
|
||||
})
|
||||
).toBeUndefined()
|
||||
})
|
||||
|
||||
it('reports per-window deltas, and none across a counter reset', () => {
|
||||
expect(sandboxTcpCounterDeltas(counters(1), counters(4, 900))).toEqual({
|
||||
tcpInSegsDelta: 415,
|
||||
tcpOutSegsDelta: 0,
|
||||
tcpRetransSegsDelta: 3,
|
||||
tcpInErrsDelta: 0,
|
||||
tcpOutRstsDelta: 0,
|
||||
tcpEstabResetsDelta: 0
|
||||
})
|
||||
expect(sandboxTcpCounterDeltas(counters(4), counters(1))).toBeUndefined()
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,62 @@
|
||||
import { readFileSync } from 'node:fs'
|
||||
|
||||
export type SandboxTcpCounters = {
|
||||
inSegs: number
|
||||
outSegs: number
|
||||
retransSegs: number
|
||||
inErrs: number
|
||||
outRsts: number
|
||||
estabResets: number
|
||||
}
|
||||
|
||||
// The two `Tcp:` lines of /proc/net/snmp: field names, then values.
|
||||
export function parseProcNetSnmpTcp(text: string): SandboxTcpCounters | undefined {
|
||||
const [names, values] = text
|
||||
.split('\n')
|
||||
.filter((line) => line.startsWith('Tcp:'))
|
||||
.map((line) => line.trim().split(/\s+/).slice(1))
|
||||
if (!names || !values || names.length !== values.length) {
|
||||
return undefined
|
||||
}
|
||||
const field = (name: string): number => {
|
||||
const value = Number(values[names.indexOf(name)])
|
||||
return Number.isSafeInteger(value) && value >= 0 ? value : Number.NaN
|
||||
}
|
||||
const counters = {
|
||||
inSegs: field('InSegs'),
|
||||
outSegs: field('OutSegs'),
|
||||
retransSegs: field('RetransSegs'),
|
||||
inErrs: field('InErrs'),
|
||||
outRsts: field('OutRsts'),
|
||||
estabResets: field('EstabResets')
|
||||
}
|
||||
return Object.values(counters).every(Number.isFinite) ? counters : undefined
|
||||
}
|
||||
|
||||
// Cloud Run runs every container of an instance on one network stack (checked on gen1, 10-26), so
|
||||
// this counts the Cloud SQL proxy sidecar's flows to Postgres as well as the relay's own.
|
||||
export function readSandboxTcpCounters(
|
||||
read: () => string = () => readFileSync('/proc/net/snmp', 'utf8')
|
||||
): SandboxTcpCounters | undefined {
|
||||
try {
|
||||
return parseProcNetSnmpTcp(read())
|
||||
} catch {
|
||||
return undefined
|
||||
}
|
||||
}
|
||||
|
||||
export function sandboxTcpCounterDeltas(
|
||||
previous: SandboxTcpCounters,
|
||||
current: SandboxTcpCounters
|
||||
): Record<string, number> | undefined {
|
||||
const deltas = {
|
||||
tcpInSegsDelta: current.inSegs - previous.inSegs,
|
||||
tcpOutSegsDelta: current.outSegs - previous.outSegs,
|
||||
tcpRetransSegsDelta: current.retransSegs - previous.retransSegs,
|
||||
tcpInErrsDelta: current.inErrs - previous.inErrs,
|
||||
tcpOutRstsDelta: current.outRsts - previous.outRsts,
|
||||
tcpEstabResetsDelta: current.estabResets - previous.estabResets
|
||||
}
|
||||
// A counter that went backwards was reset; that window has no honest delta.
|
||||
return Object.values(deltas).every((delta) => delta >= 0) ? deltas : undefined
|
||||
}
|
||||
Reference in New Issue
Block a user