Improve LogReporter and fix tests. (#36)

Author: jmjoy

.github/workflows/ci.yaml

@@ -32,6 +32,18 @@ env:
 
 jobs:
   CI:
+    needs:
+    - license-check
+    - fmt
+    - check-and-test
+    - doc
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+        with:
+          submodules: 'recursive'
+
+  license-check:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v2
@@ -41,17 +53,50 @@ jobs:
         uses: apache/skywalking-eyes/header/@501a28d2fb4a9b962661987e50cf0219631b32ff
         with:
           config: .github/licenserc.yaml
-      - name: Check dependencies License
+      - name: Check Dependencies License
         uses: apache/skywalking-eyes/dependency/@501a28d2fb4a9b962661987e50cf0219631b32ff
         with:
           config: .github/licenserc.yaml
+
+  fmt:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+        with:
+          submodules: 'recursive'
       - name: Install Rust toolchain
-        run: rustup toolchain install stable --component clippy --component rustfmt
+        run: rustup toolchain install stable --component rustfmt
       - name: Run format
         run: cargo fmt --all -- --check
+
+  check-and-test:
+    strategy:
+      fail-fast: false
+      matrix:
+        features:
+          - ""
+          - "--all-features"
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+        with:
+          submodules: 'recursive'
+      - name: Install Rust toolchain
+        run: rustup toolchain install stable --component clippy
+      - name: Run check
+        run: cargo check --workspace --release ${{ matrix.features }}
       - name: Run clippy
-        run: cargo clippy --workspace --tests -- -D warnings
+        run: cargo clippy --workspace --release ${{ matrix.features }} -- -D warnings
       - name: Run tests
-        run: cargo test --workspace
+        run: cargo test --workspace --release ${{ matrix.features }}
+
+  doc:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+        with:
+          submodules: 'recursive'
+      - name: Install Rust toolchain
+        run: rustup toolchain install stable
       - name: Run docs
-        run: cargo rustdoc --all-features -- -D warnings
+        run: cargo rustdoc --release -- -D warnings

Cargo.toml

@@ -55,7 +55,7 @@ uuid = { version = "1.1.0", features = ["serde", "v4"] }
 tonic-build = "0.7.2"
 
 [dev-dependencies]
-tokio = { version = "1.18.2", features = ["rt-multi-thread"] }
+tokio = { version = "1.18.2", features = ["rt-multi-thread", "signal"] }
 tokio-stream = { version = "0.1.8", features = ["net"] }
 
 [[test]]

README.md

@@ -81,11 +81,11 @@ async fn main() -> Result<(), Box<dyn Error>> {
     tokio::spawn(handle_request(tracer.clone()));
 
     // Start to report.
-    let handle = tracer.reporting(async move {
-        let _ = signal::ctrl_c().await;
-    });
-
-    handle.await?;
+    tracer
+        .reporting(async move {
+            let _ = signal::ctrl_c().await.unwrap();
+        })
+        .await?;
 
     Ok(())
 }
@@ -104,6 +104,7 @@ use `cargo run --example simple_trace_report`
 which outputs executable, then run it.
 
 # Release
+
 The SkyWalking committer(PMC included) could follow [this doc](Release-guide.md) to release an official version.
 
 # License

e2e/src/main.rs

@@ -25,7 +25,7 @@ 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::future;
 use std::net::SocketAddr;
 use structopt::StructOpt;
 
@@ -153,12 +153,12 @@ async fn main() -> Result<(), Box<dyn Error>> {
 
     let handle = if opt.mode == "consumer" {
         tracer::set_global_tracer(Tracer::new("consumer", "node_0", reporter));
-        let handle = tracer::reporting(pending());
+        let handle = tracer::reporting(future::pending());
         run_consumer_service([0, 0, 0, 0]).await;
         handle
     } else if opt.mode == "producer" {
         tracer::set_global_tracer(Tracer::new("producer", "node_0", reporter));
-        let handle = tracer::reporting(pending());
+        let handle = tracer::reporting(future::pending());
         run_producer_service([0, 0, 0, 0]).await;
         handle
     } else {

examples/simple_trace_report.rs

@@ -54,11 +54,11 @@ async fn main() -> Result<(), Box<dyn Error>> {
     tokio::spawn(handle_request(tracer.clone()));
 
     // Start to report.
-    let handle = tracer.reporting(async move {
-        let _ = signal::ctrl_c().await;
-    });
-
-    handle.await?;
+    tracer
+        .reporting(async move {
+            let _ = signal::ctrl_c().await.unwrap();
+        })
+        .await?;
 
     Ok(())
 }

src/context/tracer.rs

@@ -20,9 +20,12 @@ use crate::{
     skywalking_proto::v3::SegmentObject,
 };
 use std::future::Future;
+use std::pin::Pin;
 use std::sync::Weak;
+use std::task::{Context, Poll};
 use std::{collections::LinkedList, sync::Arc};
 use tokio::sync::OnceCell;
+use tokio::task::JoinError;
 use tokio::{
     sync::{
         mpsc::{self},
@@ -58,9 +61,7 @@ pub fn create_trace_context_from_propagation(context: PropagationContext) -> Tra
 /// 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<Output = ()> + Send + Sync + 'static,
-) -> JoinHandle<()> {
+pub fn reporting(shutdown_signal: impl Future<Output = ()> + Send + Sync + 'static) -> Reporting {
     global_tracer().reporting(shutdown_signal)
 }
 
@@ -142,8 +143,10 @@ impl Tracer {
     pub fn reporting(
         &self,
         shutdown_signal: impl Future<Output = ()> + Send + Sync + 'static,
-    ) -> JoinHandle<()> {
-        tokio::spawn(Self::do_reporting(self.clone(), shutdown_signal))
+    ) -> Reporting {
+        Reporting {
+            handle: tokio::spawn(self.clone().do_reporting(shutdown_signal)),
+        }
     }
 
     async fn do_reporting(self, shutdown_signal: impl Future<Output = ()> + Send + Sync + 'static) {
@@ -216,6 +219,19 @@ impl WeakTracer {
     }
 }
 
+/// Created by [Tracer::reporting].
+pub struct Reporting {
+    handle: JoinHandle<()>,
+}
+
+impl Future for Reporting {
+    type Output = Result<(), JoinError>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        Pin::new(&mut self.handle).poll(cx)
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

src/reporter/grpc.rs

@@ -19,7 +19,7 @@ use crate::skywalking_proto::v3::{
     trace_segment_report_service_client::TraceSegmentReportServiceClient, SegmentObject,
 };
 use futures_util::stream;
-use std::collections::LinkedList;
+use std::{collections::LinkedList, error::Error};
 use tonic::{
     async_trait,
     transport::{self, Channel, Endpoint},
@@ -47,7 +47,7 @@ impl GrpcReporter {
 
 #[async_trait]
 impl Reporter for GrpcReporter {
-    async fn collect(&mut self, segments: LinkedList<SegmentObject>) -> crate::Result<()> {
+    async fn collect(&mut self, segments: LinkedList<SegmentObject>) -> Result<(), Box<dyn Error>> {
         let stream = stream::iter(segments);
         self.client.collect(stream).await?;
         Ok(())

src/reporter/log.rs

@@ -16,16 +16,58 @@
 
 use super::Reporter;
 use crate::skywalking_proto::v3::SegmentObject;
-use std::collections::LinkedList;
+use std::{collections::LinkedList, error::Error};
 use tonic::async_trait;
 
-pub struct LogReporter;
+enum Used {
+    Println,
+    Tracing,
+}
+
+pub struct LogReporter {
+    tip: String,
+    used: Used,
+}
+
+impl LogReporter {
+    #[inline]
+    pub fn new() -> Self {
+        Default::default()
+    }
+
+    pub fn tip(mut self, tip: String) -> Self {
+        self.tip = tip;
+        self
+    }
+
+    pub fn use_tracing(mut self) -> Self {
+        self.used = Used::Tracing;
+        self
+    }
+
+    pub fn use_println(mut self) -> Self {
+        self.used = Used::Println;
+        self
+    }
+}
+
+impl Default for LogReporter {
+    fn default() -> Self {
+        Self {
+            tip: "Collect".to_string(),
+            used: Used::Println,
+        }
+    }
+}
 
 #[async_trait]
 impl Reporter for LogReporter {
-    async fn collect(&mut self, segments: LinkedList<SegmentObject>) -> crate::Result<()> {
+    async fn collect(&mut self, segments: LinkedList<SegmentObject>) -> Result<(), Box<dyn Error>> {
         for segment in segments {
-            tracing::info!(?segment, "Do trace");
+            match self.used {
+                Used::Println => println!("{} segment={:?}", self.tip, segment),
+                Used::Tracing => tracing::info!(?segment, "{}", self.tip),
+            }
         }
         Ok(())
     }

src/reporter/mod.rs

@@ -19,11 +19,12 @@ pub mod log;
 
 use crate::skywalking_proto::v3::SegmentObject;
 use std::collections::LinkedList;
+use std::{error::Error, result::Result};
 use tonic::async_trait;
 
 pub(crate) type DynReporter = dyn Reporter + Send + Sync + 'static;
 
 #[async_trait]
 pub trait Reporter {
-    async fn collect(&mut self, segments: LinkedList<SegmentObject>) -> crate::Result<()>;
+    async fn collect(&mut self, segments: LinkedList<SegmentObject>) -> Result<(), Box<dyn Error>>;
 }

tests/propagation.rs

@@ -64,7 +64,7 @@ fn invalid_sample() {
 
 #[test]
 fn basic_encode() {
-    let tracer = Tracer::new("mesh", "instance", LogReporter);
+    let tracer = Tracer::new("mesh", "instance", LogReporter::new());
     let tc = tracer.create_trace_context();
     let res = encode_propagation(&tc, "/api/v1/health", "example.com:8080");
     let res2 = decode_propagation(&res).unwrap();