Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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 .github/workflows/main.yml
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
on:

Check warning on line 1 in .github/workflows/main.yml

View workflow job for this annotation

GitHub Actions / zizmor-output

excessive-permissions

main.yml:1: overly broad permissions: default permissions used due to no permissions: block
push:
branches:
- main
Expand All @@ -16,11 +16,12 @@
cancel-in-progress: "${{ github.ref != 'refs/heads/main' }}"

jobs:
test:

Check warning on line 19 in .github/workflows/main.yml

View workflow job for this annotation

GitHub Actions / zizmor-output

excessive-permissions

main.yml:19: overly broad permissions: default permissions used due to no permissions: block
name: Test
runs-on: ubuntu-latest
strategy:
matrix:
opentelemetry: ["", "--no-default-features"]
serde: ["--features serde1", ""]
tokio: ["--features tokio1", ""]
serde-transport: ["--features serde-transport", ""]
Expand All @@ -42,22 +43,23 @@
- unix: "--features unix"
serde-transport: ""
steps:
- uses: actions/checkout@v6

Check failure on line 46 in .github/workflows/main.yml

View workflow job for this annotation

GitHub Actions / zizmor-output

zizmor/unpinned-uses

unpinned action reference: action is not pinned to a hash (required by blanket policy)

Check failure on line 46 in .github/workflows/main.yml

View workflow job for this annotation

GitHub Actions / zizmor-output

unpinned-uses

main.yml:46: unpinned action reference: action is not pinned to a hash (required by blanket policy)
- uses: dtolnay/rust-toolchain@stable

Check failure on line 47 in .github/workflows/main.yml

View workflow job for this annotation

GitHub Actions / zizmor-output

zizmor/unpinned-uses

unpinned action reference: action is not pinned to a hash (required by blanket policy)

Check failure on line 47 in .github/workflows/main.yml

View workflow job for this annotation

GitHub Actions / zizmor-output

unpinned-uses

main.yml:47: unpinned action reference: action is not pinned to a hash (required by blanket policy)
- run: >
cargo test --manifest-path tarpc/Cargo.toml
${{ matrix.opentelemetry }}
${{ matrix.serde }} ${{ matrix.tokio }} ${{ matrix.serde-transport }}
${{ matrix.serde-transport-json }} ${{ matrix.serde-transport-bincode }}
${{ matrix.tcp }} ${{ matrix.unix }}

list-examples:

Check warning on line 55 in .github/workflows/main.yml

View workflow job for this annotation

GitHub Actions / zizmor-output

excessive-permissions

main.yml:55: overly broad permissions: default permissions used due to no permissions: block
name: List Examples
runs-on: ubuntu-latest
outputs:
examples: ${{ steps.matrix.outputs.examples }}
steps:
- uses: actions/checkout@v6

Check failure on line 61 in .github/workflows/main.yml

View workflow job for this annotation

GitHub Actions / zizmor-output

zizmor/unpinned-uses

unpinned action reference: action is not pinned to a hash (required by blanket policy)

Check failure on line 61 in .github/workflows/main.yml

View workflow job for this annotation

GitHub Actions / zizmor-output

unpinned-uses

main.yml:61: unpinned action reference: action is not pinned to a hash (required by blanket policy)
- uses: dtolnay/rust-toolchain@stable

Check failure on line 62 in .github/workflows/main.yml

View workflow job for this annotation

GitHub Actions / zizmor-output

zizmor/unpinned-uses

unpinned action reference: action is not pinned to a hash (required by blanket policy)

Check failure on line 62 in .github/workflows/main.yml

View workflow job for this annotation

GitHub Actions / zizmor-output

unpinned-uses

main.yml:62: unpinned action reference: action is not pinned to a hash (required by blanket policy)
- id: matrix
run: |
examples=$(
Expand All @@ -71,7 +73,7 @@
)
echo "examples=$examples" | tee -a $GITHUB_OUTPUT

run-example:

Check warning on line 76 in .github/workflows/main.yml

View workflow job for this annotation

GitHub Actions / zizmor-output

excessive-permissions

main.yml:76: overly broad permissions: default permissions used due to no permissions: block
name: Run Example
needs: list-examples
runs-on: ubuntu-latest
Expand All @@ -79,26 +81,26 @@
matrix:
example: ${{ fromJSON(needs.list-examples.outputs.examples) }}
steps:
- uses: actions/checkout@v6

Check failure on line 84 in .github/workflows/main.yml

View workflow job for this annotation

GitHub Actions / zizmor-output

zizmor/unpinned-uses

unpinned action reference: action is not pinned to a hash (required by blanket policy)

Check failure on line 84 in .github/workflows/main.yml

View workflow job for this annotation

GitHub Actions / zizmor-output

unpinned-uses

main.yml:84: unpinned action reference: action is not pinned to a hash (required by blanket policy)
- uses: dtolnay/rust-toolchain@stable

Check failure on line 85 in .github/workflows/main.yml

View workflow job for this annotation

GitHub Actions / zizmor-output

zizmor/unpinned-uses

unpinned action reference: action is not pinned to a hash (required by blanket policy)

Check failure on line 85 in .github/workflows/main.yml

View workflow job for this annotation

GitHub Actions / zizmor-output

unpinned-uses

main.yml:85: unpinned action reference: action is not pinned to a hash (required by blanket policy)
- run: |
cargo run --example "${{ matrix.example }}"

Check failure on line 87 in .github/workflows/main.yml

View workflow job for this annotation

GitHub Actions / zizmor-output

zizmor/template-injection

code injection via template expansion: may expand into attacker-controllable code

Check failure on line 87 in .github/workflows/main.yml

View workflow job for this annotation

GitHub Actions / zizmor-output

template-injection

main.yml:87: code injection via template expansion: may expand into attacker-controllable code

fmt:

Check warning on line 89 in .github/workflows/main.yml

View workflow job for this annotation

GitHub Actions / zizmor-output

excessive-permissions

main.yml:89: overly broad permissions: default permissions used due to no permissions: block
name: Rustfmt
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v6

Check failure on line 93 in .github/workflows/main.yml

View workflow job for this annotation

GitHub Actions / zizmor-output

zizmor/unpinned-uses

unpinned action reference: action is not pinned to a hash (required by blanket policy)

Check failure on line 93 in .github/workflows/main.yml

View workflow job for this annotation

GitHub Actions / zizmor-output

unpinned-uses

main.yml:93: unpinned action reference: action is not pinned to a hash (required by blanket policy)
- uses: dtolnay/rust-toolchain@stable

Check failure on line 94 in .github/workflows/main.yml

View workflow job for this annotation

GitHub Actions / zizmor-output

zizmor/unpinned-uses

unpinned action reference: action is not pinned to a hash (required by blanket policy)

Check failure on line 94 in .github/workflows/main.yml

View workflow job for this annotation

GitHub Actions / zizmor-output

unpinned-uses

main.yml:94: unpinned action reference: action is not pinned to a hash (required by blanket policy)
with:
components: rustfmt
- run: cargo fmt --all -- --check

clippy:

Check warning on line 99 in .github/workflows/main.yml

View workflow job for this annotation

GitHub Actions / zizmor-output

excessive-permissions

main.yml:99: overly broad permissions: default permissions used due to no permissions: block
name: Clippy
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v6

Check failure on line 103 in .github/workflows/main.yml

View workflow job for this annotation

GitHub Actions / zizmor-output

unpinned-uses

main.yml:103: unpinned action reference: action is not pinned to a hash (required by blanket policy)
- uses: dtolnay/rust-toolchain@stable
with:
components: clippy
Expand All @@ -106,7 +108,7 @@

# This job succeeds if all other tests and examples succeed. Otherwise, it fails. It is for use in
# branch protection rules.
test-suite:

Check warning on line 111 in .github/workflows/main.yml

View workflow job for this annotation

GitHub Actions / zizmor-output

excessive-permissions

main.yml:111: overly broad permissions: default permissions used due to no permissions: block
name: Test Suite
runs-on: ubuntu-latest
needs: [test, run-example, fmt, clippy]
Expand Down
15 changes: 11 additions & 4 deletions tarpc/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,13 @@ readme = "README.md"
description = "An RPC framework for Rust with a focus on ease of use."

[features]
default = []
default = ["opentelemetry"]

opentelemetry = [
"dep:opentelemetry",
"dep:opentelemetry-semantic-conventions",
"dep:tracing-opentelemetry",
]

serde1 = ["tarpc-plugins/serde1", "serde", "serde/derive", "serde/rc"]
tokio1 = ["tokio/rt"]
Expand All @@ -28,6 +34,7 @@ tcp = ["tokio/net"]
unix = ["tokio/net"]

full = [
"opentelemetry",
"serde1",
"tokio1",
"serde-transport",
Expand Down Expand Up @@ -57,9 +64,9 @@ tracing = { version = "0.1", default-features = false, features = [
"attributes",
"log",
] }
tracing-opentelemetry = { version = "0.33", default-features = false }
opentelemetry = { version = "0.32", default-features = false }
opentelemetry-semantic-conventions = "0.32"
tracing-opentelemetry = { optional = true, version = "0.33", default-features = false }
opentelemetry = { optional = true, version = "0.32", default-features = false }
opentelemetry-semantic-conventions = { optional = true, version = "0.32" }

[dev-dependencies]
anyhow = "1.0"
Expand Down
65 changes: 33 additions & 32 deletions tarpc/src/context.rs
Original file line number Diff line number Diff line change
Expand Up @@ -8,13 +8,11 @@
//! client to server and is used by the server to enforce response deadlines.

use crate::trace::{self, TraceId};
use opentelemetry::trace::TraceContextExt;
use static_assertions::assert_impl_all;
use std::{
convert::TryFrom,
time::{Duration, Instant},
};
use tracing_opentelemetry::OpenTelemetrySpanExt;

/// A request context that carries request-scoped information like deadlines and trace information.
/// It is sent from client to server and is used by the server to enforce response deadlines.
Expand Down Expand Up @@ -102,28 +100,14 @@ pub fn current() -> Context {
Context::current()
}

#[derive(Clone)]
struct Deadline(Instant);

impl Default for Deadline {
fn default() -> Self {
Self(ten_seconds_from_now())
}
}

impl Context {
/// Returns the context for the current request, or a default Context if no request is active.
pub fn current() -> Self {
let span = tracing::Span::current();
Self {
trace_context: trace::Context::try_from(&span)
.unwrap_or_else(|_| trace::Context::default()),
deadline: span
.context()
.get::<Deadline>()
.cloned()
.unwrap_or_default()
.0,
deadline: span.deadline().unwrap_or_else(ten_seconds_from_now),
}
}

Expand All @@ -135,26 +119,43 @@ impl Context {

/// An extension trait for [`tracing::Span`] for propagating tarpc Contexts.
pub(crate) trait SpanExt {
fn trace_context(&self) -> Result<trace::Context, trace::NoActiveSpan>;
fn deadline(&self) -> Option<Instant>;
/// Sets the given context on this span. Newly-created spans will be children of the given
/// context's trace context.
fn set_context(&self, context: &Context);
}

#[cfg(not(feature = "opentelemetry"))]
impl SpanExt for tracing::Span {
fn set_context(&self, context: &Context) {
// Explicitly ignore the returned result because it either means that the span has
// already started, or the Otel layer is not present, so we don't mind if the result
// is an error we silently ignore.
let _ = self.set_parent(
opentelemetry::Context::new()
.with_remote_span_context(opentelemetry::trace::SpanContext::new(
opentelemetry::trace::TraceId::from(context.trace_context.trace_id),
opentelemetry::trace::SpanId::from(context.trace_context.span_id),
opentelemetry::trace::TraceFlags::from(context.trace_context.sampling_decision),
true,
opentelemetry::trace::TraceState::default(),
))
.with_value(Deadline(context.deadline)),
);
fn trace_context(&self) -> Result<trace::Context, trace::NoActiveSpan> {
Err(trace::NoActiveSpan)
}

fn deadline(&self) -> Option<Instant> {
None
}

fn set_context(&self, _: &Context) {
// Plain tracing spans cannot store a distributed context; the request still carries it.
}
}

#[cfg(test)]
mod tests {
use super::current;
use crate::trace;
use std::time::{Duration, Instant};

#[test]
fn current_without_subscriber() {
tracing::subscriber::with_default(tracing::subscriber::NoSubscriber::default(), || {
let earliest_deadline = Instant::now() + Duration::from_secs(10);
let context = current();
let latest_deadline = Instant::now() + Duration::from_secs(10);

assert_eq!(context.trace_context, trace::Context::default());
assert!((earliest_deadline..=latest_deadline).contains(&context.deadline));
});
}
}
2 changes: 2 additions & 0 deletions tarpc/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -244,6 +244,8 @@ pub use tarpc_plugins::service;
pub(crate) mod cancellations;
pub mod client;
pub mod context;
#[cfg(feature = "opentelemetry")]
mod otel;
pub mod server;
pub mod transport;
pub(crate) mod util;
Expand Down
144 changes: 144 additions & 0 deletions tarpc/src/otel.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,144 @@
// Copyright 2026 Google LLC
//
// Use of this source code is governed by an MIT-style
// license that can be found in the LICENSE file or at
// https://opensource.org/licenses/MIT.

use crate::{
context::{self, SpanExt},
trace::{Context, NoActiveSpan, SamplingDecision, SpanId, TraceId},
};
use opentelemetry::trace::TraceContextExt;
use std::time::Instant;
use tracing_opentelemetry::OpenTelemetrySpanExt;

// The deadline travels with the OpenTelemetry context but is interpreted by tarpc.
struct Deadline(Instant);

impl SpanExt for tracing::Span {
fn trace_context(&self) -> Result<Context, NoActiveSpan> {
let context = self.context();
if context.has_active_span() {
Ok(Context::from(context.span()))
} else {
Err(NoActiveSpan)
}
}

fn deadline(&self) -> Option<Instant> {
self.context().get::<Deadline>().map(|deadline| deadline.0)
}

fn set_context(&self, context: &context::Context) {
// Explicitly ignore the returned result because it either means that the span has
// already started, or the Otel layer is not present, so we don't mind if the result
// is an error we silently ignore.
let _ = self.set_parent(
opentelemetry::Context::new()
.with_remote_span_context(opentelemetry::trace::SpanContext::new(
opentelemetry::trace::TraceId::from(context.trace_context.trace_id),
opentelemetry::trace::SpanId::from(context.trace_context.span_id),
opentelemetry::trace::TraceFlags::from(context.trace_context.sampling_decision),
true,
opentelemetry::trace::TraceState::default(),
))
.with_value(Deadline(context.deadline)),
);
}
}

impl From<opentelemetry::trace::TraceId> for TraceId {
fn from(trace_id: opentelemetry::trace::TraceId) -> Self {
Self::from(u128::from_be_bytes(trace_id.to_bytes()))
}
}

impl From<TraceId> for opentelemetry::trace::TraceId {
fn from(trace_id: TraceId) -> Self {
Self::from_bytes(u128::from(trace_id).to_be_bytes())
}
}

impl From<opentelemetry::trace::SpanId> for SpanId {
fn from(span_id: opentelemetry::trace::SpanId) -> Self {
Self::from(u64::from_be_bytes(span_id.to_bytes()))
}
}

impl From<SpanId> for opentelemetry::trace::SpanId {
fn from(span_id: SpanId) -> Self {
Self::from_bytes(u64::from(span_id).to_be_bytes())
}
}

impl From<opentelemetry::trace::SpanRef<'_>> for Context {
fn from(span: opentelemetry::trace::SpanRef<'_>) -> Self {
let otel_ctx = span.span_context();
Self {
trace_id: TraceId::from(otel_ctx.trace_id()),
span_id: SpanId::from(otel_ctx.span_id()),
sampling_decision: SamplingDecision::from(otel_ctx),
}
}
}

impl From<SamplingDecision> for opentelemetry::trace::TraceFlags {
fn from(decision: SamplingDecision) -> Self {
match decision {
SamplingDecision::Sampled => opentelemetry::trace::TraceFlags::SAMPLED,
SamplingDecision::Unsampled => opentelemetry::trace::TraceFlags::default(),
}
}
}

impl From<&opentelemetry::trace::SpanContext> for SamplingDecision {
fn from(context: &opentelemetry::trace::SpanContext) -> Self {
if context.is_sampled() {
SamplingDecision::Sampled
} else {
SamplingDecision::Unsampled
}
}
}

#[cfg(test)]
mod tests {
use super::SpanExt;
use crate::{context, trace};
use opentelemetry::trace::TracerProvider as _;
use std::time::{Duration, Instant};
use tracing_subscriber::layer::SubscriberExt;

#[test]
fn current_inherits_request_context() {
let provider = opentelemetry_sdk::trace::SdkTracerProvider::builder().build();
let subscriber = tracing_subscriber::registry()
.with(tracing_opentelemetry::layer().with_tracer(provider.tracer("tarpc-test")));

tracing::subscriber::with_default(subscriber, || {
let mut parent = context::current();
parent.deadline = Instant::now() + Duration::from_secs(30);
parent.trace_context = trace::Context {
trace_id: 42.into(),
span_id: 7.into(),
sampling_decision: trace::SamplingDecision::Sampled,
};

let span = tracing::info_span!("RPC");
span.set_context(&parent);
let _entered = span.enter();
let current = context::current();

assert_eq!(current.deadline, parent.deadline);
assert_eq!(
current.trace_context.trace_id,
parent.trace_context.trace_id
);
assert_eq!(
current.trace_context.sampling_decision,
parent.trace_context.sampling_decision
);
assert!(!current.trace_context.span_id.is_none());
});
}
}
64 changes: 2 additions & 62 deletions tarpc/src/trace.rs
Original file line number Diff line number Diff line change
Expand Up @@ -16,14 +16,13 @@
//! This crate's design is based on [opencensus
//! tracing](https://opencensus.io/core-concepts/tracing/).

use opentelemetry::trace::TraceContextExt;
use crate::context::SpanExt;
use rand::{Rng, RngExt};
use std::{
convert::TryFrom,
fmt::{self, Formatter},
num::{NonZeroU64, NonZeroU128},
};
use tracing_opentelemetry::OpenTelemetrySpanExt;

/// A context for tracing the execution of processes, distributed or otherwise.
///
Expand Down Expand Up @@ -137,70 +136,11 @@ impl From<u64> for SpanId {
}
}

impl From<opentelemetry::trace::TraceId> for TraceId {
fn from(trace_id: opentelemetry::trace::TraceId) -> Self {
Self::from(u128::from_be_bytes(trace_id.to_bytes()))
}
}

impl From<TraceId> for opentelemetry::trace::TraceId {
fn from(trace_id: TraceId) -> Self {
Self::from_bytes(u128::from(trace_id).to_be_bytes())
}
}

impl From<opentelemetry::trace::SpanId> for SpanId {
fn from(span_id: opentelemetry::trace::SpanId) -> Self {
Self::from(u64::from_be_bytes(span_id.to_bytes()))
}
}

impl From<SpanId> for opentelemetry::trace::SpanId {
fn from(span_id: SpanId) -> Self {
Self::from_bytes(u64::from(span_id).to_be_bytes())
}
}

impl TryFrom<&tracing::Span> for Context {
type Error = NoActiveSpan;

fn try_from(span: &tracing::Span) -> Result<Self, NoActiveSpan> {
let context = span.context();
if context.has_active_span() {
Ok(Self::from(context.span()))
} else {
Err(NoActiveSpan)
}
}
}

impl From<opentelemetry::trace::SpanRef<'_>> for Context {
fn from(span: opentelemetry::trace::SpanRef<'_>) -> Self {
let otel_ctx = span.span_context();
Self {
trace_id: TraceId::from(otel_ctx.trace_id()),
span_id: SpanId::from(otel_ctx.span_id()),
sampling_decision: SamplingDecision::from(otel_ctx),
}
}
}

impl From<SamplingDecision> for opentelemetry::trace::TraceFlags {
fn from(decision: SamplingDecision) -> Self {
match decision {
SamplingDecision::Sampled => opentelemetry::trace::TraceFlags::SAMPLED,
SamplingDecision::Unsampled => opentelemetry::trace::TraceFlags::default(),
}
}
}

impl From<&opentelemetry::trace::SpanContext> for SamplingDecision {
fn from(context: &opentelemetry::trace::SpanContext) -> Self {
if context.is_sampled() {
SamplingDecision::Sampled
} else {
SamplingDecision::Unsampled
}
span.trace_context()
}
}

Expand Down
Loading
Loading