8000 chore: add tracing instrumentation to execution-path async fns by norrietaylor · Pull Request #221 · gominimal/minimal · GitHub
[go: up one dir, main page]
Skip to content

chore: add tracing instrumentation to execution-path async fns #221

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
twitchyliquid64 merged 1 commit into from
May 24, 2026
Merged

chore: add tracing instrumentation to execution-path async fns #221

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 1 addition & 0 deletions crates/check/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ regex.workspace = true
tempfile.workspace = true
tokio.workspace = true
moka.workspace = true
tracing.workspace = true

common.workspace = true
decode.workspace = true
Expand Down
7 changes: 7 additions & 0 deletions crates/check/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -371,6 +371,7 @@ impl CheckCtx {
///
/// If `None` is given for a `packages_dir`, `profiles_dir`, or `stacks_dir`, the corresponding
/// class of checkers are skipped.
#[tracing::instrument(skip_all, err)]
pub fn run_checks(
packages_dir: Option<PathBuf>,
profiles_dir: Option<PathBuf>,
Expand Down Expand Up @@ -402,6 +403,7 @@ pub fn run_checks(
.collect::<FuturesUnordered<_>>())
}

#[tracing::instrument(skip_all, err)]
fn package_check_futures(packages_dir: PathBuf, ctx: CheckCtx) -> Result<Vec<CheckFuture>, Error> {
let package_dirs: Vec<_> = match decode::build_decls_in_dir(&packages_dir) {
Ok(i) => i,
Expand Down Expand Up @@ -434,6 +436,7 @@ fn package_check_futures(packages_dir: PathBuf, ctx: CheckCtx) -> Result<Vec<Che
.collect::<Vec<_>>())
}

#[tracing::instrument(skip_all, err)]
fn profile_check_futures(profiles_dir: PathBuf, ctx: CheckCtx) -> Result<Vec<CheckFuture>, Error> {
let dirs: Vec<std::ffi::OsString> = match std::fs::read_dir(&profiles_dir) {
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(vec![]),
Expand Down Expand Up @@ -482,6 +485,7 @@ fn profile_check_futures(profiles_dir: PathBuf, ctx: CheckCtx) -> Result<Vec<Che
.collect())
}

#[tracing::instrument(skip_all, err)]
fn stack_check_futures(stacks_dir: PathBuf, ctx: CheckCtx) -> Result<Vec<CheckFuture>, Error> {
let dirs: Vec<std::ffi::OsString> = match std::fs::read_dir(&stacks_dir) {
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(vec![]),
Expand Down Expand Up @@ -529,6 +533,7 @@ fn stack_check_futures(stacks_dir: PathBuf, ctx: CheckCtx) -> Result<Vec<CheckFu
.collect())
}

#[tracing::instrument(skip_all, fields(pkg = %pkg), err)]
async fn check_package(
pkg: String,
package_dir: PathBuf,
Expand Down Expand Up @@ -583,8 +588,10 @@ async fn check_package(
let file_based_results = {
let ctx = ctx.clone();
let pkg = pkg.clone();
let span = tracing::Span::current();

tokio::task::spawn_blocking(move || {
let _enter = span.enter();
let file_based: Vec<Box<dyn FileBasedChecker>> = vec![
Box::new(ParseCheck),
Box::new(ImportLineCheck),
Expand Down
3 changes: 3 additions & 0 deletions crates/op/src/oci_image.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@ pub struct OciImageCreate {
impl Runnable for OciImageCreate {
type Result = ();

#[tracing::instrument(skip_all, err)]
async fn run<'b>(&mut self, opts: &Options<'b>) -> Result<Self::Result, Error> {
let mut all_deps: Vec<(BuildSpecRef, TransitivesDep)> =
Transitives::for_toplevels(opts.graph, opts.graph.top_levels.to_vec(), false)
Expand Down Expand Up @@ -281,6 +282,7 @@ impl BuiltLayer {
}
}

#[tracing::instrument(err)]
async fn create_base_layer() -> anyhow::Result<BuiltLayer> {
let enc = GzEncoder::new(tempfile::tempfile()?, Compression::best());
let mut tar = tar::Builder::new(enc);
Expand Down Expand Up @@ -328,6 +330,7 @@ async fn create_base_layer() -> anyhow::Result<BuiltLayer> {
})
}

#[tracing::instrument(skip_all, fields(package_name = %package_name), err)]
async fn create_layer_from_cache(
cache: &lcache::Cache<lcache::LocalDir>,
spec_hash: &SpecHash,
Expand Down
4 changes: 4 additions & 0 deletions crates/rcache/src/remote.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,7 @@ pub const INDEX_FILENAME: &str = "index.shisha";
const INDEX_EXPIRY_SECONDS: u64 = 5 * 60; // how long a fetch of the remote index is considered fresh for

impl RemoteCache<Client> {
#[tracing::instrument(skip_all, err)]
pub async fn new_over_https<URL: Into<ReqwestUrl>>(
url: URL,
index_dir: Option<PathBuf>,
Expand All @@ -91,6 +92,7 @@ impl RemoteCache<Client> {

impl RemoteCache<Storage> {
/// Instantiates a new remote cache using the given GCS client + bucket.
#[tracing::instrument(skip_all, fields(bucket_id = %bucket_id), err)]
pub async fn new_with_gcs_bucket(
storage: Storage,
bucket_id: &str,
Expand All @@ -110,6 +112,7 @@ impl RemoteCache<Storage> {
/// cache fast path (and therefore has no recorded generation), the index
/// is refetched from GCS so the writer's compare-and-swap commit has an
/// authoritative generation to match against.
#[tracing::instrument(skip_all, err)]
pub async fn into_writer(
self,
ot: Option<OpTracker>,
Expand Down Expand Up @@ -218,6 +221,7 @@ impl<B: FetchBackend> RemoteCache<B> {
}

/// Download the given spec hash into the local cache.
#[tracing::instrument(skip_all, fields(span_name = %span_name), err)]
pub async fn materialize(
&self,
spec_hash: &SpecHash,
Expand Down
2 changes: 2 additions & 0 deletions crates/rcache/src/remote_writer.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -136,6 +136,7 @@ impl<S: StubStorage + 'static> RemoteCacheWriter<S> {
/// The artifact blob upload is content-addressed, so concurrent uploads
/// of the same blob bytes are safe (last-writer-wins on identical content).
/// The index update is deferred to [Self::finish_uploads].
#[tracing::instrument(skip_all, err)]
pub async fn upload(
&mut self,
spec_hash: &SpecHash,
Expand Down Expand Up @@ -195,6 +196,7 @@ impl<S: StubStorage + 'static> RemoteCacheWriter<S> {
///
/// Bounded by [MAX_INDEX_WRITE_RETRIES] attempts with exponential backoff
/// + jitter; after exhaustion the underlying GCS error is returned.
#[tracing::instrument(skip_all, err)]
pub async fn finish_uploads(self) -> Result<(), Error<GcsError>> {
let Self {
backend,
Expand Down
1 change: 1 addition & 0 deletions crates/remote-client/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ http-body.workspace = true
tokio-stream.workspace = true
tokio-util.workspace = true
tonic.workspace = true
tracing.workspace = true

check.workspace = true
common.workspace = true
Expand Down
178 changes: 97 additions & 81 deletions crates/remote-client/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ use remote_proto::{res::*, *};
use tempfile::NamedTempFile;
use tokio::io::{AsyncWrite, AsyncWriteExt};
use tonic::transport::{Channel, Error as TransportError};
use tracing::Instrument;

type CreateEnvStream = Pin<Box<dyn futures::Stream<Item = CreateEnvMessage> + Send>>;
type CreateBuildStream = Pin<Box<dyn futures::Stream<Item = OrchestrateBuildMessage> + Send>>;
Expand Down Expand Up @@ -75,6 +76,7 @@ where
<T::ResponseBody as http_body::Body>::Error:
Into<Box<dyn std::error::Error + Send + Sync>> + Send,
{
#[tracing::instrument(skip_all)]
pub async fn make_env<'b>(&mut self, cwd: Worktree<'b>) -> Result<Env<'b>, Error> {
use create_env_message::*;
let client_id = format!(
Expand All @@ -87,12 +89,15 @@ where
use stream_config::*;
let (graph_writer, graph_reader) = tokio::io::duplex(8 * 1024);
let graph_clone = self.g.clone();
tokio::spawn(async move {
graph::wire::AsyncGraphWriter::new(graph_writer)
.write_graph(&graph_clone)
.await
.expect("graph wire serialization failed");
});
tokio::spawn(
async move {
graph::wire::AsyncGraphWriter::new(graph_writer)
.write_graph(&graph_clone)
.await
.expect("graph wire serialization failed");
}
.instrument(tracing::Span::current()),
);
let graph_stream: (StreamConfig, CreateEnvStream) = (
StreamConfig {
format: Some(Format::Graph(GraphFormat::GfStreamingV1.into())),
Expand Down Expand Up @@ -204,6 +209,7 @@ where
})
}

#[tracing::instrument(skip_all)]
pub async fn exec<'b>(
&mut self,
env: &Env<'b>,
Expand Down Expand Up @@ -275,12 +281,15 @@ where
use stream_config::*;
let (graph_writer, graph_reader) = tokio::io::duplex(8 * 1024);
let graph_clone = self.g.clone();
tokio::spawn(async move {
graph::wire::AsyncGraphWriter::new(graph_writer)
.write_graph(&graph_clone)
.await
.expect("graph wire serialization failed");
});
tokio::spawn(
async move {
graph::wire::AsyncGraphWriter::new(graph_writer)
.write_graph(&graph_clone)
.await
.expect("graph wire serialization failed");
}
.instrument(tracing::Span::current()),
);
let (graph_config, graph_stream): (StreamConfig, CreateBuildStream) = (
StreamConfig {
format: Some(Format::Graph(GraphFormat::GfStreamingV1.into())),
Expand Down Expand Up @@ -490,81 +499,88 @@ where
.into_inner();

let (tx, rx) = mpsc::unbounded();
tokio::spawn(async move {
while let Some(msg) = rpc.next().await {
let msg = match msg {
Ok(m) => m,
Err(e) => {
let _ = tx.unbounded_send((
check::CheckObj::Package(String::new()),
Err(Error::Other(e.into())),
));
break;
}
};
match msg.msg {
Some(check_response::Msg::Error(e)) => {
let _ = tx.unbounded_send((
check::CheckObj::Package(String::new()),
Err(Error::Other(anyhow!("check failed: {}", e.msg).into())),
));
break;
}
Some(check_response::Msg::Check(c)) => {
let obj = match c.obj.and_then(|o| o.msg) {
Some(check_object::Msg::Package(n)) => check::CheckObj::Package(n),
Some(check_object::Msg::Profile(n)) => check::CheckObj::Profile(n),
Some(check_object::Msg::Harness(n)) => check::CheckObj::Stack(n),
None => {
let _ = tx.unbounded_send((
check::CheckObj::Package(String::new()),
Err(Error::Other(
anyhow!("check response missing object").into(),
)),
));
tokio::spawn(
async move {
while let Some(msg) = rpc.next().await {
let msg = match msg {
Ok(m) => m,
Err(e) => {
let _ = tx.unbounded_send((
check::CheckObj::Package(String::new()),
Err(Error::Other(e.into())),
));
break;
}
};
match msg.msg {
Some(check_response::Msg::Error(e)) => {
let _ = tx.unbounded_send((
check::CheckObj::Package(String::new()),
Err(Error::Other(anyhow!("check failed: {}", e.msg).into())),
));
break;
}
Some(check_response::Msg::Check(c)) => {
let obj = match c.obj.and_then(|o| o.msg) {
Some(check_object::Msg::Package(n)) => check::CheckObj::Package(n),
Some(check_object::Msg::Profile(n)) => check::CheckObj::Profile(n),
Some(check_object::Msg::Harness(n)) => check::CheckObj::Stack(n),
None => {
let _ = tx.unbounded_send((
check::CheckObj::Package(String::new()),
Err(Error::Other(
anyhow!("check response missing object").into(),
)),
));
break;
}
};
let check_results: Result<Vec<_>, Error> = c
.results
.into_iter()
.map(|r| -> Result<check::CheckResult, Error> {
let verdict =
match check_result::CheckVerdict::try_from(r.verdict) {
Ok(check_result::CheckVerdict::CvFail) => {
check::CheckVerdict::Fail
}
Ok(check_result::CheckVerdict::CvFixed) => {
check::CheckVerdict::Fixed
}
Ok(check_result::CheckVerdict::CvSkip) => {
check::CheckVerdict::Skip
}
Ok(check_result::CheckVerdict::CvPass) => {
check::CheckVerdict::Pass
}
Ok(check_result::CheckVerdict::CvUnspecified)
| Err(_) => {
return Err(Error::Other(
anyhow!(
"invalid check verdict value {}",
r.verdict
)
.into(),
));
}
};
Ok(check::CheckResult {
check: r.check.into(),
verdict,
err: r.errors,
})
})
.collect();
if tx.unbounded_send((obj, check_results)).is_err() {
break;
}
};
let check_results: Result<Vec<_>, Error> = c
.results
.into_iter()
.map(|r| -> Result<check::CheckResult, Error> {
let verdict = match check_result::CheckVerdict::try_from(r.verdict)
{
Ok(check_result::CheckVerdict::CvFail) => {
check::CheckVerdict::Fail
}
Ok(check_result::CheckVerdict::CvFixed) => {
check::CheckVerdict::Fixed
}
Ok(check_result::CheckVerdict::CvSkip) => {
check::CheckVerdict::Skip
}
Ok(check_result::CheckVerdict::CvPass) => {
check::CheckVerdict::Pass
}
Ok(check_result::CheckVerdict::CvUnspecified) | Err(_) => {
return Err(Error::Other(
anyhow!("invalid check verdict value {}", r.verdict)
.into(),
));
}
};
Ok(check::CheckResult {
check: r.check.into(),
verdict,
err: r.errors,
})
})
.collect();
if tx.unbounded_send((obj, check_results)).is_err() {
break;
}
None => {}
}
None => {}
}
}
});
.instrument(tracing::Span::current()),
);

Ok(rx)
}
Expand Down
Loading
0