libdd-crashtracker 3.0.0

// SPDX-License-Identifier: Apache-2.0
// Copyright 2023-Present Datadog, Inc. https://www.datadoghq.com/

use super::receive_report::receive_report_from_stream;
use crate::crash_info::CrashInfo;
use crate::CrashtrackerConfiguration;
#[cfg(target_os = "linux")]
use crate::StacktraceCollection;
use anyhow::Context;
use std::time::Duration;
use tokio::{
    io::{AsyncBufReadExt, BufReader},
    net::UnixListener,
};

/*-----------------------------------------
|                Public API               |
------------------------------------------*/

pub fn receiver_entry_point_stdin() -> anyhow::Result<()> {
    let stream = BufReader::new(tokio::io::stdin());
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()?;
    rt.block_on(receiver_entry_point(receiver_timeout(), stream))?;
    Ok(())
}

pub async fn async_receiver_entry_point_unix_listener(
    listener: &UnixListener,
) -> anyhow::Result<()> {
    let (unix_stream, _) = listener.accept().await?;
    let stream = BufReader::new(unix_stream);
    receiver_entry_point(receiver_timeout(), stream).await
}

pub async fn async_receiver_entry_point_stream(
    stream: impl AsyncBufReadExt + std::marker::Unpin,
) -> anyhow::Result<()> {
    receiver_entry_point(receiver_timeout(), stream).await
}

pub async fn async_receiver_entry_point_unix_socket(
    socket_path: impl AsRef<str>,
    one_shot: bool,
) -> anyhow::Result<()> {
    let listener = get_receiver_unix_socket(socket_path)?;
    loop {
        let res = async_receiver_entry_point_unix_listener(&listener).await;
        // TODO, should we log failures somewhere?
        if one_shot {
            return res;
        }
    }
}

pub fn receiver_entry_point_unix_socket(socket_path: impl AsRef<str>) -> anyhow::Result<()> {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()?;
    rt.block_on(async_receiver_entry_point_unix_socket(socket_path, true))?;
    Ok(())
    // Dropping the stream closes it, allowing the collector to exit if it was waiting.
}

pub fn get_receiver_unix_socket(socket_path: impl AsRef<str>) -> anyhow::Result<UnixListener> {
    fn path_bind(socket_path: impl AsRef<str>) -> anyhow::Result<UnixListener> {
        let socket_path = socket_path.as_ref();
        if std::fs::metadata(socket_path).is_ok() {
            std::fs::remove_file(socket_path)
                .with_context(|| format!("could not delete previous socket at {socket_path:?}"))?;
        }
        Ok(UnixListener::bind(socket_path)?)
    }

    #[cfg(target_os = "linux")]
    let unix_listener = if socket_path.as_ref().starts_with(['.', '/']) {
        path_bind(socket_path)
    } else {
        use std::os::linux::net::SocketAddrExt;
        std::os::unix::net::SocketAddr::from_abstract_name(socket_path.as_ref())
            .and_then(|addr| {
                std::os::unix::net::UnixListener::bind_addr(&addr)
                    .and_then(|listener| {
                        listener.set_nonblocking(true)?;
                        Ok(listener)
                    })
                    .and_then(UnixListener::from_std)
            })
            .map_err(anyhow::Error::msg)
    };
    #[cfg(not(target_os = "linux"))]
    let unix_listener = path_bind(socket_path);
    unix_listener.context("Could not create the unix socket")
}

/// Receives data from a crash collector via a stream, formats it into
/// `CrashInfo` json, and emits it to the endpoint/file defined in `config`.
///
/// At a high-level, this exists because doing anything in a
/// signal handler is dangerous, so we fork a sidecar to do the stuff we aren't
/// allowed to do in the handler.
///
/// See comments in [libdd-crashtracker/lib.rs] for a full architecture
/// description.
pub(crate) async fn receiver_entry_point(
    timeout: Duration,
    mut stream: impl AsyncBufReadExt + std::marker::Unpin,
) -> anyhow::Result<()> {
    if let Some((config, mut crash_info)) = receive_report_from_stream(timeout, &mut stream).await?
    {
        // Symbolization reads /proc/<pid>/maps, and the crashing process is
        // waiting on POLLHUP from this connection to terminate. Hold it open
        // until the report is symbolized, then release the process before the
        // upload so a slow endpoint does not extend the crash pause.
        if let Err(e) = resolve_frames(&config, &mut crash_info) {
            crash_info
                .log_messages
                .push(format!("Error resolving frames: {e}"));
        }
        drop(stream);

        if config.demangle_names() {
            if let Err(e) = crash_info.demangle_names() {
                crash_info
                    .log_messages
                    .push(format!("Error demangling names: {e}"));
            }
        }
        crash_info
            .async_upload_to_endpoint(config.endpoint())
            .await?;
    }
    Ok(())
}

fn receiver_timeout() -> Duration {
    // https://github.com/DataDog/libdatadog/issues/717
    if let Ok(s) = std::env::var("DD_CRASHTRACKER_RECEIVER_TIMEOUT_MS") {
        if let Ok(v) = s.parse() {
            return Duration::from_millis(v);
        }
    }
    // Default value
    Duration::from_millis(4000)
}

fn resolve_frames(
    config: &CrashtrackerConfiguration,
    crash_info: &mut CrashInfo,
) -> anyhow::Result<()> {
    // enrich_callstacks uses blazesym's normalize_user_addrs (reads /proc/<pid>/maps)
    // and assumes ELF binaries. Both are Linux-specific; macOS has no procfs and
    // uses Mach-O binaries.
    #[cfg(target_os = "linux")]
    if config.resolve_frames() == StacktraceCollection::EnabledWithSymbolsInReceiver {
        let pid = crash_info
            .proc_info
            .as_ref()
            .context("Unable to resolve frames: No PID specified")?
            .pid;
        crash_info.enrich_callstacks(pid)?;
    }
    #[cfg(not(target_os = "linux"))]
    let _ = (config, crash_info);
    Ok(())
}