Veel Onderzoek, Weinig Bandbreedte

Een onderzoek naar een netwerkmysterie binnen het wereldwijde compute-cluster van HRT.

Naarmate onze compute-capaciteit zich wereldwijd verspreidt, versturen we voortdurend data van de ene locatie naar de andere. We hebben een systeem gebouwd dat servicetelemetrie combineert met eBPF-tracing en byte-tellers van netwerkhardware om inzichtelijk te maken welke gebruikers en programma's de meeste traffic genereren.

Introductie

Het trading-proces van HRT wordt ondersteund door een enorme hoeveelheid research-computatie. We trainen een diverse set modellen op basis van jaren aan historische data van beurzen over de hele wereld, om vervolgens hun gedrag te simuleren in talloze trading-scenario's. Vanuit een systeemperspectief betekent dit dat veel CPU's en GPU's data verwerken van vele schijven, met sterk variërende toegangspatronen en veel netwerkcommunicatie tussen machines.

Lange tijd vond het grootste deel van dit werk plaats binnen één locatie. Daarna hebben we jaren besteed aan het wereldwijd uitbreiden van onze opslag en compute, inclusief naar verre locaties zoals Noorwegen. Bijgevolg werden programma's steeds vaker uitgevoerd op grote afstand van de data die ze nodig hadden, waardoor WAN-bandbreedte een belangrijke beperkende factor werd. Wanneer het inter-site verkeer onvermijdelijk het netwerk overbelastte, ontdekten we dat we over zwakke observeerbaarheid beschikten: het was te moeilijk om vast te stellen welke gebruikers en workflows verantwoordelijk waren.

Een Trage Dag

Er deed zich een incident voor waarbij een model-trainingsrun te lang duurde. Dit programma gebruikte GPU's in Chicago om data uit New York te analyseren. Onze cross-region data-transfers verlopen via een speciale service genaamd DupeDog; zo werd de data naar Chicago gespiegeld voor de training. Deze stap stagneerde, dus we controleerden ons dashboard, waarop duidelijk te zien was dat onze transfersnelheid was ingestort.

Gedeelde Capaciteit

Dit is niet het persoonlijke cluster van ons model. Op elk gegeven moment is DupeDog grote hoeveelheden data aan het "dupen" voor gebruikers in het hele bedrijf. Het lijkt dus voor de hand te liggen wat er mis is: andere mensen zijn gestart met transfers en we moesten de capaciteit delen. Door alle andere gebruikers aan de grafiek toe te voegen, zouden we het antwoord hebben.

We zagen inderdaad andere DupeDog-gebruikers die de bandbreedte deelden, maar er verscheen niets nieuws op het moment dat de vertraging begon (10:25). De transfers van iedereen vertraagden plotseling. De DupeDog-logs gaven geen antwoord. We besloten daarom te kijken naar de ruwe netwerkswitch-tellers om te zien hoeveel bytes er de locatie binnenkwamen, ongeacht de methode.

De resultaten toonden aan dat een onbekende bron van traffic de netwerkcapaciteit naar Chicago volledig verzadigde. Wat dit ook veroorzaakte, het was geen DupeDog.

Meer dan Eén Weg naar Binnen

Hoewel DupeDog de "standaard" manier is om bestanden tussen sites te verplaatsen met een eerlijke verdeling, is het niet verrassend dat dit niet alles dekt. We geven onderzoekers veel vrijheid in de keuze van hun tools en het bouwen van hun eigen workflows, zodat ze snel kunnen itereren op ideeën. Hoewel het mooi zou zijn als 100% van het WAN-verkeer via één service zou lopen, is dat niet pragmatisch gezien de diversiteit aan redenen waarom data wordt verplaatst.

Bijgevolg hebben we meerdere services die het netwerk delen. Een belangrijk voorbeeld is onze distributed file cache, Cashy, die soms het WAN gebruikt om data uit de cache van een andere site op te halen. Wanneer we Cashy aan de grafiek toevoegden, bleek dat ook niet de boosdoener te zijn.

Nu hadden we een probleem. Services zoals DupeDog en Cashy hebben de eigenschap dat we verzoeken kunnen loggen met rijke informatie over gebruikers en workflows. Maar deze methode dekt alleen verkeer via die services. Onze machines kunnen op veel andere manieren communiceren. We konden te maken hebben met socket-communicatie tussen jobs, storage cross-mounts, iemand die iets via scp verplaatst, een nieuwe service die we zijn vergeten te loggen, of een custom webserver van een gebruiker.

Hoe kunnen we bij willekeurig netwerkverkeer de gebruikers en programma's koppelen aan hun WAN-verbruik?

Uitgebreide Capturing met eBPF

Een service zoals DupeDog heeft een RPC-methode die grofweg als volgt is gestructureerd:

void dupe(data=myData, to=there, ...) {
  // Eerst exact vastleggen wat we gevraagd is te doen.
  log(user=..., program=...,
      from=here, to=there,
      data=myData, bytes=myData.size(), ...);

  // Daarna daadwerkelijk de data kopiëren.
  int sock = socket(AF_INET, SOCK_STREAM, 0);
  connect(sock, ...);
  send(sock, ...);
  ...
}

Het cruciale punt is dat we een gelogd event krijgen zodra de service iets doet. Maar wat doen we als een gebruiker direct zijn eigen networking-code schrijft? Hun custom socket-code slaat onze logging over, maar gaat nog steeds door de networking-stack van de Linux-kernel. We hebben een manier nodig om direct in te haken op kernel-netwerkfuncties om dit gebruik te detecteren.

Gelukkig biedt eBPF dat mechanisme.

Vroeg of Laat: tcp_recvmsg (meestal)

Gebruikers programmeren in verschillende talen en gebruiken diverse bibliotheken en functies om over het netwerk te communiceren. Bij het ontvangen van data via TCP kunnen ze indirect een Linux-syscall aanroepen zoals read, readv, recvfrom, of recvmsg. Na enkele socket- en protocol-dispatch-stappen in de kernel, komen de codepaden over het algemeen terecht in een functie genaamd tcp_recvmsg.

eBPF stelt ons in staat om kleine hook-programma's te bevestigen aan de ingang of uitgang van de meeste kernelfuncties. Deze hooks draaien in de kernel met minimale overhead. In het kort ziet dat er zo uit:

SEC("fexit/tcp_recvmsg")
int BPF_PROG(
    fexit__tcp_recvmsg, struct sock *sk, struct msghdr *msg,
    size_t len, int flags, int *addr_len, int ret) {
  if (ret > 0) {
    __u32 pid = bpf_get_current_pid_tgid() >> 32;
    // dit voorbeeld gaat uit van IPv4
    __be32 remote_ip = sk->__sk_common.skc_daddr;
    bpf_printk(
        "Hello from tcp_recvmsg! PID %u read %d bytes from %pI4",
        pid, ret, &remote_ip);
  }
  return 0;
}

Bovenstaande code zou een bericht in de kernel tracing buffer printen telkens wanneer tcp_recvmsg een positief aantal bytes retourneert. Cruciaal is dat we ook de PID krijgen, waardoor we het verkeer aan een gebruiker kunnen toeschrijven!

Geen Paniek! De Kernel Doet Dat Ook Niet

Hoewel het nuttig klinkt om willekeurige code in de Linux-kernel uit te voeren, zouden we ons dan niet zorgen moeten maken over bugs? Een off-by-one fout in user-code kan één proces laten crashen (segfault), maar een soortgelijk probleem in de kernel kan het hele systeem platleggen. Op het eerste gezicht lijkt het gebruik van eBPF daarom riskant.

Gelukkig is eBPF slimmer. Wanneer er een nieuw programma wordt geladen, gebruikt het een verifier om statisch de veiligheidseigenschappen van het programma te bewijzen. Dit betekent dat geheugentoegang wordt gecontroleerd, registers zijn geïnitialiseerd, er geen onbegrensde loops of recursie zijn, en pointers correct worden gebruikt. De verifier is erg strikt, en het is soms uitdagend om een programma te herstructureren totdat de verifier deze eigenschappen kan vaststellen.

Elke Byte Telt

Wat moeten we precies doen in onze callback? Een optie is om van elke aanroep een event te bouwen, dit naar de userspace te verplaatsen en vervolgens naar ons analyseplatform (gebaseerd op Iceberg-tabellen en ClickHouse) te sturen. Maar hebben we echt zoveel data nodig?

Ons hoofddoel is om de primaire drijvers van netwerkbenutting te zien. Dit betekent dat we alleen byte-totalen per minuut hoeven te tellen, gegroepeerd om toeschrijving aan gebruikers en workflows te vergemakkelijken. eBPF heeft een functie genaamd "BPF Maps", die datastructuren biedt die in de kernel-space leven. In de kernel kunnen we bij elke succesvolle tcp_recvmsg conceptueel het volgende doen:

key = (pid, localip, localport, remoteip, remoteport) bytes_received[key] += received

We houden aparte tellers bij voor elke CPU om contentie te voorkomen. Aan het einde van elke minuut halen we deze aantallen naar de userspace, tellen we ze op over de CPU's en combineren we ze met metadata over elk proces. Hierdoor kunnen we het verkeer toeschrijven aan gebruikers, programma's en workflows (grotere runs die over meerdere jobs verspreid zijn). Tot slot sturen we deze byte-counts naar ons analyseplatform.

(Hierbij is een aanzienlijke hoeveelheid extra code weggelaten, met name wat betreft race-condities en health-reporting. Bovendien kunnen zelfs totalen per minuut een overmatige hoeveelheid telemetrie genereren, dus downsamplen we deze records met een Priority Sampling Algorithm.)

Wacht, wie is deze gebruiker "root"?

Na het installeren van de eBPF-hooks in ons cluster kunnen we zien welke gebruikersaccounts hoeveel bytes hebben verzonden en ontvangen, van waar naar waar, en via welke programma's en workflows. Dit klinkt veelbelovend, maar we zijn nog niet klaar.

Wanneer we kijken naar de nieuwe eBPF-telemetrie van het oorspronkelijke incident, zien we het volgende:

1. eBPF is erin geslaagd het ontbrekende verkeer vast te leggen. Een groot deel van de nieuwe zichtbare activiteit vóór het incident kwam van de eric:trade_simulation workflow. Dit is waardevolle informatie, omdat dit programma direct TCP-sockets gebruikt.
2. Er zijn echter twee ernstige problemen:

• Het verkeer dat verantwoordelijk is voor de verzadiging komt van de root gebruiker, via een mysterieus kworker/*xprtiod programma.
• Hoewel eBPF de netwerkactiviteit van de DupeDog- en Cashy-services correct heeft waargenomen, wordt hun verkeer toegeschreven aan de dupedog en cashy accounts. Het laat ons niet zien wie deze transfers heeft aangevraagd.

Deze twee problemen lossen we op verschillende manieren op.

Het is Altijd NFS

Dit hadden we kunnen voorspellen. xprtiod is een kernel workqueue die door de NFS-client wordt gebruikt voor netwerk-I/O. NFS (Network File System) stelt een programma in staat om bestanden op een andere machine te benaderen via gewone filesystem-operaties. Het lezen van een bestand kan dus een netwerktransfer triggeren.

De NFS-client van de kernel deelt gecachte data en serververbindingen over processen, waarbij kernel-workers de data namens hen ontvangen. Als gevolg hiervan kan onze tcp_recvmsg eBPF-hook in een kernel-worker worden uitgevoerd, zelfs als een user-proces de leesactie initieerde. De NFS-client heeft hiermee onze eBPF-monitoring omzeild, die "correct" rapporteert dat root de data ontvangt, zonder te vertellen welke gebruiker erom vroeg.

We geven niet zomaar op. We kunnen geen volledige gebruikersattributie krijgen met alleen tcprecvmsg, maar NFS-leesacties gaan ook door de kernel. Het traceren van NFS-gebaseerde netwerkactiviteit terug naar user-processen is lastig, maar niet onmogelijk. Onze eerste poging was het haken naar vfsread, de kernelfunctie die wordt aangeroepen wanneer een proces read() aanroept. Het probleem is dat filesystems caching implementeren, waardoor een herhaalde leesactie de tweede keer een netwerktransfer kan vermijden. Dit negeren zou het werkelijke netwerkverkeer overschatten.

Uiteindelijk moesten we de low-level NFS-functies traceren die daadwerkelijk data ophalen, inclusief diegene die worden getriggerd via memory mapping of asynchrone I/O. We legden vast welk proces een verzoek initieerde en koppelden die informatie aan de resulterende datatransfers. Dit was een moeizaam proces, maar het is gelukt.

Eindelijk wisten we wat er was gebeurd. Het was, zoals verwacht, een configuratiefout. Duizenden jobs in Chicago wilden een groot databestand lezen dat in New York was opgeslagen. Normaal gesproken lossen we dit op door het bestand te laden in de Chicago-instantie van Cashy (onze distributed file cache). Het bestand verhuist dan één keer over het WAN en wordt vervolgens efficiënt geserveerd aan de compute-jobs.

In plaats daarvan waren we per ongeluk alles direct naar de bron-filer in New York gaan wijzen, die het bestand vervolgens duizenden keren over het land verstuurde. Erger nog: de lezers wilden allemaal hetzelfde bestand, waardoor de lokale cache van de NY-filer voorkwam dat disk-I/O de boel vertraagde.

Achteraf gezien was dit een simpele fout met een eenvoudige oplossing. Onze tracing hielp bij het moeilijkste deel: uitzoeken wat de fout precies was.

We Hebben de Servicelogs Nog Steeds Nodig

Veel van ons WAN-verkeer kwam van eric:tradesimulation, zowel direct via TCP-sockets als indirect via (onbedoelde) NFS-leesacties. Maar hoe zit het met dupedog:movefiles? Als we een volledig overzicht willen van al het netwerkverkeer van eric:trade_simulation, moeten we ook alle bestanden meerekenen die deze workflow aan DupeDog heeft gevraagd om te transfereren.

eBPF socket-hooks kunnen niet zien wie oorspronkelijk de bestandsoverdracht heeft aangevraagd, maar DupeDog wel. Het beste overzicht is daarom een synthese van meerdere bronnen:

• eBPF voor socket-verkeer via tcprecvmsg / tcpsendmsg.
• eBPF voor NFS-gerelateerd netwerkverbruik.
• Servicelogs voor DupeDog, Cashy en andere relevante services.
• Netwerkswitch byte-tellers, als onafhankelijk en gezaghebbend totaal om our user-attributed traffic mee te vergelijken.

Op deze manier maakt het niet uit dat eric:trade_simulation data op drie verschillende manieren verplaatst (DupeDog, NFS en custom TCP socket-code). We zijn alleen geïnteresseerd in het totaal aantal verplaatste bytes. Dit is het inzicht dat we aan het begin zochten: een volledige attributie van netwerkactiviteit aan de verantwoordelijke gebruikers en workflows.

Elke Byte Telt, maar Slechts Eén Keer

Wat gebeurt er als we een service zoals DupeDog vragen om data van NY naar Chicago te verplaatsen?

1. eBPF ziet een tcp_sendmsg op host NY123, remote host CH234. user=dupedog.
2. eBPF ziet een tcp_recvmsg op host CH234, remote host NY123. user=dupedog.
3. DupeDog-logs registreren een NY123 → CH234 transfer namens eric:trade_simulation.

Ons doel is om het totale verbruik per gebruiker en workflow te ontleden, maar we kunnen deze records niet naïef optellen, omdat we dan een meting zouden hebben die het werkelijke netwerkverbruik drie keer telt en overschrijdt.

In plaats daarvan zorgen we ervoor dat we elk type flow slechts één keer tellen. In het bovenstaande voorbeeld zouden we eBPF-data voor de dupedog gebruiker/poort uitsluiten, wetende dat dit beter wordt gerepresenteerd door de DupeDog-logs. Op dezelfde manier sluiten we poort-2049 (NFS) verkeer uit van onze eBPF TCP-dataset, omdat we dit apart vastleggen met onze NFS-hooks.

We gebruiken heatmaps om bij te houden hoe goed de user-attributed totalen overeenkomen met onze switch-tellers. Groen geeft een goede overeenkomst aan, en rood laat zien wanneer we iets missen. Zo merkten we onlangs dat we filer-to-filer replicatieverkeer niet meerekenen, omdat aan geen van beide uiteinden van de verbinding onze eBPF-recorder draait. Gelukkig hebben we aparte metrieken die dit replicatieverkeer beschrijven.

Grotendeels In Beeld

We gebruiken deze data nu constant. Als onze netwerken plotseling overbelast raken, kunnen we zien welke workflows het verkeer genereren. Wanneer we een programma naar een nieuwe locatie verplaatsen, kunnen we zien met welke data en services het communiceert en beslissen of we die ook moeten verplaatsen. We hebben ook geautomatiseerde processen om workflows te knijpen (throttle) die chaos veroorzaken. Deze telemetrie stelt ons in staat om "het verzadigen van het WAN" aan die lijst toe te voegen.