Debugguer avec la télemétrie en Rust

Bonjour à toutes et tous 😀

Un court article sur ma frustration du moment. Je bosse actuellement avec de l’asynchrone et du parrallélisme.

C’est très cool quand ça marche mais c’est une galère sans nom à débugguer quand ça veut pas. 🙄

Et quand je dis ça veut pas, je veux dire, l’app bloque sans messages de log parce que vous êtes dans un test d’inté par exemple (non c’est pas du tout du vécu 😡).

Aujourd’hui, je vous rédige ce billet pour vous montrer de quelle manière l’on peut débugguer ce genre de situations.

Installer les outils

Nous allons utiliser un projet en Go qui s’appelle Jaeger (rien à voir avoir avec les Titans 😋) , ceci est un outil de tracing permettant de visualiser l’éxécution d’un programme asynchrone.

Comme pré-requis, je considère que vous avez Docker d’installé et de fonctionnel sur votre machine.

Ainsi que Rust 1.62+ également installé et de fonctionnel.

Lancez jaeger.

docker run -d --name jaeger \
    -p6831:6831/udp \
    -p6832:6832/udp \
    -p16686:16686 \
    -p14268:14268 \
    jaegertracing/all-in-one:latest

Si vous vous rendez sur localhost:16686. Vous devriez vous retrouver devant une interface comme celle ci:

Nous allons aussi avoir besoin d’un projet en Rust.

cargo new lab-telementry
cd lab-telementry
cargo add opentelemetry
cargo add opentelemetry-jaeger
cargo add tracing
cargo add tracing-opentelemetry
cargo add tracing-subscriber

Ou dans le Cargo.toml

[dependencies]
opentelemetry = "0.17.0"
opentelemetry-jaeger = "0.16.0"
tracing = "0.1.35"
tracing-opentelemetry = "0.17.4"
tracing-subscriber = "0.3.14"

Les concepts de base

Tracer et Layer

Avant de savoir courir il faut apprendre à marcher.

Notre cerveau primitif, n’étant vraiment pas câblé pour l’asynchrone, nous allons expliquer les concepts de bases dans un environnement synchrone.

Pour cela dans le main.rs, nous allons créer un petit code de test.

Nous allons d’abord déclarer notre Tracer jaeger.

let tracer_jaeger = opentelemetry_jaeger::new_agent_pipeline()
    .with_service_name("lab-telemetry")
    .install_simple()
    .unwrap()

Nous lui donnons le nom de service lab-telemetry. C’est en quelque sorte l’identifiant projet dans Jaeger, en effet on peut faire le tracing d’une multitude de projet à la fois. Il faut bien quelque choses pour les distinguer.

Ensuite nous allons déclarer un Layer OpenTelemetry avec notre Tracer jaeger.

let opentelemetry_layer = tracing_opentelemetry::layer()
    .with_tracer(tracer_jaeger);

OpenTelemetry est une initiative d’uniformisation des standards. Son but est de rendre compatibles l’ensemble des outils de l’écosystème de l’observabilité déjà existants.

Puis nous allons enregister notre Layer dans un Registry.

tracing_subscriber::registry()
    .with(opentelemetry_layer)
    .init();

Le rôle du Registry va être de capturer les Span et de les distribuer aux différent Layers qui ont souscrit à ce Registry. Dans notre cas le opentelemetry_layer.

En parlant de Span créons en une.

let _ = span!(tracing::Level::TRACE, "my span");

Une Span est une portion bornée temporellement dans l’éxécution d’un programme, qui a une date de début et une date de fin.

Entre ces deux bornes temporelles, il va se dérouler des évènements.

Nous allons aussi forcer les Tracer à envoyer toutes les Spans qu’ils connaissent avant de terminer le programme.

shutdown_tracer_provider();

use opentelemetry::global::shutdown_tracer_provider;
use tracing::span;
use tracing_subscriber::layer::SubscriberExt;
use tracing_subscriber::util::SubscriberInitExt;

fn main() {
    // Create a Jaeger Tracer
    let tracer_jaeger = opentelemetry_jaeger::new_pipeline()
        .with_service_name("lab-telemetry")
        .install_simple()
        .unwrap();

    // Create a tracing layer with the configured tracer
    let opentelemetry_layer = tracing_opentelemetry::layer().with_tracer(tracer_jaeger);

    // Register Layers
    tracing_subscriber::registry()
        .with(opentelemetry_layer)
        .init();

    // Declare a span
    let _ = span!(tracing::Level::TRACE, "my span");

    // Force all Span to be sent
    shutdown_tracer_provider();
}

Si vous allez sur localhost:16686. En sélectionnant le service lab-telemetry et ensuite en appuyant sur le bouton Find Traces.

Vous devriez désormais voir la span dans le volet de droite, ici entouré en rouge.

Cela signifie que notre système est opérationnel. Il envoie des Span à Jaeger.

• Nous avons un Tracer jaeger qui est capable de transformer des Spans en dans un format de messages que Jaeger est capable de comprendre.
• Ce Tracer est inclue dans un Layer OpenTelemetry qui sert de couche de compatibilité au reste de la télémétrie.
• Ce Layer est enregistré dans un Registry qui écoute les spans de notre éxécution de code.
• Les Spans sont des intervalles de temps dans l’éxécution de notre code.

Les spans

Je vous dis depuis tout à l’heure qu’une Span est un intervalle de temps. Mais il ne semble pas y avoir ni de début ni de fin à notre Span.

Déclarons le début de notre Span ainsi que sa fin.

    let guard = span!(Level::TRACE, "my_span bound").entered();
    sleep(Duration::from_secs(2));
    guard.exit();

Nous avons bien une Span de 2 secondes.

Ce qui est intéressant, c’est que les Spans peuvent elles-même contenir d’autres Spans

let parent = span!(Level::TRACE, "parent").entered();

    sleep(Duration::from_secs(2));

    let child = span!(Level::TRACE, "child").entered();
        sleep(Duration::from_secs(1));
    child.exit();

parent.exit();

Dans Jaeger cela nous donne bien une Span de 3 secondes.

Mais détail supplémentaire, cette entrée possède 2 spans.

Cliquez sur 2 Spans.

Nous voyons bien l’imbrication des deux Spans. Le parent dure 3 secondes, l’enfant démarre deux secondes après le début du parent et dure 1 seconde.

Ce que vous voyez là ce nomme un flamechart et constitue le nerf de la guerre de notre article.

Il permet de visualiser temporellement les Spans de notre application et ainsi comprendre les agencements des appels lors de l’éxécution du programme.

Pour le moment tout est synchrone donc simple à comprendre mais ça va se corser très vite. 😇

Le code actualisé.

use opentelemetry::global::shutdown_tracer_provider;
use std::thread::sleep;
use std::time::Duration;
use tracing::{span, Level};
use tracing_subscriber::layer::SubscriberExt;
use tracing_subscriber::util::SubscriberInitExt;

fn main() {
    // Create a Jaeger Tracer
    let tracer_jaeger = opentelemetry_jaeger::new_pipeline()
        .with_service_name("lab-telemetry")
        .install_simple()
        .unwrap();

    // Create a tracing layer with the configured tracer
    let opentelemetry_layer = tracing_opentelemetry::layer().with_tracer(tracer_jaeger);

    // Register Layers
    tracing_subscriber::registry()
        .with(opentelemetry_layer)
        .init();

    // Declare a span
    let parent = span!(Level::TRACE, "parent").entered();

    sleep(Duration::from_secs(2));

    let child = span!(Level::TRACE, "child").entered();
    sleep(Duration::from_secs(1));
    child.exit();

    parent.exit();

    // Force all Span to be sent
    shutdown_tracer_provider();
}

Les logs

Nous sommes désormais dans la capacité de déterminer quand telles où telles Span commencent et se terminent et même de les imbriquer.

C’est bien de savoir le Quand mais le Quoi c’est encore mieux.

C’est à ce stade que la notion de logs est introduite.

Elle va nous permettre de définir des messages rattacher à la Span.

let parent = span!(Level::TRACE, "parent").entered();
    info!("Début de la span parent");

    debug!("Nous allons attendre 2 secondes...");
    sleep(Duration::from_secs(2));
    debug!("Fin du délai d'attente de 2 secondes");

    let child = span!(Level::TRACE, "child").entered();
        info!("Début de la span child");
        debug!("Nous allons attendre 2 secondes...");
        sleep(Duration::from_secs(1));
        debug!("Fin du délai d'attente de 2 secondes");
        info!("Fin de la span child");
    child.exit();
    info!("Fin de la span parent");
parent.exit();

Dans Jaeger.

Si vous cliquez sur les deux textes encadrés en rouge à gauche, puis sur Logs. Vous devriez visualiser ceci:

Ce qui est encadré en bleu se nomme un évènement ou Event. Une même Span peut contenir de zéro à plusieurs événements.

Ici, nos deux Spans possèdent chacune 4 Events.

Il est également possible de les ouvrir pour avoir encore plus de détails.

Ici, nous retrouvons le corps du message ainsi que le niveau de log, le numéro de ligne et le fichier qui a émit cet Event.

Ce qui est intéressant c’est que ces Event peuvent-être variabilisés et enrichis.

let id = 42;
let parent = span!(Level::TRACE, "parent").entered();
  info!(request.id = id, "Début de la span parent");

Et même avec des données complexes comme des structures.

#[derive(Debug)]
struct Request {
    id: u8,
    path: &'static str,
    method: &'static str,
}

impl Display for Request {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        write!(f, "Je suis une requête")
    }
}

let request = Request {
    id: 42,
    path: "/home",
    method: "GET",
};

let parent = span!(Level::TRACE, "parent").entered();
info!(
    request.id = request.id,
    request_display = %request,
    request_debug = ?request,
    "Début de la span parent"
);

• Nous pouvons accéder au champ request.id.
• Nous pouvons appeler le trait Dispay de Request via le sigil %.
• Nous pouvons également appeler le trait Debug de Request via le sigil ?.

Sur Jaeger cela nous donne:

Les champs qui sont rattachés à un Event sont des Tags. Un Tag est une entité indexée qui peut être recherchée dans Jaeger et dans tout autre plateforme d’analyse de télémétrie.

Il en va de même pour notre tag personnalisé request.id.

.

Et là on s’aperçoit très vite que si l’on balade un identifiant dans toute nos spans on va pouvoir avoir une vision d’ensemble extrêmement précise de ce qui se passe dans notre système pour un identifiant donné.

• Nous avons un Tracer jaeger qui est capable de transformer des Span en dans un format de messages que Jaeger est capable de comprendre.
• Ce Tracer est compris dans un Layer OpenTelemetry qui sert de couche de compatibilité au reste de la télémétrie.
• Ce Layer est enregistré dans un Registry qui écoute les spans de notre éxécution de code.
• Les Spans sont des intervalles de temps dans l’éxécution de notre code.
• Les Spans peuvent avoir des Events
• Les Events sont des points datés dans l’éxécution d’une Span
• Un Event peut avoir des Tags
• Les Tags sont des clefs-valeurs qui sont indexés et donc qui peuvent être recherchés au sein d’une plateforme d’analyse de télémétrie.

Le code de cette partie:

use opentelemetry::global::shutdown_tracer_provider;
use std::fmt::{Display, Formatter};
use std::thread::sleep;
use std::time::Duration;
use tracing::{debug, info, span, Level};
use tracing_subscriber::layer::SubscriberExt;
use tracing_subscriber::util::SubscriberInitExt;

fn main() {
    // Create a Jaeger Tracer
    let tracer_jaeger = opentelemetry_jaeger::new_pipeline()
        .with_service_name("lab-telemetry")
        .install_simple()
        .unwrap();

    // Create a tracing layer with the configured tracer
    let opentelemetry_layer = tracing_opentelemetry::layer().with_tracer(tracer_jaeger);

    // Register Layers
    tracing_subscriber::registry()
        .with(opentelemetry_layer)
        .init();

    #[derive(Debug)]
    struct Request {
        id: u8,
        path: &'static str,
        method: &'static str,
    }

    impl Display for Request {
        fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
            write!(f, "Je suis une requête")
        }
    }

    let request = Request {
        id: 42,
        path: "/home",
        method: "GET",
    };

    let parent = span!(Level::TRACE, "parent").entered();
    info!(
        request.id = request.id,
        request_display = %request,
        request_debug = ?request,
        "Début de la span parent"
    );

    debug!("Nous allons attendre 2 secondes...");
    sleep(Duration::from_secs(2));
    debug!("Fin du délai d'attente de 2 secondes");

    let child = span!(Level::TRACE, "child").entered();
    info!("Début de la span child");
    debug!("Nous allons attendre 2 secondes...");
    sleep(Duration::from_secs(1));
    debug!("Fin du délai d'attente de 2 secondes");
    info!("Fin de la span child");
    child.exit();
    info!("Fin de la span parent");
    parent.exit();

    // Force all Span to be sent
    shutdown_tracer_provider();
}

Instrumentation des fonctions

Pour le moment notre système fonctionne bien dans notre méthode main, mais comme tout programme un peu conséquent il faudra invariablement le modulariser.

Et donc le découper en fonction.

Nous allons en écrire deux et réutiliser notre structure Request:

#[derive(Debug)]
struct Request {
    id: u8,
    path: &'static str,
    method: &'static str,
    token: &'static str,
}

fn handle_request(request: Request) {
    if auth(request) {
        // ...
    } else {
        // ...
    }
}

fn auth(request: Request) -> bool {
    request.token == "blabla"
}

Nous allons maintenant rajouter des spans, des events et des tags à nos fonctions.

fn handle_request(request: Request) {
    let span = trace_span!("handle_request").entered();
    if auth(&request) {
        info!(request = ?request, id = request.id, "Request authenticated")
    } else {
        error!(request = ?request, id = request.id, "Request forbidden")
    }
    span.exit();
}

fn auth(request: &Request) -> bool {
    let span = trace_span!("auth").entered();
    // Emulate call for db
    sleep(Duration::from_millis(500));
    let value = request.token == "blabla";
    span.exit();
    value
}

Et dans notre main

let request = Request {
    id: 42,
    path: "/home",
    method: "GET",
    token: "blabla",
};

let request2 = Request {
    id: 43,
    path: "/home",
    method: "GET",
    token: "bleubleu",
};

let span = span!(Level::TRACE, "main").entered();
handle_request(request);
handle_request(request2);
span.exit();

Sur jaeger, cela nous donne ceci:

.

On y voit une span différente pour chaque appel à handle_request.

En ouvrant les détails nous pouvons voir le token en cause.

.

Maintenant nous allons simplifier l’écriture en utilisant un attribut de fonction.

Cet attribut est #[tracing::instrument].

#[tracing::instrument(name = "main")]
fn handle_requests() {
    let request = Request {
        id: 42,
        path: "/home",
        method: "GET",
        token: "blabla",
    };

    let request2 = Request {
        id: 43,
        path: "/home",
        method: "GET",
        token: "bleubleu",
    };

    handle_request(request);
    handle_request(request2);
}

#[tracing::instrument]
fn handle_request(request: Request) {
    if auth(&request) {
        info!(request = ?request, id = request.id, "Request authenticated")
    } else {
        error!(request = ?request, id = request.id, "Request forbidden")
    }
}

#[tracing::instrument]
fn auth(request: &Request) -> bool {
    // Emulate call for db
    sleep(Duration::from_millis(500));
    request.token == "blabla"
}

#[tracing::instrument(name = "main")]

Permet de redéfinir le nom de la span; par défaut il s’agit du nom de la fonction décorée.

Il devient alors facile d’outiller les fonctions et d’en rajouter un peu partout.

En bonus, on gagne également le tag request qui provient des paramètres de notre appel à handle_request. Et ce, gratuitement.

Code de la partie.

use opentelemetry::global::shutdown_tracer_provider;

use std::thread::sleep;
use std::time::Duration;
use tracing::{error, info, span, trace_span, Level};
use tracing_subscriber::layer::SubscriberExt;
use tracing_subscriber::util::SubscriberInitExt;

#[derive(Debug)]
struct Request {
    id: u8,
    path: &'static str,
    method: &'static str,
    token: &'static str,
}

fn main() {
    // Create a Jaeger Tracer
    let tracer_jaeger = opentelemetry_jaeger::new_pipeline()
        .with_service_name("lab-telemetry")
        .install_simple()
        .unwrap();

    // Create a tracing layer with the configured tracer
    let opentelemetry_layer = tracing_opentelemetry::layer()
        .with_tracer(tracer_jaeger);

    // Register Layers
    tracing_subscriber::registry()
        .with(opentelemetry_layer)
        .init();

    handle_requests();

    // Force all Span to be sent
    shutdown_tracer_provider();
}

#[tracing::instrument(name = "main")]
fn handle_requests() {
    let request = Request {
        id: 42,
        path: "/home",
        method: "GET",
        token: "blabla",
    };

    let request2 = Request {
        id: 43,
        path: "/home",
        method: "GET",
        token: "bleubleu",
    };

    handle_request(request);
    handle_request(request2);
}

#[tracing::instrument]
fn handle_request(request: Request) {
    if auth(&request) {
        info!(request = ?request, id = request.id, "Request authenticated")
    } else {
        error!(request = ?request, id = request.id, "Request forbidden")
    }
}

#[tracing::instrument]
fn auth(request: &Request) -> bool {
    // Emulate call for db
    sleep(Duration::from_millis(500));
    request.token == "blabla"
}

L’asynchrone

Nous allons rendre asynchrone et parallélisé le code du dessus.

Avant tout installons tokio.

cargo add tokio --features macros --features rt --features rt-multi-thread

Ce qui nous donne dans le Cargo.toml

tokio = { version = "1.20.0", features = ["macros", "rt", "rt-multi-thread"] }

Puis, il faut que l’on fasse évoluer notre méthode handling_requests.

Elle devient asynchrone

#[tracing::instrument(name = "main")]
async fn handle_requests() {
    // corps de la méthode
}

Puis nous allons modifier notre fonction handle_requests.

#[tracing::instrument(name = "main")]
async fn handle_requests() {
    let request = Request {
        id: 42,
        path: "/home",
        method: "GET",
        token: "blabla",
    };

    let request2 = Request {
        id: 43,
        path: "/home",
        method: "GET",
        token: "bleubleu",
    };

    let span = Span::current();

    let t1 = tokio::spawn(handle_request(request));

    let span = Span::current();
    let t2 = tokio::spawn(
        async {
            thread::sleep(Duration::from_millis(100));
            handle_request(request2).await
        }
    );

    let _ = join!(t1, t2);
}

Pour que l’asynchrone fonctionne il lui faut un reactor, ici ce sera tokio.

#[tokio::main]
async fn main() {
    // ...
}

Étant donné que handling_requests est désormais asynchrone, il faut la poll pour qu’elle fasse quelque chose.

handle_requests().await;

Le code complet

use opentelemetry::global::shutdown_tracer_provider;
use std::thread;

use std::thread::sleep;
use std::time::Duration;
use tokio::join;
use tracing::{error, info, Span};
use tracing_subscriber::layer::SubscriberExt;
use tracing_subscriber::util::SubscriberInitExt;

#[derive(Debug)]
struct Request {
    id: u8,
    path: &'static str,
    method: &'static str,
    token: &'static str,
}

#[tokio::main]
async fn main() {
    // Create a Jaeger Tracer
    let tracer_jaeger = opentelemetry_jaeger::new_pipeline()
        .with_service_name("lab-telemetry")
        .install_simple()
        .unwrap();

    // Create a tracing layer with the configured tracer
    let opentelemetry_layer = tracing_opentelemetry::layer().with_tracer(tracer_jaeger);

    // Register Layers
    tracing_subscriber::registry()
        .with(opentelemetry_layer)
        .init();

    handle_requests().await;

    // Force all Span to be sent
    shutdown_tracer_provider();
}

#[tracing::instrument(name = "main")]
async fn handle_requests() {
    let request = Request {
        id: 42,
        path: "/home",
        method: "GET",
        token: "blabla",
    };

    let request2 = Request {
        id: 43,
        path: "/home",
        method: "GET",
        token: "bleubleu",
    };

    let t1 = tokio::spawn(handle_request(request));
    
    let t2 = tokio::spawn(async {
        // Je décale l'exécution de 100 ms pour un rendu plus visuel
        thread::sleep(Duration::from_millis(100));
        handle_request(request2).await
    });

    let _ = join!(t1, t2);
}

#[tracing::instrument]
async fn handle_request(request: Request) {
    if auth(&request) {
        info!(request = ?request, id = request.id, "Request authenticated")
    } else {
        error!(request = ?request, id = request.id, "Request forbidden")
    }
}

#[tracing::instrument]
fn auth(request: &Request) -> bool {
    // Emulate call for db
    sleep(Duration::from_millis(500));
    request.token == "blabla"
}

En effet, les différentes fonctions évoluant dans des threads différents les contextes et donc les root spans sont aussi différents.

Ce qui nous donne le résultat suivant:

Pour réconciller ces threads nous allons devoir opérer quelques changements.

La crate tracing nous permet de facilement attacher une span à un thread et plus particulèrement à la futur éxécutée.

let future = async {
    
};

tokio::spawn(future.instrument(info_span!("ma span")))

Cet appel à instrument va venir accrocher la span “ma span” à l’éxécution de la futur ainsi sera repésenté par “ma span” dans Jaeger.

Faites le test. 😛

On peut donc en extrapole quelque chose comme:

let fut1 = handle_request(request);
let fut2 = async {
    thread::sleep(Duration::from_millis(100));
    handle_request(request2).await
};

let t1 = tokio::spawn(fut1.instrument(info_span!("handle request success")));
let t2 = tokio::spawn(fut2.instrument(info_span!("handle request fail")));

let _ = join!(t1, t2);

Ce qui nous donne le résultat suivant sur Jaeger

Et si l’on désire ne avoir de wrapping on peut également écrire ceci:

let t1 = tokio::spawn(fut1.instrument(Span::current()));
let t2 = tokio::spawn(fut2.instrument(Span::current()));

Span::current() permet de récupérer la span du thread parent.

Et voilà nous sommes en mesure de pouvoir comprendre d’un coup d’oeil ce qu’il se passe temporellement lors de l’éxécution en parallèle de plusieurs threads.

Le code final

use opentelemetry::global::shutdown_tracer_provider;
use std::thread;

use std::thread::sleep;
use std::time::Duration;
use tokio::join;
use tracing::{error, info, Instrument, Span};
use tracing_subscriber::layer::SubscriberExt;
use tracing_subscriber::util::SubscriberInitExt;

#[derive(Debug)]
struct Request {
    id: u8,
    path: &'static str,
    method: &'static str,
    token: &'static str,
}

#[tokio::main]
async fn main() {
    // Create a Jaeger Tracer
    let tracer_jaeger = opentelemetry_jaeger::new_pipeline()
        .with_service_name("lab-telemetry")
        .install_simple()
        .unwrap();

    // Create a tracing layer with the configured tracer
    let opentelemetry_layer = tracing_opentelemetry::layer().with_tracer(tracer_jaeger);

    // Register Layers
    tracing_subscriber::registry()
        .with(opentelemetry_layer)
        .init();

    handle_requests().await;

    // Force all Span to be sent
    shutdown_tracer_provider();
}

#[tracing::instrument(name = "main")]
async fn handle_requests() {
    let request = Request {
        id: 42,
        path: "/home",
        method: "GET",
        token: "blabla",
    };

    let request2 = Request {
        id: 43,
        path: "/home",
        method: "GET",
        token: "bleubleu",
    };

    let fut1 = handle_request(request);
    let fut2 = async {
        thread::sleep(Duration::from_millis(100));
        handle_request(request2).await
    };

    let t1 = tokio::spawn(fut1.instrument(Span::current()));
    let t2 = tokio::spawn(fut2.instrument(Span::current()));

    let _ = join!(t1, t2);
}

#[tracing::instrument]
async fn handle_request(request: Request) {
    if auth(&request) {
        info!(request = ?request, id = request.id, "Request authenticated")
    } else {
        error!(request = ?request, id = request.id, "Request forbidden")
    }
}

#[tracing::instrument]
fn auth(request: &Request) -> bool {
    // Emulate call for db
    sleep(Duration::from_millis(500));
    request.token == "blabla"
}

Problèmes pouvant être rencontrés

La span n’est jamais envoyée.

Parfois nous trouvons dans des situations où une span devrait apparaître mais ne le fait jamais.

Pour comprendre cette situation, prenons le code suivant:

use opentelemetry::global::shutdown_tracer_provider;
use std::thread::sleep;
use std::time::Duration;
use tracing::info;

use tracing_subscriber::layer::SubscriberExt;
use tracing_subscriber::util::SubscriberInitExt;

fn main() {
    // Create a Jaeger Tracer
    let tracer_jaeger = opentelemetry_jaeger::new_pipeline()
        .with_service_name("infinite")
        .install_simple()
        .unwrap();

    // Create a tracing layer with the configured tracer
    let opentelemetry_layer = tracing_opentelemetry::layer()
        .with_tracer(tracer_jaeger)
        .with_threads(true);

    // Register Layers
    tracing_subscriber::registry()
        .with(opentelemetry_layer)
        .init();

    for_loop();
    infinite_loop();

    // Force all Span to be sent
    shutdown_tracer_provider();
}

#[tracing::instrument]
fn infinite_loop() {
    info!("Début de la boucle");
    let mut i = 0;
    loop {
        info!(iteration = i, "Itération");
        sleep(Duration::from_millis(1));
        i += 1;
    }
    info!("Fin de la boucle")
}

#[tracing::instrument]
fn for_loop() {
    info!("Début de la boucle");
    for i in 1..10 {
        info!(iteration = i, "Itération");
        sleep(Duration::from_millis(1));
    }
    info!("Fin de la boucle")
}

Si l’on éxécute celui-ci nous allons avoir le résultat suivant:

Contenant les différents events des itérations.

Par contre, aucune trace de la span infinite_loop.

Une span étant bornée temporellement, si celle-ci ne finit jamais, elle ne sera jamais envoyé au Tracer et par conséquent les Events qui y sont rattachés non plus.

Pour nous permettre de visualiser le comportement de notre fonctions infinite_loop. Nous allons devoir créer une span qui est bornée dans le temps et donc transmissible au Tracer.

Et finalement affichable dans Jaeger.

Pour ce faire nous allon utiliser une méthode qui se nomme in_scope.

info_span!("ma span").in_scope(||{
    info!("Dans ma span");
})

Tous les évènements qui sont déclenchés dans in_scope seront rattachés à ma span et envoyés sur Jaeger comme tel.

Remarque

Le code

info_span!("ma span").in_scope(||{
    info!("Dans ma span");
})

Est équivalent à

let span = info_span!("ma span");
{
    let _enter = span.enter();
    info!("Dans ma span");
}
drop(span);

Si drop(span) n’est pas appelé et que vous vous trouvez dans une boucle infinie.

Votre span ma span ne sera jamais envoyée!

Nous pouvons réécrire notre fonction infinite_loop ainsi:

#[tracing::instrument]
fn infinite_loop() {
    info_span!("infinite_loop").in_scope(|| {
        info!(flag = "start", "Début de la boucle");
    });

    let mut i = 0;
    loop {
        info_span!("infinite_loop").in_scope(|| {
            info!(iteration = i, "Itération");
        });

        sleep(Duration::from_millis(1));
        i += 1;
    }
    info!("Fin de la boucle")
}

Sur Jaeger nous avons désormais une span infinite_loop! 😎

Et en ouvrant son détails on peut voir que toutes les spans ayant le même nom sont regroupés sous la forme d’un bel escalier de spans.

Nous avons notre flag=start tout en haut et les iteration=N ensuite.

D’ailleurs nous pouvons créer le même genre de retour pour notre for_loop, cela permettra de visualiser le temps pris par chaque itération.

#[tracing::instrument]
fn for_loop() {
    info!("Début de la boucle");
    for i in 1..10 {
        info_span!("for_loop").in_scope(|| {
            info!(iteration = i, "Itération");
        });
        sleep(Duration::from_millis(1));
    }
    info!("Fin de la boucle")
}

Cette fois-ci sur Jaeger notre span for_loop devient:

On peut y voir les temps passés dans les in_scope de chaque itération.

Et si l’on ouvre les détails cela nous donne:

Nous avons bien nos events de début et de chaque itération.

Nous avons en prime l’event de fin car cette span est finie.

Nous pouvons également le réaliser en asynchrone.

use opentelemetry::global::shutdown_tracer_provider;
use std::thread::sleep;
use std::time::Duration;
use tokio::{join, spawn};
use tracing::{info, info_span};

use tracing_subscriber::layer::SubscriberExt;
use tracing_subscriber::util::SubscriberInitExt;

#[tokio::main]
async fn main() {
    // Create a Jaeger Tracer
    let tracer_jaeger = opentelemetry_jaeger::new_pipeline()
        .with_service_name("infinite")
        .install_simple()
        .unwrap();

    // Create a tracing layer with the configured tracer
    let opentelemetry_layer = tracing_opentelemetry::layer()
        .with_tracer(tracer_jaeger)
        .with_threads(true);

    // Register Layers
    tracing_subscriber::registry()
        .with(opentelemetry_layer)
        .init();

    let fut1 = spawn(for_loop());
    let fut2 = spawn(infinite_loop());

    let _ = join!(fut1, fut2);

    // Force all Span to be sent
    shutdown_tracer_provider();
}

#[tracing::instrument]
async fn infinite_loop() {
    info_span!("infinite_loop").in_scope(|| {
        info!(flag = "start", "Début de la boucle");
    });

    let mut i = 0;
    loop {
        info_span!("infinite_loop").in_scope(|| {
            info!(iteration = i, "Itération");
        });

        sleep(Duration::from_millis(1));
        i += 1;
    }
    info!("Fin de la boucle")
}

#[tracing::instrument]
async fn for_loop() {
    info!("Début de la boucle");
    for i in 1..10 {
        info_span!("for_loop").in_scope(|| {
            info!(iteration = i, "Itération");
        });
        sleep(Duration::from_millis(1));
    }
    info!("Fin de la boucle")
}

Mon Jaeger crash !

Dans certaines condition il se peut que votre Jeager crash, et qu’avant il prenne un nombre exponentiel de mémoire.

Ce cas, je l’ai observé lorsqu’un trop grand nombre de span sont envoyées en très peu de temps.

Vous risquez de vous retrouver avec ce genre de message d’erreur:

On peut aisément reproduire ce comportement grâce à ce code

fn main() {


    // Déclaration du tracing
    
    info_span!("will crash").in_scope(|| {
        for i in 0..1000000 {
            dummy(i)
        }
    });
}

#[tracing::instrument]
fn dummy(index: u64) {
    info!("I was here {}", index)
}

Celui-ci va engorger le système et finir par faire crasher le système de télémétrie.

Heureusement il est possible de régler ce soucis en configurant la manière dont les spans sont traitées.

En effet, il est possible d’affecter un niveau de log à nos spans.

#[tracing::instrument(level = "debug")]
fn dummy(index: u64) {
    debug!("I was here {}", index)
}

Par contre en l’état le système ne filtrera rien, il faut rajouter un filtre qui aura pour effet de venir conditionner le traitement ou non de notre span.

Pour en bénéficier, il faut activer une feature

cargo add tracing-subscriber --features env-filter

Ou via le Cargo.toml

tracing-subscriber = { version = "0.3.14", features = ["env-filter"] }

    // Register Layers
    tracing_subscriber::registry()
        .with(EnvFilter::from_default_env())
        .with(opentelemetry_layer)
        .init();

La 3ème ligne, indique d’utiliser la variable d’environnement standard RUST_LOG.

Ce qui signifie que l’on peut venir piloter les spans au travers d’elle.

RUST_LOG=info cargo run

Pour les besoin du test et pour des raisons de visibilité, nous allons réduire le nombre d’itérations à 10.

fn main() {


    // Déclaration du tracing
    
    info_span!("won't crash").in_scope(|| {
        for i in 0..10 {
            dummy(i)
        }
    });
}

Avec ce réglage vous allez voir dans Jaeger ceci:

Aucune span ne sera produite.

Par contre en éxécutant

RUST_LOG=debugcargo run

Vous verrez

Nous pouvons donc régler le degré de finesse de nos spans en fonction de nos besoins.

Que ce soit de la télémétrie de production ou de développement et donc du debug.

Récapitulatif de code:

use futures::future::join_all;
use futures::{Stream, StreamExt};
use opentelemetry::global::shutdown_tracer_provider;
use opentelemetry::runtime::Tokio;
use tracing::{debug, info_span};

use tracing_subscriber::EnvFilter;

use tracing_subscriber::layer::SubscriberExt;
use tracing_subscriber::util::SubscriberInitExt;

#[tokio::main]
async fn main() {
    // Create a Jaeger Tracer
    let tracer_jaeger = opentelemetry_jaeger::new_pipeline()
        .with_service_name("too much spans")
        .install_batch(Tokio)
        .unwrap();

    // Create a tracing layer with the configured tracer
    let opentelemetry_layer = tracing_opentelemetry::layer()
        .with_tracer(tracer_jaeger)
        .with_threads(true);

    // Register Layers
    tracing_subscriber::registry()
        .with(EnvFilter::from_default_env())
        .with(opentelemetry_layer)
        .init();

    info_span!("won't crash").in_scope(|| {
        for i in 0..10 {
            dummy(i)
        }
    });

    shutdown_tracer_provider();
}

#[tracing::instrument(level = "debug")]
fn dummy(index: u64) {
    debug!("I was here {}", index)
}

Un petit exemple concret

Nous allons réaliser un exemple qui va récapituler les différents concepts et les mettre en action dans un cas concret.

Le système que je vous propose sera un serveur web qui ira chercher des données sur un serveur distant. Et qui mettra en cache dans une base de données sqlite ces réponses de requêtes avant de servir le client.

Disclaimer

Ici le but n’est pas de créer quelque chose qui doit aller en prod, les décisions d’architectures peuvent être qualifier au mieux de discutables. 😅