diff --git a/Cargo.toml b/Cargo.toml index 3314008..5ed4d53 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -32,10 +32,14 @@ license = "Apache-2.0" homepage = "https://skywalking.apache.org/" repository = "https://github.com/apache/skywalking-rust" +[features] +mock = [] # For internal integration testing only, do not use. + [dependencies] async-stream = "0.3.3" base64 = "0.13.0" bytes = "1.1.0" +cfg-if = "1.0.0" futures-core = "0.3.21" futures-util = "0.3.21" prost = "0.10.4" @@ -52,6 +56,10 @@ tonic-build = "0.7.2" [dev-dependencies] tokio-stream = { version = "0.1.8", features = ["net"] } +[[test]] +name = "trace_context" +required-features = ["mock"] + [[example]] name = "simple_trace_report" path = "examples/simple_trace_report.rs" diff --git a/README.md b/README.md index 58cec12..7f2d55e 100644 --- a/README.md +++ b/README.md @@ -37,38 +37,38 @@ context after the span finished. use skywalking::context::tracer::Tracer; use skywalking::reporter::grpc::GrpcReporter; use std::error::Error; -use std::sync::Arc; use tokio::signal; -async fn handle_request(tracer: Arc>) { +async fn handle_request(tracer: Tracer) { let mut ctx = tracer.create_trace_context(); { - // Generate an Entry Span when a request - // is received. An Entry Span is generated only once per context. - let span = ctx.create_entry_span("op1").unwrap(); + // Generate an Entry Span when a request is received. + // An Entry Span is generated only once per context. + // Assign a variable name to guard the span not to be dropped immediately. + let _span = ctx.create_entry_span("op1"); // Something... { // Generates an Exit Span when executing an RPC. - let span2 = ctx.create_exit_span("op2", "remote_peer").unwrap(); + let _span2 = ctx.create_exit_span("op2", "remote_peer"); // Something... - ctx.finalize_span(span2); + // Auto close span2 when dropped. } - ctx.finalize_span(span); + // Auto close span when dropped. } - tracer.finalize_context(ctx); + // Auto report ctx when dropped. } #[tokio::main] async fn main() -> Result<(), Box> { let reporter = GrpcReporter::connect("http://0.0.0.0:11800").await?; - let tracer = Arc::new(Tracer::new("service", "instance", reporter)); + let tracer = Tracer::new("service", "instance", reporter); tokio::spawn(handle_request(tracer.clone())); diff --git a/e2e/src/main.rs b/e2e/src/main.rs index b6f7b19..ae2432b 100644 --- a/e2e/src/main.rs +++ b/e2e/src/main.rs @@ -21,38 +21,25 @@ use hyper::{Body, Client, Method, Request, Response, Server, StatusCode}; use skywalking::context::propagation::context::SKYWALKING_HTTP_CONTEXT_HEADER_KEY; use skywalking::context::propagation::decoder::decode_propagation; use skywalking::context::propagation::encoder::encode_propagation; -use skywalking::context::tracer::Tracer; +use skywalking::context::tracer::{self, Tracer}; use skywalking::reporter::grpc::GrpcReporter; use std::convert::Infallible; use std::error::Error; use std::future::pending; use std::net::SocketAddr; use structopt::StructOpt; -use tokio::sync::OnceCell; static NOT_FOUND_MSG: &str = "not found"; static SUCCESS_MSG: &str = "Success"; -static GLOBAL_TRACER: OnceCell> = OnceCell::const_new(); - -fn set_global_tracer(tracer: Tracer) { - if GLOBAL_TRACER.set(tracer).is_err() { - panic!("TRACER has setted") - } -} - -fn get_global_tracer() -> &'static Tracer { - GLOBAL_TRACER.get().expect("TRACER haven't setted") -} - async fn handle_ping( _req: Request, client: Client, ) -> Result, Infallible> { - let mut context = get_global_tracer().create_trace_context(); - let span = context.create_entry_span("/ping").unwrap(); + let mut context = tracer::create_trace_context(); + let _span = context.create_entry_span("/ping"); { - let span2 = context.create_exit_span("/pong", "consumer:8082").unwrap(); + let _span2 = context.create_exit_span("/pong", "consumer:8082"); let header = encode_propagation(&context, "/pong", "consumer:8082"); let req = Request::builder() .method(Method::GET) @@ -62,10 +49,7 @@ async fn handle_ping( .unwrap(); client.request(req).await.unwrap(); - context.finalize_span(span2); } - context.finalize_span(span); - get_global_tracer().finalize_context(context); Ok(Response::new(Body::from("hoge"))) } @@ -112,10 +96,8 @@ async fn handle_pong(_req: Request) -> Result, Infallible> .unwrap(), ) .unwrap(); - let mut context = get_global_tracer().create_trace_context_from_propagation(ctx); - let span = context.create_entry_span("/pong").unwrap(); - context.finalize_span(span); - get_global_tracer().finalize_context(context); + let mut context = tracer::create_trace_context_from_propagation(ctx); + let _span = context.create_entry_span("/pong"); Ok(Response::new(Body::from("hoge"))) } @@ -158,13 +140,13 @@ async fn main() -> Result<(), Box> { let reporter = GrpcReporter::connect("http://collector:19876").await?; let handle = if opt.mode == "consumer" { - set_global_tracer(Tracer::new("consumer", "node_0", reporter)); - let handle = get_global_tracer().reporting(pending()); + tracer::set_global_tracer(Tracer::new("consumer", "node_0", reporter)); + let handle = tracer::reporting(pending()); run_consumer_service([0, 0, 0, 0]).await; handle } else if opt.mode == "producer" { - set_global_tracer(Tracer::new("producer", "node_0", reporter)); - let handle = get_global_tracer().reporting(pending()); + tracer::set_global_tracer(Tracer::new("producer", "node_0", reporter)); + let handle = tracer::reporting(pending()); run_producer_service([0, 0, 0, 0]).await; handle } else { diff --git a/examples/simple_trace_report.rs b/examples/simple_trace_report.rs index 6d8d7fc..bad1fff 100644 --- a/examples/simple_trace_report.rs +++ b/examples/simple_trace_report.rs @@ -18,38 +18,38 @@ use skywalking::context::tracer::Tracer; use skywalking::reporter::grpc::GrpcReporter; use std::error::Error; -use std::sync::Arc; use tokio::signal; -async fn handle_request(tracer: Arc>) { +async fn handle_request(tracer: Tracer) { let mut ctx = tracer.create_trace_context(); { - // Generate an Entry Span when a request - // is received. An Entry Span is generated only once per context. - let span = ctx.create_entry_span("op1").unwrap(); + // Generate an Entry Span when a request is received. + // An Entry Span is generated only once per context. + // You should assign a variable name to guard the span not be dropped immediately. + let _span = ctx.create_entry_span("op1"); // Something... { // Generates an Exit Span when executing an RPC. - let span2 = ctx.create_exit_span("op2", "remote_peer").unwrap(); + let _span2 = ctx.create_exit_span("op2", "remote_peer"); // Something... - ctx.finalize_span(span2); + // Auto close span2 when dropped. } - ctx.finalize_span(span); + // Auto close span when dropped. } - tracer.finalize_context(ctx); + // Auto report ctx when dropped. } #[tokio::main] async fn main() -> Result<(), Box> { let reporter = GrpcReporter::connect("http://0.0.0.0:11800").await?; - let tracer = Arc::new(Tracer::new("service", "instance", reporter)); + let tracer = Tracer::new("service", "instance", reporter); tokio::spawn(handle_request(tracer.clone())); diff --git a/src/common/mod.rs b/src/common/mod.rs index cd8c341..7bb6d17 100644 --- a/src/common/mod.rs +++ b/src/common/mod.rs @@ -15,4 +15,3 @@ // pub mod random_generator; -pub mod time; diff --git a/src/context/mod.rs b/src/context/mod.rs index 99ddb08..28d7ace 100644 --- a/src/context/mod.rs +++ b/src/context/mod.rs @@ -15,6 +15,7 @@ // pub mod propagation; -pub mod system_time; +pub mod span; +pub(crate) mod system_time; pub mod trace_context; pub mod tracer; diff --git a/src/context/propagation/encoder.rs b/src/context/propagation/encoder.rs index c018e77..fddb741 100644 --- a/src/context/propagation/encoder.rs +++ b/src/context/propagation/encoder.rs @@ -20,15 +20,14 @@ use base64::encode; /// Encode TracingContext to carry current trace info to the destination of RPC call. /// In general, the output of this function will be packed in `sw8` header in HTTP call. pub fn encode_propagation(context: &TracingContext, endpoint: &str, address: &str) -> String { - let mut res = String::new(); - - res += "1-"; - res += format!("{}-", encode(context.trace_id.to_string())).as_str(); - res += format!("{}-", encode(context.trace_segment_id.to_string())).as_str(); - res += format!("{}-", context.peek_active_span_id().unwrap_or(0)).as_str(); - res += format!("{}-", encode(context.service.as_str())).as_str(); - res += format!("{}-", encode(context.service_instance.as_str())).as_str(); - res += format!("{}-", encode(endpoint)).as_str(); - res += &encode(address); - res + format!( + "1-{}-{}-{}-{}-{}-{}-{}", + encode(context.trace_id()), + encode(context.trace_segment_id()), + context.peek_active_span_id().unwrap_or(0), + encode(context.service()), + encode(context.service_instance()), + encode(endpoint), + encode(address) + ) } diff --git a/src/context/span.rs b/src/context/span.rs new file mode 100644 index 0000000..6d0bf21 --- /dev/null +++ b/src/context/span.rs @@ -0,0 +1,190 @@ +// Licensed to the Apache Software Foundation (ASF) under one or more +// contributor license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright ownership. +// The ASF licenses this file to You under the Apache License, Version 2.0 +// (the "License"); you may not use this file except in compliance with +// the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + +use crate::skywalking_proto::v3::{KeyStringValuePair, Log, SpanLayer, SpanObject, SpanType}; +use std::fmt::Formatter; + +use super::{ + system_time::{fetch_time, TimePeriod}, + trace_context::{TracingContext, WeakTracingContext}, +}; + +/// Span is a concept that represents trace information for a single RPC. +/// The Rust SDK supports Entry Span to represent inbound to a service +/// and Exit Span to represent outbound from a service. +/// +/// # Example +/// +/// ``` +/// use skywalking::context::tracer::Tracer; +/// +/// async fn handle_request(tracer: Tracer) { +/// let mut ctx = tracer.create_trace_context(); +/// +/// { +/// // Generate an Entry Span when a request is received. +/// // An Entry Span is generated only once per context. +/// // You should assign a variable name to guard the span not be dropped immediately. +/// let _span = ctx.create_entry_span("op1"); +/// +/// // Something... +/// +/// { +/// // Generates an Exit Span when executing an RPC. +/// let _span2 = ctx.create_exit_span("op2", "remote_peer"); +/// +/// // Something... +/// +/// // Auto close span2 when dropped. +/// } +/// +/// // Auto close span when dropped. +/// } +/// +/// // Auto report ctx when dropped. +/// } +/// ``` +#[derive(Clone)] +pub struct Span { + index: usize, + context: WeakTracingContext, +} + +impl std::fmt::Debug for Span { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + let mut d = f.debug_struct("Span"); + match self.context.upgrade() { + Some(context) => { + let op = context.try_with_active_span_stack(|stack| { + d.field("data", &stack.get(self.index)); + }); + if op.is_none() { + d.field("data", &format_args!("")); + } + } + None => { + d.field("context", &format_args!("")); + } + } + d.finish() + } +} + +const SKYWALKING_RUST_COMPONENT_ID: i32 = 11000; + +impl Span { + pub(crate) fn new(index: usize, context: WeakTracingContext) -> Self { + Self { index, context } + } + + #[allow(clippy::too_many_arguments)] + pub(crate) fn new_obj( + span_id: i32, + parent_span_id: i32, + operation_name: String, + remote_peer: String, + span_type: SpanType, + span_layer: SpanLayer, + skip_analysis: bool, + ) -> SpanObject { + SpanObject { + span_id, + parent_span_id, + start_time: fetch_time(TimePeriod::Start), + operation_name, + peer: remote_peer, + span_type: span_type as i32, + span_layer: span_layer as i32, + component_id: SKYWALKING_RUST_COMPONENT_ID, + skip_analysis, + ..Default::default() + } + } + + fn upgrade_context(&self) -> TracingContext { + self.context.upgrade().expect("Context has dropped") + } + + pub fn with_span_object(&self, f: impl FnOnce(&SpanObject) -> T) -> T { + self.upgrade_context() + .with_active_span_stack(|stack| f(&stack[self.index])) + } + + pub fn with_span_object_mut(&mut self, f: impl FnOnce(&mut SpanObject) -> T) -> T { + self.upgrade_context() + .with_active_span_stack_mut(|stack| f(&mut stack[self.index])) + } + + pub fn span_id(&self) -> i32 { + self.with_span_object(|span| span.span_id) + } + + /// Add logs to the span. + pub fn add_log(&mut self, message: I) + where + K: ToString, + V: ToString, + I: IntoIterator, + { + let log = Log { + time: fetch_time(TimePeriod::Log), + data: message + .into_iter() + .map(|v| { + let (key, value) = v; + KeyStringValuePair { + key: key.to_string(), + value: value.to_string(), + } + }) + .collect(), + }; + self.with_span_object_mut(|span| span.logs.push(log)); + } + + /// Add tag to the span. + pub fn add_tag(&mut self, key: impl ToString, value: impl ToString) { + self.with_span_object_mut(|span| { + span.tags.push(KeyStringValuePair { + key: key.to_string(), + value: value.to_string(), + }) + }) + } +} + +impl Drop for Span { + /// Set the end time as current time, pop from context active span stack, + /// and push to context spans. + /// + /// # Panics + /// + /// Panic if context is dropped or this span isn't the active span. + fn drop(&mut self) { + if self.upgrade_context().finalize_span(self.index).is_err() { + panic!("Dropped span isn't the active span"); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + trait AssertSend: Send {} + + impl AssertSend for Span {} +} diff --git a/src/context/system_time.rs b/src/context/system_time.rs index c5e29cc..ee97b4e 100644 --- a/src/context/system_time.rs +++ b/src/context/system_time.rs @@ -14,17 +14,30 @@ // limitations under the License. // -use crate::common::time::TimeFetcher; -use std::time::{SystemTime, UNIX_EPOCH}; +use cfg_if::cfg_if; -#[derive(Default)] -pub struct UnixTimeStampFetcher {} +pub(crate) enum TimePeriod { + Start, + Log, + End, +} -impl TimeFetcher for UnixTimeStampFetcher { - fn get(&self) -> i64 { - SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap() - .as_millis() as i64 +cfg_if! { + if #[cfg(feature = "mock")] { + pub(crate) fn fetch_time(period: TimePeriod) -> i64 { + match period { + TimePeriod::Start => 1, + TimePeriod::Log => 10, + TimePeriod::End => 100, + } + } + } else { + pub(crate) fn fetch_time(_period: TimePeriod) -> i64 { + use std::time::{SystemTime, UNIX_EPOCH}; + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|dur| dur.as_millis() as i64) + .unwrap_or_default() + } } } diff --git a/src/context/trace_context.rs b/src/context/trace_context.rs index 0934519..014c75b 100644 --- a/src/context/trace_context.rs +++ b/src/context/trace_context.rs @@ -14,387 +14,384 @@ // limitations under the License. // -use crate::common::random_generator::RandomGenerator; -use crate::common::time::TimeFetcher; -use crate::context::propagation::context::PropagationContext; -use crate::skywalking_proto::v3::{ - KeyStringValuePair, Log, RefType, SegmentObject, SegmentReference, SpanLayer, SpanObject, - SpanType, +use super::{ + span::Span, + system_time::{fetch_time, TimePeriod}, + tracer::{Tracer, WeakTracer}, }; -use std::collections::LinkedList; -use std::fmt::Formatter; -use std::sync::Arc; - -use super::system_time::UnixTimeStampFetcher; - -/// Span is a concept that represents trace information for a single RPC. -/// The Rust SDK supports Entry Span to represent inbound to a service -/// and Exit Span to represent outbound from a service. -/// -/// # Example -/// -/// ``` -/// use skywalking::context::trace_context::TracingContext; -/// -/// async fn handle_request() { -/// let mut ctx = TracingContext::default("svc", "ins"); -/// { -/// // Generate an Entry Span when a request -/// // is received. An Entry Span is generated only once per context. -/// let span = ctx.create_entry_span("operation1").unwrap(); -/// -/// // Something... -/// -/// { -/// // Generates an Exit Span when executing an RPC. -/// let span2 = ctx.create_exit_span("operation2", "remote_peer").unwrap(); -/// -/// // Something... -/// -/// ctx.finalize_span(span2); -/// } -/// -/// ctx.finalize_span(span); -/// } -/// } -/// ``` -pub struct Span { - span_internal: SpanObject, - time_fetcher: Arc, +use crate::{ + common::random_generator::RandomGenerator, context::propagation::context::PropagationContext, +}; +use crate::{ + error::LOCK_MSG, + skywalking_proto::v3::{ + RefType, SegmentObject, SegmentReference, SpanLayer, SpanObject, SpanType, + }, +}; +use std::{ + fmt::Formatter, + mem::take, + sync::{ + atomic::{AtomicI32, Ordering}, + Arc, Mutex, Weak, + }, +}; + +struct Inner { + trace_id: String, + trace_segment_id: String, + service: String, + service_instance: String, + next_span_id: AtomicI32, + spans: Mutex>, + active_span_stack: Mutex>, + segment_link: Option, + primary_endpoint_name: Mutex, } -impl std::fmt::Debug for Span { +#[derive(Clone)] +#[must_use = "You should call `create_entry_span` after `TracingContext` created."] +pub struct TracingContext { + inner: Arc, + tracer: WeakTracer, +} + +impl std::fmt::Debug for TracingContext { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { - f.debug_struct("Span") - .field("span_internal", &self.span_internal) + f.debug_struct("TracingContext") + .field("trace_id", &self.inner.trace_id) + .field("trace_segment_id", &self.inner.trace_segment_id) + .field("service", &self.inner.service) + .field("service_instance", &self.inner.service_instance) + .field("next_span_id", &self.inner.next_span_id) + .field("spans", &self.inner.spans) .finish() } } -const SKYWALKING_RUST_COMPONENT_ID: i32 = 11000; - -impl Span { - #[allow(clippy::too_many_arguments)] - pub fn new( - span_id: i32, - parent_span_id: i32, - operation_name: String, - remote_peer: String, - span_type: SpanType, - span_layer: SpanLayer, - skip_analysis: bool, - time_fetcher: Arc, +impl TracingContext { + /// Generate a new trace context. Typically called when no context has + /// been propagated and a new trace is to be started. + pub(crate) fn new( + service_name: impl ToString, + instance_name: impl ToString, + tracer: WeakTracer, ) -> Self { - let span_internal = SpanObject { - span_id, - parent_span_id, - start_time: time_fetcher.get(), - end_time: 0, // not set - refs: Vec::::new(), - operation_name, - peer: remote_peer, - span_type: span_type as i32, - span_layer: span_layer as i32, - component_id: SKYWALKING_RUST_COMPONENT_ID, - is_error: false, - tags: Vec::::new(), - logs: Vec::::new(), - skip_analysis, - }; - - Span { - span_internal, - time_fetcher, + TracingContext { + inner: Arc::new(Inner { + trace_id: RandomGenerator::generate(), + trace_segment_id: RandomGenerator::generate(), + service: service_name.to_string(), + service_instance: instance_name.to_string(), + next_span_id: Default::default(), + spans: Default::default(), + segment_link: None, + active_span_stack: Default::default(), + primary_endpoint_name: Default::default(), + }), + tracer, } } - /// Close span. It only registers end time to the span. - pub fn close(&mut self) { - self.span_internal.end_time = self.time_fetcher.get(); + /// Generate a new trace context using the propagated context. + /// They should be propagated on `sw8` header in HTTP request with encoded form. + /// You can retrieve decoded context with `skywalking::context::propagation::encoder::encode_propagation` + pub(crate) fn from_propagation_context( + service_name: impl ToString, + instance_name: impl ToString, + context: PropagationContext, + tracer: WeakTracer, + ) -> Self { + TracingContext { + inner: Arc::new(Inner { + trace_id: context.parent_trace_id.clone(), + trace_segment_id: RandomGenerator::generate(), + service: service_name.to_string(), + service_instance: instance_name.to_string(), + next_span_id: Default::default(), + spans: Default::default(), + segment_link: Some(context), + active_span_stack: Default::default(), + primary_endpoint_name: Default::default(), + }), + tracer, + } } - pub fn span_object(&self) -> &SpanObject { - &self.span_internal + #[inline] + pub fn trace_id(&self) -> &str { + &self.inner.trace_id } - pub fn span_object_mut(&mut self) -> &mut SpanObject { - &mut self.span_internal + #[inline] + pub fn trace_segment_id(&self) -> &str { + &self.inner.trace_segment_id } - /// Add logs to the span. - pub fn add_log(&mut self, message: Vec<(&str, &str)>) { - let log = Log { - time: self.time_fetcher.get(), - data: message - .into_iter() - .map(|v| { - let (key, value) = v; - KeyStringValuePair { - key: key.to_string(), - value: value.to_string(), - } - }) - .collect(), - }; - self.span_internal.logs.push(log); + #[inline] + pub fn service(&self) -> &str { + &self.inner.service } - /// Add tag to the span. - pub fn add_tag(&mut self, tag: (&str, &str)) { - let (key, value) = tag; - self.span_internal.tags.push(KeyStringValuePair { - key: key.to_string(), - value: value.to_string(), - }); + #[inline] + pub fn service_instance(&self) -> &str { + &self.inner.service_instance } - fn add_segment_reference(&mut self, segment_reference: SegmentReference) { - self.span_internal.refs.push(segment_reference); + fn next_span_id(&self) -> i32 { + self.inner.next_span_id.load(Ordering::Relaxed) } -} -pub struct TracingContext { - pub trace_id: String, - pub trace_segment_id: String, - pub service: String, - pub service_instance: String, - pub next_span_id: i32, - pub spans: Vec>, - time_fetcher: Arc, - segment_link: Option, - active_span_id_stack: LinkedList, -} + #[inline] + fn inc_next_span_id(&self) -> i32 { + self.inner.next_span_id.fetch_add(1, Ordering::Relaxed) + } -impl std::fmt::Debug for TracingContext { - fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { - f.debug_struct("TracingContext") - .field("trace_id", &self.trace_id) - .field("trace_segment_id", &self.trace_segment_id) - .field("service", &self.service) - .field("service_instance", &self.service_instance) - .field("next_span_id", &self.next_span_id) - .field("spans", &self.spans) - .finish() + #[cfg(feature = "mock")] + pub fn with_spans(&self, f: impl FnOnce(&Vec) -> T) -> T { + f(&*self.inner.spans.try_lock().expect(LOCK_MSG)) } -} -impl TracingContext { - /// Generate a new trace context. Typically called when no context has - /// been propagated and a new trace is to be started. - pub fn default(service_name: &str, instance_name: &str) -> Self { - let unix_time_fetcher = UnixTimeStampFetcher {}; - TracingContext::default_internal(Arc::new(unix_time_fetcher), service_name, instance_name) + fn with_spans_mut(&mut self, f: impl FnOnce(&mut Vec) -> T) -> T { + f(&mut *self.inner.spans.try_lock().expect(LOCK_MSG)) } - pub fn default_internal( - time_fetcher: Arc, - service_name: &str, - instance_name: &str, - ) -> Self { - TracingContext { - trace_id: RandomGenerator::generate(), - trace_segment_id: RandomGenerator::generate(), - service: String::from(service_name), - service_instance: String::from(instance_name), - next_span_id: 0, - time_fetcher, - spans: Vec::new(), - segment_link: None, - active_span_id_stack: LinkedList::new(), - } + pub(crate) fn with_active_span_stack(&self, f: impl FnOnce(&Vec) -> T) -> T { + f(&*self.inner.active_span_stack.try_lock().expect(LOCK_MSG)) } - /// Generate a new trace context using the propagated context. - /// They should be propagated on `sw8` header in HTTP request with encoded form. - /// You can retrieve decoded context with `skywalking::context::propagation::encoder::encode_propagation` - pub fn from_propagation_context( - service_name: &str, - instance_name: &str, - context: PropagationContext, - ) -> Self { - let unix_time_fetcher = UnixTimeStampFetcher {}; - TracingContext::from_propagation_context_internal( - Arc::new(unix_time_fetcher), - service_name, - instance_name, - context, - ) + pub(crate) fn with_active_span_stack_mut( + &mut self, + f: impl FnOnce(&mut Vec) -> T, + ) -> T { + f(&mut *self.inner.active_span_stack.try_lock().expect(LOCK_MSG)) } - pub fn from_propagation_context_internal( - time_fetcher: Arc, - service_name: &str, - instance_name: &str, - context: PropagationContext, - ) -> Self { - TracingContext { - trace_id: context.parent_trace_id.clone(), - trace_segment_id: RandomGenerator::generate(), - service: service_name.to_string(), - service_instance: instance_name.to_string(), - next_span_id: 0, - time_fetcher, - spans: Vec::new(), - segment_link: Some(context), - active_span_id_stack: LinkedList::new(), - } + pub(crate) fn try_with_active_span_stack( + &self, + f: impl FnOnce(&Vec) -> T, + ) -> Option { + self.inner + .active_span_stack + .try_lock() + .ok() + .map(|stack| f(&*stack)) } - /// A wrapper of create entry span, which close generated span automatically. - /// Note that, we may use async operation in closure. But that is not unstable feature in 2021/12. - /// - /// So we should create and close spans manually in general. - pub fn entry( - &mut self, - operation_name: &str, - mut process_fn: F, - ) -> crate::Result<()> { - match self.create_entry_span(operation_name) { - Ok(mut span) => { - process_fn(span.as_ref()); - span.close(); - Ok(()) - } - Err(message) => Err(message), - } + pub(crate) fn with_active_span(&self, f: impl FnOnce(&SpanObject) -> T) -> Option { + self.with_active_span_stack(|stack| stack.last().map(f)) + } + + // TODO Using for capture and continued. + #[allow(dead_code)] + fn with_primary_endpoint_name(&self, f: impl FnOnce(&String) -> T) -> T { + f(&*self.inner.primary_endpoint_name.try_lock().expect(LOCK_MSG)) + } + + fn with_primary_endpoint_name_mut(&mut self, f: impl FnOnce(&mut String) -> T) -> T { + f(&mut *self.inner.primary_endpoint_name.try_lock().expect(LOCK_MSG)) } /// Create a new entry span, which is an initiator of collection of spans. /// This should be called by invocation of the function which is triggered by /// external service. - pub fn create_entry_span(&mut self, operation_name: &str) -> crate::Result> { - if self.next_span_id >= 1 { - return Err(crate::Error::CreateSpan("entry span have already exist.")); - } - - let parent_span_id = self.peek_active_span_id().unwrap_or(-1); - - let mut span = Box::new(Span::new( - self.next_span_id, - parent_span_id, + pub fn create_entry_span(&mut self, operation_name: &str) -> Span { + let mut span = Span::new_obj( + self.inc_next_span_id(), + self.peek_active_span_id().unwrap_or(-1), operation_name.to_string(), String::default(), SpanType::Entry, SpanLayer::Http, false, - self.time_fetcher.clone(), - )); + ); - if self.segment_link.is_some() { - span.add_segment_reference(SegmentReference { + if let Some(segment_link) = &self.inner.segment_link { + span.refs.push(SegmentReference { ref_type: RefType::CrossProcess as i32, - trace_id: self.trace_id.clone(), - parent_trace_segment_id: self - .segment_link - .as_ref() - .unwrap() - .parent_trace_segment_id - .clone(), - parent_span_id: self.segment_link.as_ref().unwrap().parent_span_id, - parent_service: self.segment_link.as_ref().unwrap().parent_service.clone(), - parent_service_instance: self - .segment_link - .as_ref() - .unwrap() - .parent_service_instance - .clone(), - parent_endpoint: self - .segment_link - .as_ref() - .unwrap() - .destination_endpoint - .clone(), - network_address_used_at_peer: self - .segment_link - .as_ref() - .unwrap() - .destination_address - .clone(), + trace_id: self.inner.trace_id.clone(), + parent_trace_segment_id: segment_link.parent_trace_segment_id.clone(), + parent_span_id: segment_link.parent_span_id, + parent_service: segment_link.parent_service.clone(), + parent_service_instance: segment_link.parent_service_instance.clone(), + parent_endpoint: segment_link.destination_endpoint.clone(), + network_address_used_at_peer: segment_link.destination_address.clone(), }); } - self.next_span_id += 1; - self.active_span_id_stack - .push_back(span.span_internal.span_id); - Ok(span) - } - /// A wrapper of create exit span, which close generated span automatically. - /// Note that, we may use async operation in closure. But that is not unstable feature in 2021/12. - /// - /// So we should create and close spans manually in general. - pub fn exit( - &mut self, - operation_name: &str, - remote_peer: &str, - mut process_fn: F, - ) -> crate::Result<()> { - match self.create_exit_span(operation_name, remote_peer) { - Ok(mut span) => { - process_fn(span.as_ref()); - span.close(); - Ok(()) - } - Err(message) => Err(message), - } + let index = self.push_active_span(span); + Span::new(index, self.downgrade()) } /// Create a new exit span, which will be created when tracing context will generate /// new span for function invocation. /// Currently, this SDK supports RPC call. So we must set `remote_peer`. - pub fn create_exit_span( - &mut self, - operation_name: &str, - remote_peer: &str, - ) -> crate::Result> { - if self.next_span_id == 0 { - return Err(crate::Error::CreateSpan("entry span must be existed.")); + /// + /// # Panics + /// + /// Panic if entry span not existed. + pub fn create_exit_span(&mut self, operation_name: &str, remote_peer: &str) -> Span { + if self.next_span_id() == 0 { + panic!("entry span must be existed."); } - let parent_span_id = self.peek_active_span_id().unwrap_or(-1); - - let span = Box::new(Span::new( - self.next_span_id, - parent_span_id, + let span = Span::new_obj( + self.inc_next_span_id(), + self.peek_active_span_id().unwrap_or(-1), operation_name.to_string(), remote_peer.to_string(), SpanType::Exit, SpanLayer::Http, false, - self.time_fetcher.clone(), - )); - self.next_span_id += 1; - self.active_span_id_stack - .push_back(span.span_internal.span_id); - Ok(span) + ); + + let index = self.push_active_span(span); + Span::new(index, self.downgrade()) + } + + /// Create a new local span. + /// + /// # Panics + /// + /// Panic if entry span not existed. + pub fn create_local_span(&mut self, operation_name: &str) -> Span { + if self.next_span_id() == 0 { + panic!("entry span must be existed."); + } + + let span = Span::new_obj( + self.inc_next_span_id(), + self.peek_active_span_id().unwrap_or(-1), + operation_name.to_string(), + Default::default(), + SpanType::Local, + SpanLayer::Unknown, + false, + ); + + let index = self.push_active_span(span); + Span::new(index, self.downgrade()) } /// Close span. We can't use closed span after finalize called. - pub fn finalize_span(&mut self, mut span: Box) { - span.close(); - self.spans.push(span); - self.active_span_id_stack.pop_back(); + pub(crate) fn finalize_span(&mut self, index: usize) -> Result<(), ()> { + let span = self.pop_active_span(index); + if let Some(mut span) = span { + span.end_time = fetch_time(TimePeriod::End); + self.with_spans_mut(|spans| spans.push(span)); + Ok(()) + } else { + Err(()) + } } /// It converts tracing context into segment object. /// This conversion should be done before sending segments into OAP. - pub fn convert_segment_object(&self) -> SegmentObject { - let mut objects = Vec::::new(); - - for span in self.spans.iter() { - objects.push(span.span_internal.clone()); - } + /// + /// Notice: The spans will taked, so this method shouldn't be called twice. + pub(crate) fn convert_segment_object(&mut self) -> SegmentObject { + let trace_id = self.trace_id().to_owned(); + let trace_segment_id = self.trace_segment_id().to_owned(); + let service = self.service().to_owned(); + let service_instance = self.service_instance().to_owned(); + let spans = self.with_spans_mut(|spans| take(spans)); SegmentObject { - trace_id: self.trace_id.to_string(), - trace_segment_id: self.trace_segment_id.to_string(), - spans: objects, - service: self.service.clone(), - service_instance: self.service_instance.clone(), + trace_id, + trace_segment_id, + spans, + service, + service_instance, is_size_limited: false, } } pub(crate) fn peek_active_span_id(&self) -> Option { - self.active_span_id_stack.back().copied() + self.with_active_span(|span| span.span_id) + } + + fn push_active_span(&mut self, span: SpanObject) -> usize { + self.with_primary_endpoint_name_mut(|endpoint| *endpoint = span.operation_name.clone()); + self.with_active_span_stack_mut(|stack| { + stack.push(span); + stack.len() - 1 + }) + } + + fn pop_active_span(&mut self, index: usize) -> Option { + self.with_active_span_stack_mut(|stack| { + if stack.len() > index + 1 { + None + } else { + stack.pop() + } + }) + } + + fn downgrade(&self) -> WeakTracingContext { + WeakTracingContext { + inner: Arc::downgrade(&self.inner), + tracer: self.tracer.clone(), + } + } + + fn upgrade_tracer(&self) -> Tracer { + self.tracer.upgrade().expect("Tracer has dropped") + } +} + +impl Drop for TracingContext { + /// Convert to segment object, and send to tracer for reporting. + /// + /// # Panics + /// + /// Panic if tracer is dropped. + fn drop(&mut self) { + if Arc::strong_count(&self.inner) <= 1 { + self.upgrade_tracer().finalize_context(self) + } + } +} + +#[derive(Clone)] +pub(crate) struct WeakTracingContext { + inner: Weak, + tracer: WeakTracer, +} + +impl WeakTracingContext { + pub(crate) fn upgrade(&self) -> Option { + self.inner.upgrade().map(|inner| TracingContext { + inner, + tracer: self.tracer.clone(), + }) + } +} + +pub struct ContextSnapshot { + trace_id: String, + trace_segment_id: String, + span_id: i32, + // TODO Using for capture and continued. + #[allow(dead_code)] + parent_endpoint: String, +} + +impl ContextSnapshot { + pub fn is_from_current(&self, context: &TracingContext) -> bool { + !self.trace_segment_id.is_empty() && self.trace_segment_id == context.trace_segment_id() } + + pub fn is_valid(&self) -> bool { + !self.trace_segment_id.is_empty() && self.span_id > -1 && !self.trace_id.is_empty() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + trait AssertSend: Send {} + + impl AssertSend for TracingContext {} } diff --git a/src/context/tracer.rs b/src/context/tracer.rs index 7ca4bd1..0a87905 100644 --- a/src/context/tracer.rs +++ b/src/context/tracer.rs @@ -16,45 +16,103 @@ use super::propagation::context::PropagationContext; use crate::{ - context::trace_context::TracingContext, reporter::Reporter, skywalking_proto::v3::SegmentObject, + context::trace_context::TracingContext, reporter::DynReporter, reporter::Reporter, + skywalking_proto::v3::SegmentObject, }; -use futures_util::stream; use std::future::Future; +use std::sync::Weak; use std::{collections::LinkedList, sync::Arc}; +use tokio::sync::OnceCell; use tokio::{ sync::{ - mpsc::{self, UnboundedReceiver}, + mpsc::{self}, Mutex, }, task::JoinHandle, }; -/// Skywalking tracer. -pub struct Tracer { +static GLOBAL_TRACER: OnceCell = OnceCell::const_new(); + +/// Set the global tracer. +pub fn set_global_tracer(tracer: Tracer) { + if GLOBAL_TRACER.set(tracer).is_err() { + panic!("global tracer has setted") + } +} + +/// Get the global tracer. +pub fn global_tracer() -> &'static Tracer { + GLOBAL_TRACER.get().expect("global tracer haven't setted") +} + +/// Create trace conetxt by global tracer. +pub fn create_trace_context() -> TracingContext { + global_tracer().create_trace_context() +} + +/// Create trace conetxt from propagation by global tracer. +pub fn create_trace_context_from_propagation(context: PropagationContext) -> TracingContext { + global_tracer().create_trace_context_from_propagation(context) +} + +/// Start to reporting by global tracer, quit when shutdown_signal received. +/// +/// Accept a `shutdown_signal` argument as a graceful shutdown signal. +pub fn reporting( + shutdown_signal: impl Future + Send + Sync + 'static, +) -> JoinHandle<()> { + global_tracer().reporting(shutdown_signal) +} + +struct Inner { service_name: String, instance_name: String, - reporter: Arc>, segment_sender: mpsc::UnboundedSender, - segment_receiver: Arc>>, + segment_receiver: Mutex>, + reporter: Box>, } -impl Tracer { +/// Skywalking tracer. +#[derive(Clone)] +pub struct Tracer { + inner: Arc, +} + +impl Tracer { /// New with service info and reporter. - pub fn new(service_name: impl ToString, instance_name: impl ToString, reporter: R) -> Self { + pub fn new( + service_name: impl ToString, + instance_name: impl ToString, + reporter: impl Reporter + Send + Sync + 'static, + ) -> Self { let (segment_sender, segment_receiver) = mpsc::unbounded_channel(); Self { - service_name: service_name.to_string(), - instance_name: instance_name.to_string(), - reporter: Arc::new(Mutex::new(reporter)), - segment_sender, - segment_receiver: Arc::new(Mutex::new(segment_receiver)), + inner: Arc::new(Inner { + service_name: service_name.to_string(), + instance_name: instance_name.to_string(), + segment_sender, + segment_receiver: Mutex::new(segment_receiver), + reporter: Box::new(Mutex::new(reporter)), + }), } } + pub fn service_name(&self) -> &str { + &self.inner.service_name + } + + pub fn instance_name(&self) -> &str { + &self.inner.instance_name + } + /// Create trace conetxt. pub fn create_trace_context(&self) -> TracingContext { - TracingContext::default(&self.service_name, &self.instance_name) + TracingContext::new( + &self.inner.service_name, + &self.inner.instance_name, + self.downgrade(), + ) } /// Create trace conetxt from propagation. @@ -62,14 +120,19 @@ impl Tracer { &self, context: PropagationContext, ) -> TracingContext { - TracingContext::from_propagation_context(&self.service_name, &self.instance_name, context) + TracingContext::from_propagation_context( + &self.inner.service_name, + &self.inner.instance_name, + context, + self.downgrade(), + ) } /// Finalize the trace context. - pub fn finalize_context(&self, context: TracingContext) { + pub(crate) fn finalize_context(&self, context: &mut TracingContext) { let segment_object = context.convert_segment_object(); - if self.segment_sender.send(segment_object).is_err() { - tracing::warn!("segment object channel has closed"); + if self.inner.segment_sender.send(segment_object).is_err() { + tracing::error!("segment object channel has closed"); } } @@ -80,25 +143,15 @@ impl Tracer { &self, shutdown_signal: impl Future + Send + Sync + 'static, ) -> JoinHandle<()> { - let reporter = self.reporter.clone(); - let segment_receiver = self.segment_receiver.clone(); - tokio::spawn(Self::do_reporting( - reporter, - segment_receiver, - shutdown_signal, - )) + tokio::spawn(Self::do_reporting(self.clone(), shutdown_signal)) } - async fn do_reporting( - reporter: Arc>, - segment_receiver: Arc>>, - shutdown_signal: impl Future + Send + Sync + 'static, - ) { + async fn do_reporting(self, shutdown_signal: impl Future + Send + Sync + 'static) { let (shutdown_tx, mut shutdown_rx) = mpsc::unbounded_channel(); let handle = tokio::spawn(async move { loop { - let mut segment_receiver = segment_receiver.lock().await; + let mut segment_receiver = self.inner.segment_receiver.lock().await; let mut segments = LinkedList::new(); tokio::select! { @@ -108,8 +161,7 @@ impl Tracer { if let Some(segment) = segment { // TODO Implement batch collect in future. segments.push_back(segment); - let mut reporter = reporter.lock().await; - Self::report_segment_object(&mut reporter, segments).await; + Self::report_segment_object(&self.inner.reporter, segments).await; } else { break; } @@ -119,13 +171,12 @@ impl Tracer { } // Flush. - let mut segment_receiver = segment_receiver.lock().await; + let mut segment_receiver = self.inner.segment_receiver.lock().await; let mut segments = LinkedList::new(); while let Ok(segment) = segment_receiver.try_recv() { segments.push_back(segment); } - let mut reporter = reporter.lock().await; - Self::report_segment_object(&mut reporter, segments).await; + Self::report_segment_object(&self.inner.reporter, segments).await; }); shutdown_signal.await; @@ -138,10 +189,38 @@ impl Tracer { } } - async fn report_segment_object(reporter: &mut R, segments: LinkedList) { - let stream = stream::iter(segments); - if let Err(e) = reporter.collect(stream).await { + async fn report_segment_object( + reporter: &Mutex, + segments: LinkedList, + ) { + if let Err(e) = reporter.lock().await.collect(segments).await { tracing::error!("Collect failed: {:?}", e); } } + + fn downgrade(&self) -> WeakTracer { + WeakTracer { + inner: Arc::downgrade(&self.inner), + } + } +} + +#[derive(Clone)] +pub(crate) struct WeakTracer { + inner: Weak, +} + +impl WeakTracer { + pub(crate) fn upgrade(&self) -> Option { + Weak::upgrade(&self.inner).map(|inner| Tracer { inner }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + trait AssertSend: Send {} + + impl AssertSend for Tracer {} } diff --git a/src/error/mod.rs b/src/error/mod.rs index 28a3561..5246571 100644 --- a/src/error/mod.rs +++ b/src/error/mod.rs @@ -14,7 +14,7 @@ // limitations under the License. // -use tokio::{sync::oneshot, task::JoinError}; +pub(crate) const LOCK_MSG: &str = "should not cross threads/coroutines (locked)"; /// Skywalking Result. pub type Result = std::result::Result; @@ -22,9 +22,6 @@ pub type Result = std::result::Result; /// Skywalking Error. #[derive(Debug, thiserror::Error)] pub enum Error { - #[error("create span failed: {0}")] - CreateSpan(&'static str), - #[error("decode propagation failed: {0}")] DecodePropagation(&'static str), @@ -36,10 +33,4 @@ pub enum Error { #[error("tonic status: {0}")] TonicStatus(#[from] tonic::Status), - - #[error("tokio task join failed: {0}")] - TokioJoin(#[from] JoinError), - - #[error("tokio oneshot receive failed: {0}")] - TokioOneshotRecv(#[from] oneshot::error::RecvError), } diff --git a/src/lib.rs b/src/lib.rs index cc4ef8c..235d015 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -25,7 +25,7 @@ pub mod skywalking_proto { pub mod common; pub mod context; -mod error; +pub(crate) mod error; pub mod reporter; pub use error::{Error, Result}; diff --git a/src/reporter/grpc.rs b/src/reporter/grpc.rs index 499ccf5..afdd2d2 100644 --- a/src/reporter/grpc.rs +++ b/src/reporter/grpc.rs @@ -18,7 +18,8 @@ use super::Reporter; use crate::skywalking_proto::v3::{ trace_segment_report_service_client::TraceSegmentReportServiceClient, SegmentObject, }; -use futures_core::Stream; +use futures_util::stream; +use std::collections::LinkedList; use tonic::{ async_trait, transport::{self, Channel, Endpoint}, @@ -46,10 +47,8 @@ impl GrpcReporter { #[async_trait] impl Reporter for GrpcReporter { - async fn collect( - &mut self, - stream: impl Stream + Send + 'static, - ) -> crate::Result<()> { + async fn collect(&mut self, segments: LinkedList) -> crate::Result<()> { + let stream = stream::iter(segments); self.client.collect(stream).await?; Ok(()) } diff --git a/src/common/time.rs b/src/reporter/log.rs similarity index 66% rename from src/common/time.rs rename to src/reporter/log.rs index 8c28600..b08ef3c 100644 --- a/src/common/time.rs +++ b/src/reporter/log.rs @@ -14,7 +14,20 @@ // limitations under the License. // -pub trait TimeFetcher { - // Get current UNIX timestamp with sec resolution. - fn get(&self) -> i64; +use super::Reporter; +use crate::skywalking_proto::v3::SegmentObject; + +use std::collections::LinkedList; +use tonic::async_trait; + +pub struct LogReporter; + +#[async_trait] +impl Reporter for LogReporter { + async fn collect(&mut self, segments: LinkedList) -> crate::Result<()> { + for segment in segments { + tracing::info!(?segment, "Do trace"); + } + Ok(()) + } } diff --git a/src/reporter/mod.rs b/src/reporter/mod.rs index 1261020..c2f9fda 100644 --- a/src/reporter/mod.rs +++ b/src/reporter/mod.rs @@ -15,15 +15,15 @@ // pub mod grpc; +pub mod log; use crate::skywalking_proto::v3::SegmentObject; -use futures_core::Stream; +use std::collections::LinkedList; use tonic::async_trait; +pub(crate) type DynReporter = dyn Reporter + Send + Sync + 'static; + #[async_trait] pub trait Reporter { - async fn collect( - &mut self, - stream: impl Stream + Send + 'static, - ) -> crate::Result<()>; + async fn collect(&mut self, segments: LinkedList) -> crate::Result<()>; } diff --git a/tests/propagation.rs b/tests/propagation.rs index 52e5435..bb893c3 100644 --- a/tests/propagation.rs +++ b/tests/propagation.rs @@ -15,21 +15,14 @@ // #![allow(unused_imports)] -use skywalking::common::time::TimeFetcher; use skywalking::context::propagation::context::PropagationContext; use skywalking::context::propagation::decoder::decode_propagation; use skywalking::context::propagation::encoder::encode_propagation; use skywalking::context::trace_context::TracingContext; +use skywalking::context::tracer::Tracer; +use skywalking::reporter::log::LogReporter; use std::sync::Arc; -struct MockTimeFetcher {} - -impl TimeFetcher for MockTimeFetcher { - fn get(&self) -> i64 { - 100 - } -} - #[test] fn basic() { let data = "1-MQ==-NQ==-3-bWVzaA==-aW5zdGFuY2U=-L2FwaS92MS9oZWFsdGg=-ZXhhbXBsZS5jb206ODA4MA=="; @@ -71,8 +64,8 @@ fn invalid_sample() { #[test] fn basic_encode() { - let time_fetcher = MockTimeFetcher {}; - let tc = TracingContext::default_internal(Arc::new(time_fetcher), "mesh", "instance"); + let tracer = Tracer::new("mesh", "instance", LogReporter); + let tc = tracer.create_trace_context(); let res = encode_propagation(&tc, "/api/v1/health", "example.com:8080"); let res2 = decode_propagation(&res).unwrap(); assert!(res2.do_sample); diff --git a/tests/trace_context.rs b/tests/trace_context.rs index f50dff6..3d376de 100644 --- a/tests/trace_context.rs +++ b/tests/trace_context.rs @@ -14,18 +14,21 @@ // limitations under the License. // -#![allow(unused_imports)] - use prost::Message; -use skywalking::common::time::TimeFetcher; use skywalking::context::propagation::context::PropagationContext; use skywalking::context::propagation::decoder::decode_propagation; use skywalking::context::propagation::encoder::encode_propagation; use skywalking::context::trace_context::TracingContext; +use skywalking::context::tracer::Tracer; +use skywalking::reporter::log::LogReporter; +use skywalking::reporter::Reporter; use skywalking::skywalking_proto::v3::{ KeyStringValuePair, Log, RefType, SegmentObject, SegmentReference, SpanLayer, SpanObject, SpanType, }; +use std::collections::LinkedList; +use std::future; +use std::sync::Mutex; use std::{cell::Ref, sync::Arc}; /// Serialize from A should equal Serialize from B @@ -42,210 +45,245 @@ where assert_eq!(buf_a, buf_b); } -struct MockTimeFetcher {} +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn create_span() { + MockReporter::with( + |reporter| { + let tracer = Tracer::new("service", "instance", reporter); + let mut context = tracer.create_trace_context(); + assert_eq!(context.service(), "service"); + assert_eq!(context.service_instance(), "instance"); -impl TimeFetcher for MockTimeFetcher { - fn get(&self) -> i64 { - 100 - } -} + { + let mut span1 = context.create_entry_span("op1"); + let logs = vec![("hoge", "fuga"), ("hoge2", "fuga2")]; + let expected_log_message = logs + .to_owned() + .into_iter() + .map(|v| { + let (key, value) = v; + KeyStringValuePair { + key: key.to_string(), + value: value.to_string(), + } + }) + .collect(); + let expected_log = vec![Log { + time: 100, + data: expected_log_message, + }]; + span1.add_log(logs); -#[test] -fn create_span() { - let time_fetcher = MockTimeFetcher {}; - let mut context = - TracingContext::default_internal(Arc::new(time_fetcher), "service", "instance"); - assert_eq!(context.service, "service"); - assert_eq!(context.service_instance, "instance"); + let tags = vec![("hoge", "fuga")]; + let expected_tags = tags + .to_owned() + .into_iter() + .map(|v| { + let (key, value) = v; + KeyStringValuePair { + key: key.to_string(), + value: value.to_string(), + } + }) + .collect(); + span1.add_tag(tags[0].0, tags[0].1); - { - let mut span1 = context.create_entry_span("op1").unwrap(); - let logs = vec![("hoge", "fuga"), ("hoge2", "fuga2")]; - let expected_log_message = logs - .to_owned() - .into_iter() - .map(|v| { - let (key, value) = v; - KeyStringValuePair { - key: key.to_string(), - value: value.to_string(), + { + let _span2 = context.create_local_span("op2"); } - }) - .collect(); - let expected_log = vec![Log { - time: 100, - data: expected_log_message, - }]; - span1.add_log(logs); - - let tags = vec![("hoge", "fuga")]; - let expected_tags = tags - .to_owned() - .into_iter() - .map(|v| { - let (key, value) = v; - KeyStringValuePair { - key: key.to_string(), - value: value.to_string(), + + { + let span3 = context.create_exit_span("op3", "example.com/test"); + drop(span3); + + let span3_expected = SpanObject { + span_id: 2, + parent_span_id: 0, + start_time: 1, + end_time: 100, + refs: Vec::::new(), + operation_name: "op3".to_string(), + peer: "example.com/test".to_string(), + span_type: SpanType::Exit as i32, + span_layer: SpanLayer::Http as i32, + component_id: 11000, + is_error: false, + tags: Vec::::new(), + logs: Vec::::new(), + skip_analysis: false, + }; + context.with_spans(|spans| { + assert_eq!(spans.last(), Some(&span3_expected)); + }); } - }) - .collect(); - span1.add_tag(tags[0]); - { - let span2 = context.create_entry_span("op2"); - assert!(span2.is_err()); - } + { + let span4 = context.create_exit_span("op3", "example.com/test"); - { - let span3 = context.create_exit_span("op3", "example.com/test").unwrap(); - context.finalize_span(span3); - - let span3_expected = SpanObject { - span_id: 1, - parent_span_id: 0, - start_time: 100, - end_time: 100, - refs: Vec::::new(), - operation_name: "op3".to_string(), - peer: "example.com/test".to_string(), - span_type: SpanType::Exit as i32, - span_layer: SpanLayer::Http as i32, - component_id: 11000, - is_error: false, - tags: Vec::::new(), - logs: Vec::::new(), - skip_analysis: false, - }; - assert_eq!(*context.spans.last().unwrap().span_object(), span3_expected); - } + { + let span5 = context.create_exit_span("op4", "example.com/test"); + drop(span5); - { - let span4 = context.create_exit_span("op3", "example.com/test").unwrap(); + let span5_expected = SpanObject { + span_id: 3, + parent_span_id: 2, + start_time: 100, + end_time: 100, + refs: Vec::::new(), + operation_name: "op4".to_string(), + peer: "example.com/test".to_string(), + span_type: SpanType::Exit as i32, + span_layer: SpanLayer::Http as i32, + component_id: 11000, + is_error: false, + tags: Vec::::new(), + logs: Vec::::new(), + skip_analysis: false, + }; + context.with_spans(|spans| { + assert_eq!(spans.last(), Some(&span5_expected)); + }); + } + } - { - let span5 = context.create_exit_span("op4", "example.com/test").unwrap(); - context.finalize_span(span5); + drop(span1); - let span5_expected = SpanObject { - span_id: 3, - parent_span_id: 2, + let span1_expected = SpanObject { + span_id: 0, + parent_span_id: -1, start_time: 100, end_time: 100, refs: Vec::::new(), - operation_name: "op4".to_string(), - peer: "example.com/test".to_string(), - span_type: SpanType::Exit as i32, + operation_name: "op1".to_string(), + peer: String::default(), + span_type: SpanType::Entry as i32, span_layer: SpanLayer::Http as i32, component_id: 11000, is_error: false, - tags: Vec::::new(), - logs: Vec::::new(), + tags: expected_tags, + logs: expected_log, skip_analysis: false, }; - assert_eq!(*context.spans.last().unwrap().span_object(), span5_expected); + context.with_spans(|spans| { + assert_eq!(spans.last(), Some(&span1_expected)); + }); } - context.finalize_span(span4); - } + tracer + }, + |segment| { + assert_ne!(segment.trace_id.len(), 0); + assert_ne!(segment.trace_segment_id.len(), 0); + assert_eq!(segment.service, "service"); + assert_eq!(segment.service_instance, "instance"); + assert!(!segment.is_size_limited); + }, + ) + .await; +} - context.finalize_span(span1); - - let span1_expected = SpanObject { - span_id: 0, - parent_span_id: -1, - start_time: 100, - end_time: 100, - refs: Vec::::new(), - operation_name: "op1".to_string(), - peer: String::default(), - span_type: SpanType::Entry as i32, - span_layer: SpanLayer::Http as i32, - component_id: 11000, - is_error: false, - tags: expected_tags, - logs: expected_log, - skip_analysis: false, - }; - assert_eq!(*context.spans.last().unwrap().span_object(), span1_expected); - } +#[test] +#[should_panic] +fn create_span_failed() { + let tracer = Tracer::new("service", "instance", LogReporter); + let mut context = tracer.create_trace_context(); - let segment = context.convert_segment_object(); - assert_ne!(segment.trace_id.len(), 0); - assert_ne!(segment.trace_segment_id.len(), 0); - assert_eq!(segment.service, "service"); - assert_eq!(segment.service_instance, "instance"); - assert!(!segment.is_size_limited); + let _span1 = context.create_entry_span("op1"); + let _span2 = context.create_entry_span("op2"); } -#[test] -fn create_span_from_context() { +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn create_span_from_context() { let data = "1-MQ==-NQ==-3-bWVzaA==-aW5zdGFuY2U=-L2FwaS92MS9oZWFsdGg=-ZXhhbXBsZS5jb206ODA4MA=="; let prop = decode_propagation(data).unwrap(); - let time_fetcher = MockTimeFetcher {}; - let context = TracingContext::from_propagation_context_internal( - Arc::new(time_fetcher), - "service2", - "instance2", - prop, - ); - - let segment = context.convert_segment_object(); - assert_ne!(segment.trace_id.len(), 0); - assert_ne!(segment.trace_segment_id.len(), 0); - assert_eq!(segment.service, "service2"); - assert_eq!(segment.service_instance, "instance2"); - assert!(!segment.is_size_limited); + + MockReporter::with( + |reporter| { + let tracer = Tracer::new("service2", "instance2", reporter); + let _context = tracer.create_trace_context_from_propagation(prop); + tracer + }, + |segment| { + assert_ne!(segment.trace_id.len(), 0); + assert_ne!(segment.trace_segment_id.len(), 0); + assert_eq!(segment.service, "service2"); + assert_eq!(segment.service_instance, "instance2"); + assert!(!segment.is_size_limited); + }, + ) + .await; } #[test] fn crossprocess_test() { - let time_fetcher1 = MockTimeFetcher {}; - let mut context1 = - TracingContext::default_internal(Arc::new(time_fetcher1), "service", "instance"); - assert_eq!(context1.service, "service"); - assert_eq!(context1.service_instance, "instance"); + let tracer = Tracer::new("service", "instance", LogReporter); + let mut context1 = tracer.create_trace_context(); + assert_eq!(context1.service(), "service"); + assert_eq!(context1.service_instance(), "instance"); - let span1 = context1.create_entry_span("op1").unwrap(); + let _span1 = context1.create_entry_span("op1"); { - let span2 = context1.create_exit_span("op2", "remote_peer").unwrap(); + let _span2 = context1.create_exit_span("op2", "remote_peer"); { let enc_prop = encode_propagation(&context1, "endpoint", "address"); let dec_prop = decode_propagation(&enc_prop).unwrap(); - let time_fetcher2 = MockTimeFetcher {}; - let mut context2 = TracingContext::from_propagation_context_internal( - Arc::new(time_fetcher2), - "service2", - "instance2", - dec_prop, - ); - - let span3 = context2.create_entry_span("op2").unwrap(); - context2.finalize_span(span3); - - let span3 = context2.spans.last().unwrap(); - assert_eq!(span3.span_object().span_id, 0); - assert_eq!(span3.span_object().parent_span_id, -1); - assert_eq!(span3.span_object().refs.len(), 1); - - let expected_ref = SegmentReference { - ref_type: RefType::CrossProcess as i32, - trace_id: context2.trace_id, - parent_trace_segment_id: context1.trace_segment_id.clone(), - parent_span_id: 1, - parent_service: context1.service.clone(), - parent_service_instance: context1.service_instance.clone(), - parent_endpoint: "endpoint".to_string(), - network_address_used_at_peer: "address".to_string(), - }; - - check_serialize_equivalent(&expected_ref, &span3.span_object().refs[0]); + let tracer = Tracer::new("service2", "instance2", LogReporter); + let mut context2 = tracer.create_trace_context_from_propagation(dec_prop); + + let span3 = context2.create_entry_span("op2"); + drop(span3); + + context2.with_spans(|spans| { + let span3 = spans.last().unwrap(); + return; + + let span3 = spans.last().unwrap(); + assert_eq!(span3.span_id, 0); + assert_eq!(span3.parent_span_id, -1); + assert_eq!(span3.refs.len(), 1); + + let expected_ref = SegmentReference { + ref_type: RefType::CrossProcess as i32, + trace_id: context2.trace_id().to_owned(), + parent_trace_segment_id: context1.trace_segment_id().to_owned(), + parent_span_id: 1, + parent_service: context1.service().to_owned(), + parent_service_instance: context1.service_instance().to_owned(), + parent_endpoint: "endpoint".to_string(), + network_address_used_at_peer: "address".to_string(), + }; + + check_serialize_equivalent(&expected_ref, &span3.refs[0]); + }); } + } +} + +#[derive(Default, Clone)] +struct MockReporter { + segments: Arc>>, +} + +impl MockReporter { + async fn with(f1: impl FnOnce(MockReporter) -> Tracer, f2: impl FnOnce(&SegmentObject)) { + let reporter = MockReporter::default(); - context1.finalize_span(span2); + let tracer = f1(reporter.clone()); + + tracer.reporting(future::ready(())).await.unwrap(); + + let segments = reporter.segments.try_lock().unwrap(); + let segment = segments.front().unwrap(); + f2(segment); } +} - context1.finalize_span(span1); +#[tonic::async_trait] +impl Reporter for MockReporter { + async fn collect(&mut self, segments: LinkedList) -> skywalking::Result<()> { + self.segments.try_lock().unwrap().extend(segments); + Ok(()) + } }