Skip to content

Span의 Start/End Time을 임의로 지정할 수 있는 기능 추가 #1

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

Open
libc-furiosa wants to merge 1 commit into
base: tracing-opentelemetry
Choose a base branch
from
Open

Span의 Start/End Time을 임의로 지정할 수 있는 기능 추가 #1

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
85 changes: 80 additions & 5 deletions tracing-opentelemetry/src/layer.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@ use std::borrow::Cow;
use std::fmt;
use std::marker;
use std::thread;
use std::time::{Instant, SystemTime};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use tracing_core::span::{self, Attributes, Id, Record};
use tracing_core::{field, Event, Subscriber};
#[cfg(feature = "tracing-log")]
Expand All @@ -23,6 +23,9 @@ const SPAN_KIND_FIELD: &str = "otel.kind";
const SPAN_STATUS_CODE_FIELD: &str = "otel.status_code";
const SPAN_STATUS_MESSAGE_FIELD: &str = "otel.status_message";

const SPAN_NPU_START_TIME_FIELD: &str = "npu.start";
const SPAN_NPU_END_TIME_FIELD: &str = "npu.end";

const FIELD_EXCEPTION_MESSAGE: &str = "exception.message";
const FIELD_EXCEPTION_STACKTRACE: &str = "exception.stacktrace";

Expand Down Expand Up @@ -315,6 +318,27 @@ impl<'a> field::Visit for SpanAttributeVisitor<'a> {
self.record(KeyValue::new(field.name(), value));
}

/// Set attributes on the underlying OpenTelemetry [`Span`] from `u128` values.
///
/// [`Span`]: opentelemetry::trace::Span
fn record_u128(&mut self, field: &field::Field, value: u128) {
match field.name() {
SPAN_NPU_START_TIME_FIELD => {
Copy link
Copy Markdown

@ileixe ileixe Aug 11, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

저도 구체적인 아이디어가 있는건 아닌데 혹시 이 부분이 조금 더 general하게 넣을수 있는 API 포인트가 있을까요? 예를 들어 지금은 npu.start 라는 특정 string이 field에 들어가는 형태인데 다른 제 3의 방법이 있을까 해서요 (아마 없을거 같긴한데.. 😇 업스트림에 넣는게 계속 조금 아쉬워서 질문드립니다.)

Copy link
Copy Markdown
Collaborator Author

@libc-furiosa libc-furiosa Aug 11, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

아예 시간 정보를 attribute로 받지 않고, span에 대한 function call로 builder의 time을 update하는 방법을 찾았습니다. SpanExt trait을 만들고 Span에 대해 구현을 아래와 같이 하면 되는 것 같습니다.

// impl OpenTelemetrySpanExt for tracing::Span
    fn set_start_time(&self, start: SystemTime) {
        self.with_subscriber(|(id, subscriber)| {
            if let Some(get_context) = subscriber.downcast_ref::<WithContext>() {
                get_context.with_context(subscriber, id, |otel_data, _tracer| {
                    otel_data.builder.start_time = Some(start);
                })
            }
        });
    }
// ...
let span = info_span!("something", ...);
span.set_start_time(start_time);

다만 유일하게 문제가 되는 부분이 아래의 on_close() 부분인데요.
span을 drop하면 사실상 항상 호출되는 on_close에서 end_time을 now로 덮어버려서 set_end_time이 의미가 없게 되어버립니다.
이 부분만 회피할 수 있다면 tracing-opentelemetry를 수정하지 않아도 될 것 같습니다. 다만 해결할 수 있는 뾰족한 방법이 보이지 않네요.

let secs = (value / 1_000_000_000) as u64;
let nanos = (value % 1_000_000_000) as u32;

self.span_builder.start_time = UNIX_EPOCH.checked_add(Duration::new(secs, nanos));
}
SPAN_NPU_END_TIME_FIELD => {
let secs = (value / 1_000_000_000) as u64;
let nanos = (value % 1_000_000_000) as u32;

self.span_builder.end_time = UNIX_EPOCH.checked_add(Duration::new(secs, nanos));
}
_ => self.record_debug(field, &value),
}
}

/// Set attributes on the underlying OpenTelemetry [`Span`] from `&str` values.
///
/// [`Span`]: opentelemetry::trace::Span
Expand Down Expand Up @@ -916,10 +940,12 @@ where
}
}

// Assign end time, build and start span, drop span to export
builder
.with_end_time(SystemTime::now())
.start_with_context(&self.tracer, &parent_cx);
if builder.end_time.is_none() {
// Assign end time
builder = builder.with_end_time(SystemTime::now());
Copy link
Copy Markdown

@ileixe ileixe Aug 11, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

realtime clock을 쓰는 부분은 monotonic clock이랑 conflict가 없는지 확인해봐야 될 거 같아요. 저희는 지금 monotonic clock을 쓰고 있거든요.

Copy link
Copy Markdown
Collaborator Author

@libc-furiosa libc-furiosa Aug 11, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

그렇지 않아도 이 부분을 npu-tools에서 리뷰를 요청드리려 했었습니다.
Opentelemetry Span에서는 SystemTime을 기준으로 하다보니 Instant를 그대로 사용할 수는 없었는데요.
instrument의 경우 일반적인 방식으로 on_new_span과 on_close 시점의 SystemTime::now()가 적용되고,
NPU profiling의 경우 profiler.sync() 시 SystemTime::now() (기존의 Instant::now()를 대체)가 불리고 이를 기준으로 상대 시간으로 계산하고 있습니다.

기존 방식과 결과 자체는 다르지 않아 보이는데 검토는 필요할 것 같습니다. 이 내용은 npu-tools 쪽에 올릴 PR에서 자세히 다루면 좋을 것 같습니다.

}
// Build and start span, drop span to export
builder.start_with_context(&self.tracer, &parent_cx);
}
}

Expand Down Expand Up @@ -1138,6 +1164,55 @@ mod tests {
assert_eq!(recorded_status_message, Some(message.into()))
}

#[test]
fn span_start_time() {
let tracer = TestTracer(Arc::new(Mutex::new(None)));
let subscriber = tracing_subscriber::registry().with(layer().with_tracer(tracer.clone()));

let now = SystemTime::now();
let duration = now.duration_since(std::time::UNIX_EPOCH).unwrap();

tracing::subscriber::with_default(subscriber, || {
tracing::debug_span!("request", npu.start = duration.as_nanos());
});

let recorded_start_time = tracer.with_data(|data| data.builder.start_time);
assert_eq!(recorded_start_time, Some(now));
}

#[test]
fn span_end_time_eager() {
let tracer = TestTracer(Arc::new(Mutex::new(None)));
let subscriber = tracing_subscriber::registry().with(layer().with_tracer(tracer.clone()));

let now = SystemTime::now();
let duration = now.duration_since(std::time::UNIX_EPOCH).unwrap();

tracing::subscriber::with_default(subscriber, || {
tracing::debug_span!("request", npu.end = duration.as_nanos());
});

let recorded_end_time = tracer.with_data(|data| data.builder.end_time);
assert_eq!(recorded_end_time, Some(now));
}

#[test]
fn span_end_time_lazy() {
let tracer = TestTracer(Arc::new(Mutex::new(None)));
let subscriber = tracing_subscriber::registry().with(layer().with_tracer(tracer.clone()));

let now = SystemTime::now();
let duration = now.duration_since(std::time::UNIX_EPOCH).unwrap();

tracing::subscriber::with_default(subscriber, || {
let span = tracing::debug_span!("request", npu.end = field::Empty);
span.record("npu.end", &duration.as_nanos());
});

let recorded_end_time = tracer.with_data(|data| data.builder.end_time);
assert_eq!(recorded_end_time, Some(now));
}

#[test]
fn trace_id_from_existing_context() {
let tracer = TestTracer(Arc::new(Mutex::new(None)));
Expand Down