Ingestion et mode daemon

Conversion OTLP

Conception en deux passes

convert_otlp_request() traite chaque bloc resource_spans en deux passes :

Passe 1 : Construction de l'index des spans :

let span_index: HashMap<&[u8], &Span> = scope_spans.iter()
    .flat_map(|ss| &ss.spans)
    .map(|span| (span.span_id.as_slice(), span))
    .collect();

Passe 2 : Conversion des spans I/O :

for span in &scope.spans {
    if let Some(event) = convert_span(span, service_name, &span_index) {
        events.push(event);
    }
}

Pourquoi deux passes ? Dans OTLP, un span parent peut apparaître après son enfant dans le message protobuf. La première passe construit une table de recherche pour que la seconde passe puisse résoudre source.endpoint en remontant la chaîne d'ancêtres. Une approche en une seule passe manquerait les spans parents définis plus loin dans le message.

source.endpoint se résout en trois étapes, chacune servant de repli à la précédente : la route HTTP entrante la plus proche en remontant la chaîne (http.route, puis http.url, puis url.full), ensuite le cadre de code code.* pour les points d'entrée qui ne portent aucun attribut HTTP (jobs planifiés, consumers de messages), enfin le littéral "unknown". La remontée est bornée par la même limite de profondeur que celle du parcours code.*. Ne résoudre que le parent direct laissait toute pile en couches sur "unknown", puisque la route se trouve sur le span SERVER deux niveaux ou plus au-dessus d'un span JDBC feuille.

L'index utilise des clés &[u8] (octets bruts du span_id), évitant l'encodage hexadécimal juste pour la recherche. Un index est construit par service.name, couvrant tous les blocs ResourceSpans que ce service possède : le batch processor du collector découpe une même trace entre plusieurs blocs, et un index par bloc perdait l'endpoint à la frontière. Les services restent séparés pour qu'une feuille ne puisse pas adopter la route ou le cadre code.* d'un appelant dans une requête d'agrégation, ce qui nommerait un fichier d'un autre dépôt. L'index de chaque service est plafonné à 100 000 spans pour prévenir l'épuisement mémoire depuis des payloads OTLP pathologiques, si bien qu'un service bruyant ne prive pas les autres de résolution de parent. Un tracing::warn! est émis quand un cap est atteint pour aider les opérateurs à diagnostiquer une résolution de parent dégradée.

Table de recherche bytes_to_hex

fn bytes_to_hex(bytes: &[u8]) -> String {
    const HEX: &[u8; 16] = b"0123456789abcdef";
    let mut buf = Vec::with_capacity(bytes.len() * 2);
    for &b in bytes {
        buf.push(HEX[(b >> 4) as usize]);
        buf.push(HEX[(b & 0x0f) as usize]);
    }
    // Tous les octets viennent de HEX (ASCII 0-9, a-f), toujours du UTF-8 valide.
    String::from_utf8(buf).expect("hex table is ASCII")
}

C'est une optimisation bien connue pour l'encodage hexadécimal. Au lieu d'utiliser write!(hex, "{b:02x}") (qui invoque la machinerie de formatage par octet à ~30ns), la table de recherche convertit chaque octet en deux caractères hexadécimaux via décalage de bits à ~5ns par octet. Le Vec<u8> est pré-alloué et l'appel from_utf8 est infaillible puisque seuls des chiffres hexadécimaux ASCII sont insérés. Pas de unsafe nécessaire : le expect est une assertion à coût zéro sur une condition qui ne peut pas échouer.

Pour un trace_id de 16 octets + un span_id de 8 octets, cela économise ~600ns par conversion de span. À 100 000 événements/sec, c'est 60ms/sec de surcoût évité.

nanos_to_iso8601 : algorithme de Howard Hinnant

Note : cette fonction vit dans le module partagé time.rs et est réutilisée par les ingestions Jaeger et Zipkin via micros_to_iso8601.

La conversion de nanosecondes Unix vers YYYY-MM-DDTHH:MM:SS.mmmZ utilise l'algorithme de date civile de Howard Hinnant. Les étapes clés :

1. Convertir les nanosecondes en jours depuis l'epoch + millisecondes restantes
2. Décaler l'epoch au 1er mars, an 0 (en ajoutant 719 468 jours)
3. Calculer l'ère (cycle de 400 ans) et le jour de l'ère
4. Dériver l'année de l'ère, le jour de l'année, le mois et le jour en utilisant une formule sans table de recherche

Cela évite le crate chrono (~150 Ko de surcoût binaire) et son surcoût de parsing de ~200ns. L'algorithme artisanal gère correctement les années bissextiles (vérifié par un test avec 2024-02-29).

Priorité du type d'événement

Quand un span possède à la fois un attribut SQL (db.statement ou db.query.text) et un attribut HTTP (http.url ou url.full), SQL prend la priorité. C'est intentionnel : l'instrumentation de base de données est plus spécifique que l'instrumentation client HTTP. L'attribut SQL contient le texte réel de la requête nécessaire à la normalisation, tandis que l'attribut HTTP pourrait représenter la même opération au niveau transport.

Les spans RPC (rpc.system, par ex. gRPC ou Dubbo) sont examinés après SQL et HTTP : ils ne portent ni statement ni URL, leur cible est donc rpc.service/rpc.method (avec repli sur le nom du span), et ils entrent dans le pipeline comme EventType::HttpOut pour que les détecteurs topologiques et d'occurrence les traitent comme des appels sortants. Seul SpanKind::Client est admis : les clés rpc.* sont posées aussi sur le span SERVER entrant (le handler), et admettre ceux-ci doublerait chaque hop et inventerait des arêtes vers soi-même. Cette réutilisation au seul niveau de l'admission laisse inchangé le chemin de normalisation/sanitisation HTTP.

Les spans messaging (messaging.system, une seule convention pour Kafka, RabbitMQ, Pulsar, SQS, NATS et JMS) sont examinés en dernier, avec messaging.destination.name comme cible et le nom du span en repli. Contrairement au RPC, ils ont leur propre EventType::Messaging, pour deux raisons : la destination ne doit pas passer par le normaliseur de chemins HTTP, qui masquerait un identifiant de compte SQS en {id}, et une publication mérite ses propres types de findings plutôt qu'une étiquette teintée HTTP. Seul SpanKind::Producer est admis, sur la même logique de double comptage que le RPC, à quoi s'ajoute qu'un consommateur qui fait du polling noierait les détecteurs d'occurrences.

L'arête producteur vers consommateur est un span link OTel, pas une relation parent-enfant, et le consommateur démarre le plus souvent sa propre trace. resolve_producer_link trouve le span CONSUMER qui porte ce lien, lit sa première entrée et pose l'identifiant de la trace productrice sur l'événement en link_trace_id. Le filtre sur CONSUMER compte : les batch span processors et les relations follows-from émettent aussi des liens. Un lien qui revient vers la trace du span est écarté. Les deux traces ne sont jamais fusionnées, voir Limites pour les raisons.

Deux topologies, parce que les instrumentations OTel ne placent pas receive au même endroit. Certaines imbriquent le travail du handler sous le span receive, ce qu'une remontée d'ancêtres retrouve. L'instrumentation Kafka officielle OpenTelemetry Java et .NET ne fait pas cela : elle émet receive comme frère du travail qu'il déclenche, sous un parent commun.

order-consumed        INTERNAL   parent=""     links=[]
├─ orders receive     CONSUMER   parent=root   links=[<trace productrice>]
└─ postgresql         CLIENT     parent=root   links=[]        <- le span analysable

La chaîne d'ancêtres du span SQL est postgresql -> order-consumed, le CONSUMER porteur du lien n'y figure donc jamais. Une remontée d'ancêtres seule a manqué 100 % des liens sur deux corpus capturés pendant que tous les tests unitaires restaient verts, justement parce qu'ils construisaient la forme ancêtre que le code attendait. La résolution essaie donc les frères d'abord, via un index des spans CONSUMER porteurs de lien indexés par parent, puis retombe sur la chaîne d'ancêtres.

Cet index n'est construit que si la requête porte un span CONSUMER lié, et il sert aussi de garde par service : un service sans un tel span saute les deux recherches. L'indexation par parent impose une décision que la remontée d'ancêtres n'avait jamais eu à prendre, car plusieurs spans receive frères peuvent partager un parent. Le premier dans l'ordre du lot l'emporte, ce qui est déterministe mais arbitraire : rien dans les spans ne dit quel message a causé quelle requête.

Les conventions sémantiques OTel legacy (pré-1.21) et stables (1.21+) sont toutes deux supportées : db.statement et db.query.text pour le SQL, http.url et url.full pour le HTTP, http.method et http.request.method pour le verbe, http.status_code et http.response.status_code pour le statut. Cela assure la compatibilité avec les anciens SDKs OTel comme avec les agents Java modernes (v2.x).

Protection contre la dérive d'horloge

if end_nanos < start_nanos {
    tracing::trace!("Span has end_time < start_time (clock skew?), duration forced to 0");
}
let duration_us = end_nanos.saturating_sub(start_nanos) / 1000;

saturating_sub retourne 0 pour les durées négatives au lieu de boucler. Un log de niveau trace aide les opérateurs à diagnostiquer les problèmes d'intégration OTLP sans inonder les logs.

Le trait MetricsSink (ingest/otlp/mod.rs)

ingest::otlp produit de la télémétrie sur chaque chemin de rejet (type de média non supporté, échec de décodage, canal plein). Avant la 0.6.0, ces appels atteignaient directement report::metrics::MetricsState, ce qui faisait fuir l'implémentation des métriques aval vers l'amont et rendait ingest inutilisable sans payer le registre Prometheus. Le trait MetricsSink est l'abstraction : MetricsState l'implémente (dans report::metrics) pour que les appelants du daemon gardent le même câblage, et les builds alternatifs (un fork à métriques OpenTelemetry, des tests avec un faux compteur) branchent leur propre sink sans toucher ingest. Les bornes Send + Sync existent parce que les chemins gRPC et HTTP partagent le sink entre tâches tokio via Arc<dyn MetricsSink>. L'impl sur MetricsState conserve le même dispatch sans branche du chemin chaud que les méthodes inhérentes pré-trait : les deux points d'entrée atteignent les compteurs IntCounter mis en cache, sans lookup de hashmap de labels.

Ingestion JSON

pub fn ingest(&self, raw: &[u8]) -> Result<Vec<SpanEvent>, Self::Error> {
    if raw.len() > self.max_size {
        return Err(JsonIngestError::PayloadTooLarge { ... });
    }
    serde_json::from_slice(raw)
}

La taille du payload est vérifiée avant la désérialisation. Cela empêche serde_json d'allouer de la mémoire pour un payload JSON de plusieurs gigaoctets avant de le rejeter.

Auto-détection du format

JsonIngest auto-détecte le format d'entrée en utilisant des heuristiques au niveau des octets. Il examine les premiers 1-4 Ko du payload pour déterminer s'il s'agit de Jaeger, Zipkin ou du format natif, évitant un double parsing coûteux.

Sanitisation à la frontière. Après le parsing, le chemin d'ingestion JSON valide cloud_region via is_valid_region_id et exécute sanitize_span_event sur chaque événement, appliquant les mêmes limites de champs et troncatures UTF-8 que le chemin OTLP. Cela garantit des données uniformément sanitisées en aval, quel que soit le format d'ingestion.

Ingestion Jaeger JSON

ingest/jaeger.rs parse le format d'export Jaeger JSON. Le startTime (microsecondes) est converti via micros_to_iso8601 du module partagé time.rs. Le parent_span_id est extrait depuis les references où refType = "CHILD_OF".

Ingestion Zipkin JSON v2

ingest/zipkin.rs parse le format Zipkin JSON v2. Le parentId est un champ direct. Les tags sont un HashMap<String, String>.

Boucle événementielle du daemon

Architecture

OTLP gRPC (port 4317)   ─┐                                          ┌─ worker d'analyse ─┐
OTLP HTTP (port 4318)   ─┤─→ mpsc(1024) ─→ TraceWindow ─→ éviction ─→ mpsc(1024) ─→ detect ─→ score ─→ NDJSON
Socket unix JSON        ─┘     (boucle select!)                      └────────────────────┘

La boucle événementielle utilise tokio::select! pour multiplexer :

• Réception d'événements depuis le canal -> normaliser -> pousser dans la fenêtre -> enfiler les évictions
• Ticker toutes les TTL/2 ms -> évincer les traces expirées -> enfiler
• Ctrl+C -> vider toutes les traces -> remettre au worker -> joindre -> arrêt

detect+score ne tournent pas sur la boucle select!. Ils s'exécutent sur un unique worker d'analyse dédié alimenté par un canal borné (voir Worker d'analyse), de sorte qu'une passe d'analyse longue ne peut plus bloquer l'ingestion ni l'éviction.

Normalisation en dehors du verrou

// Normaliser EN DEHORS du verrou :
let normalized: Vec<_> = events.into_iter().map(normalize::normalize).collect();
// Puis acquérir le verrou et pousser :
let mut w = window.lock().await;
for event in normalized { w.push(event, now_ms); }

La normalisation est un travail lié au CPU (regex, manipulation de chaînes). La déplacer en dehors du verrou Mutex minimise le temps de détention du verrou aux seules opérations HashMap. Sous contention (ticker et réception s'exécutant concurremment), cela empêche le ticker d'éviction de bloquer sur la normalisation.

Échantillonnage au niveau des traces

fn should_sample(trace_id: &str, rate: f64) -> bool {
    let mut hash: u64 = 0xcbf2_9ce4_8422_2325; // base de décalage FNV-1a
    for b in trace_id.as_bytes() {
        hash ^= u64::from(*b);
        hash = hash.wrapping_mul(0x0100_0000_01b3); // nombre premier FNV-1a
    }
    (hash as f64 / u64::MAX as f64) < rate
}

Le hash FNV-1a est un hash rapide, non cryptographique qui produit une sortie bien distribuée. La base de décalage et le nombre premier sont les constantes standard FNV-1a 64 bits.

Pourquoi FNV-1a ? Plus simple et plus rapide (~2ns pour un trace_id typique) que std::hash::DefaultHasher (SipHash, ~10ns). La qualité cryptographique n'est pas nécessaire pour l'échantillonnage : seule la distribution uniforme compte.

Déterministe : le même trace_id produit toujours la même décision d'échantillonnage, garantissant que tous les événements d'une trace sont soit conservés soit supprimés ensemble.

Cache par lot : la fonction apply_sampling() filtre un lot d'événements en utilisant un cache HashMap<String, bool>. Au sein d'un seul lot, plusieurs événements peuvent partager un trace_id. Le cache utilise get() avant insert() de sorte que trace_id n'est cloné que pour le premier événement de chaque trace, pas à chaque hit de cache. Extraire cette logique dans une fonction dédiée garde la boucle tokio::select! lisible.

Canal borné

let (tx, mut rx) = mpsc::channel::<Vec<SpanEvent>>(1024);

Le canal borné fournit une contre-pression : si la boucle événementielle prend du retard et que le buffer se remplit à 1024 lots, les émetteurs d'ingestion attendront jusqu'à ce qu'un espace soit disponible. Cela empêche la croissance mémoire non bornée par des producteurs rapides.

Worker d'analyse

detect+score sont CPU-bound. Les exécuter en ligne sur la tâche select! faisait qu'une passe d'analyse longue bloquait la boucle, l'empêchant de poller rx.recv() et le ticker d'éviction : la liveness d'ingestion et d'éviction dépendait alors de la latence d'analyse. À la place, les lots évincés (LRU), expirés (TTL) et vidés (shutdown) sont remis à un unique worker d'analyse via un second canal borné :

let (work_tx, work_rx) = mpsc::channel::<AnalysisBatch>(1024);
let worker = tokio::spawn(run_analysis_worker(work_rx, ctx));

• Un worker, un canal, FIFO. L'analyse reste mono-thread et les lots atteignent le worker dans l'ordre où ils ont quitté la fenêtre, donc le corrélateur cross-trace (avec état) observe une séquence déterministe. Pas de pool de threads, pas de spawn_blocking par lot.
• Enfilement non bloquant avec délestage compté. La boucle enfile en try_reserve (synchrone, n'attend jamais l'analyse), construisant le CarbonContext possédé seulement une fois un slot réservé, pour qu'un délestage ne paie jamais un clone jeté. File pleine (ou worker arrêté) → le lot entier est délesté et compté via perf_sentinel_analysis_shed_batches_total et perf_sentinel_analysis_shed_traces_total ; perf_sentinel_analysis_queue_depth suit le backlog. La surcharge est explicite et observable, jamais un drop silencieux. Compromis assumé : sous surcharge soutenue on perd des lots entiers plutôt que de bloquer l'ingestion (choix de liveness/contre-pression, pas de débit).
• Contrôle d'admission par pression mémoire (opt-in). Le délestage par profondeur de queue est un proxy de la mémoire : quand l'analyse suit, la queue reste vide alors que la TraceWindow (bornée par un compte de traces et un TTL, pas par des octets) peut quand même faire croître la RSS au-delà de la limite cgroup et faire OOMKiller le pod avant tout délestage. Quand [daemon] memory_high_water_pct > 0, un watcher à 1 Hz (daemon/mem_pressure.rs) lit memory.current / memory.max du cgroup v2 et bascule un flag partagé (avec hystérésis) dès que l'usage franchit le seuil. Les handlers OTLP lisent ce flag et rejettent l'ingest avec un statut retryable (perf_sentinel_otlp_rejected_total{reason="memory_pressure"}), stoppant la croissance à la source pour que l'éviction TTL draine la window et que la RSS reflue. cgroup v2 / Linux uniquement, inerte et sans surcoût quand désactivé ou non supporté.
• CarbonContext échantillonné à l'éviction. Le CarbonContext par lot (snapshots des scrapers d'énergie + intensité réseau) est construit côté boucle au moment de l'éviction et voyage avec le lot, ce qui préserve l'instant d'échantillonnage précédent.
• Shutdown : drain puis join. Sur Ctrl+C / SIGTERM la boucle vide la fenêtre, remet le reste au worker via un send bloquant (livraison garantie, sans délestage), ferme le canal et attend le worker afin que chaque lot bufferisé et en vol soit entièrement analysé avant le retour.
• Fail-loud si le worker meurt. Une quatrième branche select! surveille le JoinHandle du worker. S'il s'arrête avant le shutdown (un détecteur panique), run_event_loop retourne DaemonError::AnalysisWorkerStopped : le process sort et un superviseur le redémarre, plutôt que de rester debout à n'analyser plus rien. Cela restaure la sémantique fail-loud de la détection en ligne (une panique crashait le daemon).

Renforcement de la sécurité

Permissions du socket Unix :

use std::os::unix::fs::PermissionsExt;
std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600));

Le mode 0o600 restreint la lecture/écriture au propriétaire du socket uniquement, empêchant d'autres utilisateurs locaux d'injecter des événements. Si set_permissions échoue, le fichier socket est supprimé et le listener ne démarre pas (erreur fatale, pas un simple avertissement).

Sémaphore de connexion :

let semaphore = Arc::new(tokio::sync::Semaphore::new(128));

Limite les connexions concurrentes du socket JSON à 128. Sans cela, un attaquant local pourrait ouvrir des milliers de connexions, chacune consommant une tâche tokio et de la mémoire tampon.

Limite d'octets par connexion :

const CONNECTION_LIMIT_FACTOR: u64 = 16;
let limited = stream.take(max_payload_size as u64 * CONNECTION_LIMIT_FACTOR);

Chaque connexion est limitée à 16 × max_payload_size octets au total (défaut 16 Mo). Cela empêche une seule connexion de consommer une mémoire non bornée avec un flux de données qui ne contient jamais de saut de ligne.

Timeouts des requêtes :

• gRPC : tonic::transport::Server::builder().timeout(Duration::from_secs(60))
• HTTP : tower::timeout::TimeoutLayer::new(Duration::from_secs(60)) via le HandleErrorLayer d'axum

Ceux-ci empêchent les connexions lentes/bloquées de retenir des ressources indéfiniment. Le handler de timeout HTTP émet un log tracing::debug! avant de retourner 408 REQUEST_TIMEOUT, aidant les opérateurs à diagnostiquer les clients lents ou bloqués.

Sortie NDJSON

Les findings sont émis en JSON délimité par des sauts de ligne sur stdout en utilisant serde_json::to_writer avec un handle stdout verrouillé pour éviter les allocations String intermédiaires :

let stdout = std::io::stdout();
let mut lock = stdout.lock();
for finding in &findings {
    if serde_json::to_writer(&mut lock, finding).is_ok() {
        let _ = writeln!(lock);
    }
}

Ce format est compatible avec les outils d'agrégation de logs (Loki, ELK) qui consomment du JSON délimité par des lignes. Chaque ligne est un objet JSON complet qui peut être parsé indépendamment.

Ratio de gaspillage cumulatif

La jauge Prometheus io_waste_ratio est calculée à partir de compteurs cumulatifs :

let cumulative_total = metrics.total_io_ops.get();
if cumulative_total > 0.0 {
    metrics.io_waste_ratio.set(metrics.avoidable_io_ops.get() / cumulative_total);
}

C'est une moyenne sur toute la durée, pas une métrique fenêtrée. Les utilisateurs qui ont besoin d'un taux récent peuvent utiliser rate() de Prometheus sur les compteurs bruts (total_io_ops, avoidable_io_ops).

Exemplars Grafana

Le crate prometheus 0.14.0 ne supporte pas nativement les exemplars OpenMetrics. Plutôt que d'ajouter une dépendance, les annotations exemplars sont injectées par post-traitement du texte Prometheus rendu.

Suivi des trace_id worst-case :

MetricsState stocke les données d'exemplars dans des champs protégés par RwLock :

• worst_finding_trace: HashMap<(String, String), ExemplarData> : indexé par (finding_type, severity), mis à jour à chaque appel record_batch()
• worst_waste_trace: Option<ExemplarData> : le trace_id du finding avec le plus d'I/O évitables

RwLock est utilisé plutôt que Mutex car render() (chemin de lecture) est appelé fréquemment par les scrapes Prometheus, alors que record_batch() (chemin d'écriture) est appelé moins souvent. L'empoisonnement de lock est géré gracieusement via unwrap_or_else(PoisonError::into_inner), de sorte qu'un panic dans un thread ne cascade pas en crashs sur les acquisitions de lock suivantes.

Injection d'exemplars :

inject_exemplars() itère sur le texte rendu ligne par ligne. Pour les lignes perf_sentinel_findings_total{...}, il parse les labels type et severity pour trouver l'exemplar correspondant. Pour les lignes perf_sentinel_io_waste_ratio, il ajoute l'exemplar de gaspillage.

Le format suit la spécification OpenMetrics : metric{labels} value # {trace_id="abc123"}. Quand des exemplars sont présents, le header Content-Type passe de text/plain; version=0.0.4 (Prometheus) à application/openmetrics-text; version=1.0.0 (OpenMetrics) pour que la source de données Prometheus de Grafana puisse reconnaître et afficher les liens d'exemplars.

Ingestion pg_stat_statements

ingest/pg_stat.rs fournit un chemin d'analyse autonome pour les exports pg_stat_statements de PostgreSQL. Contrairement à l'ingestion basée sur les traces, ces données n'ont pas de trace_id ni de span_id, elles ne peuvent pas alimenter le pipeline de détection N+1/redondant. Elles fournissent un classement de hotspots et une référence croisée avec les findings de traces.

Décisions de conception

Séparé de IngestSource : le trait IngestSource retourne Vec<SpanEvent>, mais les données pg_stat_statements ne correspondent pas à SpanEvent (pas de trace_id, span_id, ni timestamp). Elles produisent leur propre type PgStatReport avec des classements.

Auto-détection du format : suit le même pattern d'heuristique byte-level que json.rs. Si le premier octet non-espace est [ ou {, parse en JSON ; sinon, parse en CSV. Pas de crate csv externe, le parseur CSV gère le quoting RFC 4180 manuellement (champs entre guillemets doubles, "" échappé).

Réutilisation de la normalisation SQL : chaque requête passe par normalize::sql::normalize_sql() pour produire un template comparable avec les findings basés sur les traces.

Sortie à quatre classements

rank_pg_stat(entries, top_n) retourne un PgStatReport avec quatre classements dans un ordre stable, indexé par position : [by_total_time, by_calls, by_mean_time, by_io_blocks]. Chaque classement contient les mêmes top-N entrées, réordonnées selon son propre critère :

• by_total_time : total_exec_time_ms décroissant. Requêtes qui dominent le temps DB wall-clock. Signal hotspot principal.
• by_calls : calls décroissant. Requêtes à fort volume, candidates N+1 typiques.
• by_mean_time : mean_exec_time_ms décroissant. Requêtes individuellement lentes indépendamment du volume.
• by_io_blocks : shared_blks_hit + shared_blks_read décroissant. Signal de pression cache : requêtes qui touchent le plus de pages du buffer partagé, peu importe si elles étaient chaudes ou froides. Complémentaire de by_total_time quand le CPU est idle mais que le cache s'agite.

Le sub-switcher du dashboard HTML onglet pg_stat consomme ces quatre classements par position, donc les nouveaux classements s'ajoutent en fin de liste (jamais réordonnés, jamais insérés au milieu) pour préserver la stabilité des indices côté consommateurs.

Référence croisée

cross_reference() accepte &mut [PgStatEntry] et &[Finding]. Il construit un HashSet des templates de findings et marque les entrées dont le normalized_template correspond. Complexité O(n + m) où n = entrées, m = findings. Le flag seen_in_traces permet à la CLI de mettre en évidence les requêtes présentes dans les deux sources de données.

Scrape Prometheus pour pg_stat

La fonction fetch_from_prometheus(endpoint, top_n) dans ingest/pg_stat.rs permet de récupérer les données pg_stat_statements directement depuis l'API HTTP de Prometheus, sans export CSV/JSON manuel.

Fonctionnement

1. Construire une requête PromQL topk(N, pg_stat_statements_seconds_total) pour obtenir les N requêtes les plus consommatrices.
2. Envoyer une requête GET /api/v1/query?query=... au endpoint Prometheus configuré via le client HTTP partagé (http_client::build_client).
3. Parser la réponse JSON au format standard Prometheus (data.result[]).
4. Extraire les labels query ou queryid du champ metric pour chaque résultat.
5. Convertir la valeur en millisecondes (la métrique est en secondes), normaliser le SQL via normalize_sql() et produire des PgStatEntry.

Le timeout est fixé à 30 secondes. Les erreurs de transport et de format sont rapportées via des variantes dédiées de PgStatError (PrometheusRequest, PrometheusFormat).

Usage CLI

perf-sentinel pg-stat --prometheus http://prometheus:9090

Le flag --prometheus est mutuellement exclusif avec --input. Le --traces pour la référence croisée fonctionne de la même manière qu'avec un fichier local.

La sous-commande report expose la même capacité via --pg-stat-prometheus URL, mutuellement exclusif avec le flag fichier --pg-stat FILE (enforced au niveau clap via conflicts_with). Quand l'un des deux est passé, le PgStatReport résultant est embarqué dans l'onglet pg_stat du dashboard HTML avec les quatre classements décrits ci-dessus. Le chemin de scrape est partagé avec pg-stat --prometheus, aucun code de fetch dupliqué.

Ingestion Tempo

ingest/tempo.rs fournit le chemin de replay post-mortem : la sous-commande interroge l'API HTTP d'un Grafana Tempo en cours d'exécution, récupère les corps de traces en protobuf OTLP, les décode via le helper existant convert_otlp_request et renvoie un Vec<SpanEvent> au pipeline d'analyse standard. Deux modes : trace unique par ID (--trace-id, un seul GET /api/traces/{id}) ou search-then-fetch (--service --lookback, un GET /api/search suivi de fetches trace par trace). Gatée derrière la cargo feature tempo.

Fetch parallèle avec cap de concurrence

La boucle de fetch par trace est parallélisée via tokio::task::JoinSet, protégée par un Arc<Semaphore> capé à FETCH_CONCURRENCY = 16 permits. Chaque task spawnée acquiert un permit via acquire_owned avant l'appel HTTP et le libère au drop (RAII). Le cap a été choisi empiriquement pour saturer une connexion Tempo distante sur lien WAN (observé ~10-20s pour 100 traces vs. ~2m30s avec la boucle séquentielle précédente) sans mettre à genoux une seule replica de query-frontend. Il est hardcodé aujourd'hui, pas exposé en configuration utilisateur. Le pattern reprend celui de score::cloud_energy::scraper, qui parallélise de la même façon les requêtes CPU Prometheus par service.

Séparation des timeouts

Deux constantes dédiées plutôt qu'une valeur unique : SEARCH_TIMEOUT = 5s pour /api/search (la réponse est une petite liste de trace IDs, un timeout serré fait échouer vite sur un endpoint cassé) et FETCH_TRACE_TIMEOUT = 30s pour /api/traces/{id} (les corps de traces peuvent légitimement faire plusieurs MiB sur une requête à fanout large et la query-frontend doit assembler les spans depuis les ingesters + le stockage long terme). Un cap unique à 5 s droppait empiriquement des dizaines de traces par batch de 100 sur les fenêtres longues ; 30 s correspond au défaut de la datasource Tempo côté Grafana. Les deux timeouts sont passés en paramètre au helper partagé fetch_raw plutôt que stockés dans une constante unique au niveau module, pour que les chemins search et fetch-trace ne puissent pas diverger silencieusement.

Ctrl-C et agrégation d'erreurs

La drain loop est conduite par un tokio::select! avec ordre biased : tokio::signal::ctrl_c() est polled avant set.join_next() pour qu'une interruption en attente ne soit pas starvée par une rafale de completions. Sur signal, set.abort_all() flague toutes les tasks in-flight pour cancellation ; les traces déjà complétées sont conservées, les tasks aborted résolvent en JoinError::is_cancelled() et sont silencieusement skippées. La variante dédiée TempoError::Interrupted est renvoyée uniquement si zéro trace n'a eu le temps de se compléter avant le signal, pour que les quality gates CI puissent distinguer un abort opérateur d'un résultat vide authentique (NoTracesFound).

Les failures par trace loguent au niveau debug, pas error. Une seule ligne de summary classifiée (emit_fetch_summary) est émise à la fin de la boucle, bucketée par type d'erreur (timeout, transport, http_status, protobuf_decode, body_read, json_parse, task_panic) pour que l'outillage downstream (Loki, CloudWatch) puisse alerter sur le bon signal sans parser 50 lignes ERROR individuelles sur un Tempo dégradé. La sévérité du summary suit la pire classe observée : warn si uniquement des skips TraceNotFound ont eu lieu (condition occasionnelle attendue, ex. une trace sortie de rétention entre le search et le fetch), error sinon. Un test unitaire (classify_fetch_error_buckets_every_hard_failure_variant) sert de garde-fou contre la dérive : si une nouvelle variante est ajoutée à TempoError plus tard, elle ne tombe pas silencieusement dans "other".

Ingestion Jaeger Query API

ingest/jaeger_query.rs est l'autre source de replay en mode pull, gardée derrière la feature cargo jaeger-query. Elle interroge n'importe quel backend qui parle l'API HTTP de requête Jaeger (Jaeger upstream et Victoria Traces, qui implémente la même surface). Contrairement au search-puis-fetch en deux étapes de Tempo, le /api/traces de Jaeger retourne les traces complètes dans la réponse de search, donc un seul aller-retour HTTP couvre toute l'ingestion. Le payload décodé réutilise le parser jaeger du mode fichier (convert_jaeger_export), donc une trace Jaeger-Query et un fichier JSON Jaeger passent par une normalisation identique. Elle partage l'unique http_client.rs et le helper auth_header.rs avec le chemin Tempo. Le validateur d'endpoint accepte toute URL http(s) et ne bloque pas les cibles RFC 1918 ou link-local, donc la sous-commande ne doit être invoquée qu'avec des valeurs d'endpoint de confiance (voir Limites).

API de requête du daemon

En mode watch, le daemon expose une API HTTP de consultation aux côtés des endpoints existants /v1/traces, /metrics et /health. Cela permet d'interroger l'état interne du daemon sans accéder directement à stdout.

FindingsStore

daemon/findings_store.rs implémente un buffer circulaire thread-safe pour les findings récents :

pub struct FindingsStore {
    inner: RwLock<VecDeque<StoredFinding>>,
    max_size: usize,
}

Le RwLock tokio permet plusieurs lecteurs simultanés (scrapes de l'API) sans bloquer entre eux, tout en garantissant l'exclusivité pour les écritures (insertion depuis process_traces). La capacité initiale du VecDeque est plafonnée à min(max_size, INITIAL_CAPACITY_CEILING) avec un plafond de 4096 pour amortir les réallocations sans gonfler le RSS au démarrage.

Court-circuit max_size == 0 : quand max_retained_findings = 0, push_batch retourne immédiatement sans allouer. Cela permet aux opérateurs qui désactivent l'API (api_enabled = false) de récupérer la mémoire du store en mettant aussi max_retained_findings = 0.

Clones hors lock : push_batch construit les nouvelles entrées StoredFinding AVANT d'acquérir le write lock, puis fait un extend + drain rapide sous lock. Les lecteurs API ne sont pas bloqués par les allocations Finding::clone().

Eviction : quand le buffer atteint sa capacité maximale (défaut 10 000), chaque nouvel ajout via push_batch évince les plus anciens via drain(..excess). Cela maintient un coût mémoire borné.

Filtrage : query() parcourt le buffer en ordre inverse (plus récent d'abord) et applique des filtres optionnels par service, type de finding et sévérité. La limite par défaut est de 100 résultats, plafonnée à MAX_FINDINGS_LIMIT = 1000.

Endpoints HTTP

daemon/query_api/ définit dix routes axum montées dans le routeur existant du daemon. Le router n'est mergé dans le stack HTTP que si [daemon] api_enabled = true (défaut true). Mettre api_enabled = false désactive toutes les routes /api/* tout en conservant l'ingestion OTLP, /metrics et /health.

Sémantique du snapshot /api/export/report

L'endpoint retourne un Report de forme identique à la sortie de analyze --format json, donc la réponse peut être pipée directement dans perf-sentinel report --input - pour matérialiser un dashboard HTML depuis un daemon vivant. Les champs sont remplis depuis l'état live du daemon : findings depuis FindingsStore::query, correlations depuis CrossTraceCorrelator::active_correlations, analysis.events_processed et traces_analyzed depuis les compteurs metrics (valeurs cumulées sur la vie du daemon, pour le contexte).

green_summary est rafraîchi par l'event loop après chaque batch terminé. Vue par batch : chaque champ numérique en dessous (total_io_ops, avoidable_io_ops, io_waste_ratio, co2.*, regions, top_offenders, transport_gco2) reflète uniquement le batch le plus récent, pas un agrégat sur la vie du daemon. Les opérateurs qui veulent des nombres GreenOps cumulés doivent scraper les compteurs Prometheus de /metrics à la place. L'onglet GreenOps du dashboard HTML ne s'affiche que quand green_summary.co2 est non nul, donc un daemon configuré avec Electricity Maps fait apparaître naturellement la bannière une fois le premier batch traité. analysis.duration_ms vaut 0, pas l'uptime du daemon : la valeur du pipeline batch chronomètre une seule analyse, et un snapshot de daemon n'a pas de telle exécution.

Gestion du cold-start : l'endpoint retourne 200 OK avec une enveloppe Report vide (findings: [], green_summary: GreenSummary::disabled(0), warnings: ["daemon has not yet processed any events"]). Avant la 0.5.16, ce chemin retournait 503 Service Unavailable, ce qui déclenchait les probes Kubernetes et perturbait les scripts CI traitant les 5xx comme un problème de santé du daemon ; l'enveloppe vide permet aux clients de détecter le cold-start sans incohérence de code de statut. La vérification du cold-start s'appuie sur un double compteur (events_processed_total > 0 ET traces_analyzed_total > 0) : des événements peuvent être ingérés quelques secondes avant le premier tick d'éviction (trace_ttl_ms / 2, 15 s par défaut), donc se contenter de events_processed > 0 exposerait une fenêtre où la cellule est encore disabled(0). Le compteur export_report_requests_total est incrémenté avant la vérification de cold-start, donc les réponses de cold-start sont comptées aussi, conformément aux conventions des access logs HTTP.

Taille de réponse : bornée par MAX_FINDINGS_LIMIT + MAX_CORRELATIONS_LIMIT (1000 + 1000 entrées) plus un green_summary borné (top_offenders plafonné, regions limité par la cardinalité des régions cloud), corps d'environ 3,5 Mo au pire. Acceptable sur un bind loopback (la posture documentée) ; le plafond mérite une revue si le daemon est un jour exposé hors loopback.

Atomicité du snapshot : le handler acquiert le read lock du FindingsStore puis le mutex du correlator en séquence, pas atomiquement. Les deux collections peuvent donc être décalées d'un batch (findings de la génération N, correlations de N+1), acceptable pour un dashboard post-mortem mais pas pour un contrat de snapshot strict.

L'état partagé est encapsulé dans QueryApiState :

pub struct QueryApiState {
    pub findings_store: Arc<FindingsStore>,
    pub window: Arc<tokio::sync::Mutex<TraceWindow>>,
    pub detect_config: DetectConfig,
    pub start_time: std::time::Instant,
    pub correlator: Option<Arc<tokio::sync::Mutex<CrossTraceCorrelator>>>,
}

Le endpoint /api/explain/{trace_id} consulte la TraceWindow pour récupérer les spans (s'ils sont encore en mémoire), exécute les détecteurs par trace, puis construit l'arbre via explain::build_tree et explain::format_tree_json. Si la trace a déjà été évincée, il retourne un objet JSON avec un champ error.

Configuration

[daemon]
api_enabled = true
max_retained_findings = 10000

api_enabled est à true par défaut. Quand api_enabled = false, les routes /api/* ne sont pas montées, mais le FindingsStore est toujours peuplé à chaque tick (la détection tourne avant la vérification du flag). Pour libérer aussi la mémoire du store, mettez max_retained_findings = 0 : push_batch court-circuite alors sans allouer.

Intégration du corrélateur cross-trace

Le CrossTraceCorrelator (décrit dans 04 : Détection) est instancié dans la boucle événementielle du daemon quand [daemon.correlation] enabled = true. À chaque tick :

1. process_traces produit un lot de findings.
2. Les findings sont insérés dans le FindingsStore via push_batch.
3. Les findings sont passés au corrélateur via ingest(findings, now_ms).
4. GET /api/correlations appelle active_correlations() pour retourner les paires au-dessus des seuils configurés.

Le corrélateur est possédé par la boucle du daemon (pas dans un Arc/Mutex séparé), puisque seul le ticker y accède. Cela évite le coût de synchronisation.

Store ack daemon : JSONL + concurrence

Le store ack côté daemon (crates/sentinel-core/src/daemon/ack.rs) complète les acknowledgments TOML côté CI (crate::acknowledgments) avec une API runtime pour les cas SRE-on-call. Les deux sources sont unionées au moment de la query, le TOML l'emportant en cas de conflit (baseline immuable livrée via revue de PR).

Format de fichier

JSONL append-only à ~/.local/share/perf-sentinel/acks.jsonl par défaut. Chaque ligne est un événement :

{"action":"ack","signature":"<sig>","by":"alice","reason":"...","at":"2026-05-04T13:30:00Z","expires_at":null}
{"action":"unack","signature":"<sig>","by":"alice","at":"2026-05-04T14:00:00Z"}

Compaction au démarrage

Le daemon rejoue le JSONL dans une HashMap<Signature, AckEntry> (apply sur Ack, remove sur Unack, drop sur expiration), puis réécrit atomiquement le fichier via tmp + rename avec uniquement les entrées actives. Une boucle ack/unack qui s'emballe ne peut donc pas accumuler indéfiniment, le fichier se reset à chaque redémarrage.

Modèle de concurrence

La map en mémoire est derrière un RwLock pour des lectures snapshot bon marché. Les écritures disque passent par un Mutex<File> pour que des appels ack/unack concurrents produisent chacun une ligne JSONL bien formée. Le mutex est tenu pour toute la durée write + map-update, donc une écriture disque qui échoue ne laisse jamais la map en avance sur l'état persisté.

Parsing du header d'autorisation

Le helper auth-header vit dans crates/sentinel-core/src/ingest/auth_header.rs. Il parse une ligne --auth-header "Name: Value" user-supplied en une paire (HeaderName, HeaderValue) hyper-safe, partagée entre les sous-commandes Tempo et Jaeger-Query.

La valeur parsée est marquée sensitive pour qu'hyper l'omette de son propre debug output et des tables de compression HPACK HTTP/2. La struct implémente aussi un Debug manuel qui n'imprime jamais la valeur, donc un AuthHeader loggé ne fuit jamais le credential.

Règles de validation

Le parsing est volontairement strict. Au-delà des checks au niveau hyper (nom token-only, valeur VCHAR + SP + HTAB, donc les tabs et espaces internes dans la valeur sont préservés tels quels et seuls CR/LF + ASCII non-visible sont rejetés), le parseur refuse aussi :

• Les inputs bruts plus longs que 8 KiB, pour borner le clone par tâche dans le fanout parallèle Tempo et stopper un --auth-header "X: $(cat /dev/urandom | head -c 50M | base64)" pathologique à la porte. Un JWT typique fait 2 à 4 KiB, 8 KiB laisse de la marge pour des tokens multi-claims longs sans ouvrir la porte à des blobs arbitraires.
• Les valeurs vides après trim, qui enverraient un Authorization: inutile au backend et produiraient un 401 confus.
• Les noms de header qui activeraient du request smuggling ou un override d'authority si user-supplied : Host, Content-Length, Transfer-Encoding, Connection, Upgrade, TE, Proxy-Connection. Les utilisateurs voulant tweaker ceux-là devraient passer par un proxy local, pas par ce flag.

Ingestion des digests performance_schema MySQL

ingest/mysql_stat.rs est le pendant MySQL de pg_stat_statements : il lit un export CSV ou JSON de events_statements_summary_by_digest et classe les points chauds SQL. Comme la voie Postgres, il n'a pas de trace_id, c'est donc une vue complémentaire côté base et non quelque chose que le corrélateur peut joindre.

Deux détails mécaniques faciles à rater. Les colonnes de temps (SUM_TIMER_WAIT, AVG_TIMER_WAIT) arrivent en picosecondes et sont converties au parsing. Les classements sortent dans un ordre fixe (temps total, nombre d'appels, temps moyen, lignes examinées) et les consommateurs indexent dans ce tableau, un nouveau classement s'ajoute donc à la fin et les positions existantes ne bougent jamais.

Le pont digest vers trace est la partie intéressante. Marquer un digest comme "vu dans les traces" demande de comparer le DIGEST_TEXT de MySQL à un template normalisé depuis du SQL applicatif, et une comparaison littérale ne correspond jamais : MySQL espace chaque token (`c` . `name`), met les mots-clés en majuscules et force les backticks, rien de tout cela ne survivant à la normalisation côté application. Les deux côtés sont donc canonicalisés d'abord, en retirant les backticks, en supprimant les espaces autour de la ponctuation, en réduisant les suites d'espaces et en passant en minuscules.

Cette dernière étape a un plafond assumé qui mérite d'être écrit, car c'est un arbitrage et non un oubli : la mise en minuscules replie les identifiants en même temps que les mots-clés, donc sur un serveur sensible à la casse (lower_case_table_names=0) deux tables qui ne diffèrent que par la casse partagent une clé et le marqueur peut sur-correspondre. Ne replier que les mots-clés demanderait une table complète des mots-clés MySQL. Pour un marqueur informatif, la sur-correspondance est l'erreur la moins chère.

Gestion des signaux d'arrêt

shutdown.rs se résout sur SIGINT partout, et en plus sur SIGTERM sous Unix, ce que Kubernetes envoie à la terminaison d'un pod, ce que kill envoie par défaut et ce que systemd utilise pour arrêter une unité. Les deux déclenchent le même nettoyage. Construisez le future une seule fois et faites tokio::pin! avant une boucle select!, sinon les listeners se réenregistrent à chaque itération.

Une réserve mérite d'être connue avant de réutiliser ceci hors du daemon : sous Unix, l'enregistrement du handler SIGTERM est global au processus et définitif. Tokio ne restaure jamais la disposition par défaut, donc une fois ce future attendu, le processus ne meurt plus par défaut sur SIGTERM, pour le reste de sa vie, même après que le future a été détruit. C'est exactement ce que veut un daemon de longue durée et exactement ce que ne veut pas une commande one-shot.

Helpers d'ingestion partagés

ingest/lookback.rs et ingest/url_enc.rs existent parce que tempo et jaeger_query avaient besoin des deux mêmes choses et commençaient chacun à faire pousser les siennes. Les deux gardent leur type d'erreur au site d'appel et ne partagent que la logique.

Le parseur de lookback accepte les suffixes h, m, s et somme les formes composées (2h30m vaut 9000 s), avec de l'arithmétique vérifiée partout pour que 999999999h remonte en erreur de débordement au lieu de boucler en release.

Les helpers d'URL écrivent à la main un encodeur pour-cent minimal plutôt que de tirer percent-encoding pour douze lignes. Le validateur d'endpoint est volontairement étroit : il rejette un schéma autre que http(s) et les identifiants dans l'autorité, et accepte délibérément un @ littéral dans le chemin ou la query, pour que /api/traces?owner=foo%40example.com fonctionne.