Elasticsearch ML Jobs: Automatische Inventarisatie, Analyse en Herstel met Python

Machine Learning jobs in Elasticsearch zijn krachtig voor anomaly detection, maar in productieomgevingen met tientallen jobs kan het beheer ervan complex worden. Jobs falen door memory-limieten, datafeeds raken geblokkeerd, en zonder monitoring loop je het risico dat detectieregels stilzwijgend stoppen met werken. In dit artikel beschrijf ik hoe ik een Python-tool heb gebouwd die het volledige ML-landschap automatisch inventariseert, analyseert en herstelt.

Het probleem

In een Elastic Cloud Enterprise (ECE) omgeving met 44 ML anomaly detection jobs liep ik tegen een combinatie van problemen aan. Jobs in failed state door memory hard limits, datafeeds die gestart waren maar geen data konden verwerken omdat de onderliggende job was gefaald, en actieve jobs die tegen hun memory-limiet aanzaten en elk moment konden omvallen. Kibana toont deze informatie verspreid over verschillende schermen, wat het lastig maakt om een totaaloverzicht te krijgen en de juiste volgorde van herstelacties te bepalen.

De handmatige aanpak — per job de status bekijken, memory-limieten berekenen, datafeeds stoppen, jobs sluiten en heropenen — is tijdrovend en foutgevoelig. Bovendien moet je rekening houden met de belasting op je ML nodes: als je alle jobs tegelijk herstart, creëer je een piek die de hele cluster kan beïnvloeden.

De aanpak: volledige auto-discovery

Het doel was een script dat niets hardcoded heeft. Geen lijsten van job-namen, geen vaste node-ID’s, geen vooraf bepaalde memory-limieten. Het script moet zelf het complete plaatje opbouwen door de Elasticsearch API’s te bevragen. Dit maakt het herbruikbaar op elk cluster zonder aanpassingen.

De tool bestaat uit drie kerncomponenten: een MLAnalyzer die het cluster inventariseert en elk probleem classificeert, een MLNodeMonitor die de actuele belasting van ML nodes bewaakt, en een JobRestorer die het daadwerkelijke herstel uitvoert met load-aware pacing.

Stap 1: ML infrastructuur ontdekken

De eerste stap is het identificeren van alle ML nodes in het cluster. Niet elke node heeft de ml role, en je wilt weten hoeveel capaciteit er beschikbaar is voordat je jobs gaat herstarten.

nodes_info = es.nodes.info(node_id="_all", metric=["os", "jvm", "settings"])
nodes_stats = es.nodes.stats(node_id="_all", metric=["os", "jvm"])

for node_id, info in nodes_info["nodes"].items():
    roles = info.get("roles", [])
    if "ml" not in roles:
        continue

    stats = nodes_stats["nodes"].get(node_id, {})
    ml_node = MLNodeInfo(
        node_id=node_id,
        name=info.get("name", "unknown"),
        memory_total_bytes=info["os"]["mem"]["total_in_bytes"],
        cpu_percent=stats["os"]["cpu"]["percent"],
        jvm_heap_used_percent=stats["jvm"]["mem"]["heap_used_percent"],
    )

Dit levert een compleet overzicht op van de ML-infrastructuur, inclusief RAM, CPU-belasting en heap-gebruik per node. In ons geval: twee nodes met elk 4GB RAM.

Stap 2: Jobs en datafeeds inventariseren

Vervolgens halen we alle ML anomaly detection jobs en hun datafeeds op. Door de _stats API te gebruiken in plaats van de configuratie-API krijgen we direct de runtime-informatie: huidige state, memory-gebruik, verwerkte records en eventuele foutmeldingen.

jobs_response = es.ml.get_job_stats(job_id="_all")
datafeeds_response = es.ml.get_datafeed_stats(datafeed_id="_all")

# Bouw een lookup van datafeed state per job
datafeed_map = {}
for df in datafeeds_response.get("datafeeds", []):
    job_id = df.get("timing_stats", {}).get("job_id") or df["datafeed_id"].replace("datafeed-", "")
    datafeed_map[job_id] = df

Stap 3: Probleem classificatie

Dit is het hart van de analyse. Elke job wordt geclassificeerd in een van zes categorieën, elk met een eigen herstelstrategie en prioriteit.

Geblokkeerde datafeeds (prioriteit 1) zijn het meest urgent: de datafeed draait, maar de job is gefaald. Elasticsearch probeert continu data te verwerken die nergens heen kan. Dit verspilt resources en genereert ruis in de logs. Herstel: stop de datafeed, sluit de job, heropen en start opnieuw.

Preventieve gevallen (prioriteit 2) zijn actieve jobs die tegen hun memory-limiet aanlopen. Het veld bucket_allocation_failures is hiervoor de indicator: als dit boven nul staat, heeft het model geprobeerd meer geheugen te alloceren dan beschikbaar. De job werkt nog, maar kan elk moment falen. Hier is het zaak om proactief de limiet te verhogen.

Failed jobs met memory issues (prioriteit 10) hebben hun hard limit bereikt. De memory_status staat op hard_limit en de job is gestopt met verwerken. Herstel vereist een memory-update vóór het heropenen.

Failed jobs zonder memory issues (prioriteit 15) zijn gefaald om andere redenen: netwerk timeouts, index-problemen, of configuratiefouten. Vaak is een simpele close-open cyclus voldoende.

Stap 4: Memory-aanbevelingen berekenen

Voor jobs met memory-problemen berekent het script automatisch een veilige nieuwe limiet. De formule neemt het maximum van het huidige model en de historische piek, vermenigvuldigt dat met 1.8 voor headroom, en rondt af naar boven in stappen van 5MB.

base_bytes = max(model_bytes, peak_model_bytes)
recommended_bytes = int(base_bytes * 1.8)
recommended_mb = math.ceil(recommended_bytes / (1024 * 1024))
# Afronden naar boven op 5MB stappen
recommended_mb = math.ceil(recommended_mb / 5) * 5
recommended_limit = f"{recommended_mb}mb"

De factor 1.8 biedt voldoende ruimte voor groei van het model zonder overdreven veel geheugen te reserveren. Een job met een model van 29MB en een limiet van 30MB (slechts 3% headroom) krijgt zo een aanbeveling van 65MB — ruim genoeg om niet bij de eerstvolgende piek om te vallen.

Stap 5: Load-aware herstel

Het herstellen van ML jobs is niet simpelweg alles tegelijk herstarten. Elke job die opent, begint direct data te verwerken en verbruikt CPU en geheugen. Op een cluster met twee ML nodes van 4GB wil je dat gecontroleerd doen.

De JobRestorer volgt per probleemtype een specifieke herstelsequentie. Voor een geblokkeerde datafeed is dat: stop datafeed → close job → open job → start datafeed. Voor een preventief geval met een actieve job: stop datafeed → close job → update memory limit → open job → start datafeed. Het verschil is subtiel maar belangrijk: je kunt de memory-limiet alleen wijzigen op een gesloten job.

# Update memory limit (alleen mogelijk op gesloten job)
es.ml.update_job(
    job_id=job.job_id,
    body={
        "analysis_limits": {
            "model_memory_limit": job.recommended_memory_limit
        }
    }
)

# Heropen met timeout
es.ml.open_job(job_id=job.job_id, timeout="30s")

# Start datafeed
es.ml.start_datafeed(datafeed_id=f"datafeed-{job.job_id}", timeout="60s")

Tussen elke job wacht het script een configureerbare delay (standaard 30 seconden) en controleert het de CPU-belasting van alle ML nodes. Als de CPU boven de drempel komt (standaard 80%), pauzeert het script automatisch tot er weer capaciteit is. Dit voorkomt dat het herstelproces zelf de cluster destabiliseert.

Het rapport

Naast het herstel genereert de tool een compleet Markdown-rapport met een overzicht van de ML-infrastructuur, een samenvatting per categorie, gedetailleerde tabellen met memory-metrics, een lijst van performance hotspots (jobs met de hoogste search time), en een geprioriteerd herstelplan.

Een voorbeeld van de samenvatting uit een productieomgeving:

| Categorie                     | Aantal |
|-------------------------------|--------|
| ✅ Gezond                      | 27     |
| 🔴 Failed (memory)             | 1      |
| 🟠 Failed (overig)             | 4      |
| 🟡 Geblokkeerde datafeed       | 1      |
| ⚠️ Preventief (memory druk)    | 2      |
| ⏹️ Nooit data ontvangen        | 9      |
| Totaal                        | 44     |
| Te herstellen                 | 8      |

De “nooit data ontvangen” categorie is ook waardevol: in dit geval bleken 9 Linux-specifieke jobs (endpoint detection, auditd-analyse) al maandenlang geen data te ontvangen omdat er geen Linux agents in het cluster aanwezig waren. Die jobs verbruiken weliswaar geen resources als ze gesloten zijn, maar vervuilen wel het overzicht en kunnen verwijderd worden.

Credential management met connections.py

Het script maakt gebruik van een centrale connections.py module voor het beheer van verbindingen naar enterprise services. Deze module slaat credentials versleuteld op met Fernet-encryptie, waardoor je nooit wachtwoorden hardcoded in scripts hoeft te hebben. De encryptiesleutel en het credential-bestand staan buiten de projectdirectory om accidentele commits naar versiebeheer te voorkomen.

# Het hele connectiebeheer in één regel
from connections import get_es_connection
es = get_es_connection()

Dit patroon maakt scripts herbruikbaar: dezelfde connections.py module kan door verschillende tools en projecten worden gebruikt, elk met hun eigen Elasticsearch-clusters, Active Directory-domeinen of Confluence-instanties.

Gebruik in de praktijk

Het script biedt verschillende modes voor verschillende situaties:

# Alleen inventariseren en rapport genereren
python3 ml_check.py --analyse-only --report ml-rapport.md

# Dry-run: toon wat er zou gebeuren zonder iets te wijzigen
python3 ml_check.py --dry-run

# Volledig automatisch herstel
python3 ml_check.py --auto-fix --delay 30

# Alleen geblokkeerde datafeeds herstellen
python3 ml_check.py --auto-fix --group blocked

# Voorzichtig: langere delay, lagere CPU drempel
python3 ml_check.py --auto-fix --delay 60 --max-cpu 70

De --analyse-only mode is bijzonder nuttig als periodieke gezondheidscheck. Je kunt het script wekelijks draaien om vroegtijdig memory-druk of stille failures op te sporen, voordat ze impact hebben op je security monitoring.

Lessen uit de praktijk

Tijdens het bouwen en testen van deze tool heb ik een aantal zaken geleerd die niet direct uit de Elasticsearch-documentatie blijken.

CPU-metrics op kleine ML nodes zijn misleidend. Een ML node die 99% CPU rapporteert maar 0 open jobs heeft, is niet noodzakelijk overbelast. Bij containers met minder dan 1 vCPU (in ons geval 0.79 allocated processors) kan de OS-level CPU-metric onbetrouwbaar zijn. De _nodes/hot_threads API geeft een veel betrouwbaarder beeld van wat een node daadwerkelijk doet.

Job-distributie lost zichzelf op. Een scheve verdeling van jobs over ML nodes (27 jobs op node A, 0 op node B) betekent niet dat je handmatig moet herverdelen. Wanneer je jobs herstart, kiest Elasticsearch automatisch de node met de meeste vrije capaciteit. Na het herstellen van de 8 problematische jobs zagen we een betere verdeling ontstaan.

De factor 1.8 voor memory headroom is een goede balans. Te weinig headroom (onder 1.3) leidt tot herhaalde failures bij pieken. Te veel (boven 2.5) verspilt schaars ML-geheugen. De 1.8 factor met afronding naar 5MB stappen biedt voldoende ruimte voor modelgroei zonder overallocatie.

Preventief ingrijpen is waardevoller dan reactief herstellen. De twee jobs die we als “preventief” identificeerden — actief maar met allocation failures — zouden zonder interventie binnen dagen gefaald zijn. Door ze proactief te behandelen voorkomen we gaten in de anomaly detection coverage.

Takeaway

Elasticsearch ML anomaly detection is een essentieel onderdeel van security monitoring, maar vereist actief beheer. Handmatige controles schalen niet mee met het aantal jobs, en Kibana biedt onvoldoende overzicht voor operationele beslissingen. Een geautomatiseerde tool die inventariseert, analyseert en load-aware herstelt, maakt het verschil tussen reactief brandjes blussen en proactief beheer. De combinatie van auto-discovery (geen hardcoded waarden), intelligente classificatie en gecontroleerd herstel met load-monitoring maakt dit script inzetbaar op elk Elasticsearch-cluster zonder aanpassingen.

Hulp nodig met Elasticsearch ML Jobs?

Bij Hasecon hebben we uitgebreide ervaring met het beheren en optimaliseren van Elasticsearch ML jobs in productieomgevingen. Of het nu gaat om anomaly detection voor security monitoring of het debuggen van falende datafeeds — wij helpen u verder.

Bekijk onze Elasticsearch & SIEM diensten →