diff --git a/Cargo.toml b/Cargo.toml
index f46438e..498478e 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -62,6 +62,10 @@ tokio-stream = { version = "0.1.8", features = ["net"] }
 name = "trace_context"
 required-features = ["mock"]
 
+[[test]]
+name = "logging"
+required-features = ["mock"]
+
 [[example]]
 name = "simple_trace_report"
 path = "examples/simple_trace_report.rs"
diff --git a/README.md b/README.md
index 573d147..169b391 100644
--- a/README.md
+++ b/README.md
@@ -16,7 +16,9 @@ and core concepts to keep best compatibility and performance.
 
 All concepts are from the official SkyWalking definitions.
 
-## Span
+## Tracing
+
+### Span
 
 Span is an important and common concept in distributed tracing system. Learn Span from Google Dapper Paper. For better
 performance, we extend the span into 3 kinds.
@@ -34,19 +36,29 @@ Tag and Log are similar attributes of the span.
 - Log is heavier than tag, with one timestamp and multiple key:value pairs. Log represents an event, typically an error
   happens.
 
-## TracingContext
+### TracingContext
 
 TracingContext is the context of the tracing process. Span should only be created through context, and be archived into
 the context after the span finished.
 
+## Logging
+
+### LogRecord
+
+LogRecord is the simple builder for the LogData, which is the Log format of Skywalking.
+
 # Example
 
 ```rust, no_run
-use skywalking::{reporter::grpc::GrpcReporter, trace::tracer::Tracer};
+use skywalking::{
+    logging::{logger::Logger, record::{LogRecord, RecordType}},
+    reporter::grpc::GrpcReporter,
+    trace::tracer::Tracer,
+};
 use std::error::Error;
 use tokio::signal;
 
-async fn handle_request(tracer: Tracer) {
+async fn handle_request(tracer: Tracer, logger: Logger) {
     let mut ctx = tracer.create_trace_context();
 
     {
@@ -59,10 +71,20 @@ async fn handle_request(tracer: Tracer) {
 
         {
             // Generates an Exit Span when executing an RPC.
-            let _span2 = ctx.create_exit_span("op2", "remote_peer");
+            let span2 = ctx.create_exit_span("op2", "remote_peer");
 
             // Something...
 
+            // Do logging.
+            logger.log(
+                LogRecord::new()
+                    .add_tag("level", "INFO")
+                    .with_tracing_context(&ctx)
+                    .with_span(&span2)
+                    .record_type(RecordType::Text)
+                    .content("Something...")
+            );
+
             // Auto close span2 when dropped.
         }
 
@@ -74,17 +96,24 @@ async fn handle_request(tracer: Tracer) {
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn Error>> {
+    // Connect to skywalking oap server.
     let reporter = GrpcReporter::connect("http://0.0.0.0:11800").await?;
-    let tracer = Tracer::new("service", "instance", reporter);
 
-    tokio::spawn(handle_request(tracer.clone()));
-
-    // Start to report.
-    tracer
-        .reporting(async move {
-            let _ = signal::ctrl_c().await.unwrap();
+    // Spawn the reporting in background, with listening the graceful shutdown signal.
+    let handle = reporter
+        .reporting()
+        .await
+        .with_graceful_shutdown(async move {
+            signal::ctrl_c().await.expect("failed to listen for event");
         })
-        .await?;
+        .spawn();
+
+    let tracer = Tracer::new("service", "instance", reporter.clone());
+    let logger = Logger::new("service", "instance", reporter);
+
+    handle_request(tracer, logger).await;
+
+    handle.await?;
 
     Ok(())
 }
diff --git a/docker-compose.e2e.yml b/docker-compose.e2e.yml
index a1f15fb..3462189 100644
--- a/docker-compose.e2e.yml
+++ b/docker-compose.e2e.yml
@@ -18,7 +18,7 @@
 version: "3.9"
 services:
   collector:
-    image: ghcr.io/apache/skywalking-agent-test-tool/mock-collector:5acb890f225ca37ee60675ce3e330545e23e3cbc
+    image: ghcr.io/apache/skywalking-agent-test-tool/mock-collector:f4f5ef22b1df623464772816bb6b42ba611444ff
     ports:
       - "19876:19876"
       - "12800:12800"
diff --git a/e2e/data/expected_context.yaml b/e2e/data/expected_context.yaml
index f4ff6e1..9778a2a 100644
--- a/e2e/data/expected_context.yaml
+++ b/e2e/data/expected_context.yaml
@@ -16,7 +16,7 @@
 # under the License.
 #
 segmentItems:
-- segmentSize: gt 1
+- segmentSize: ge 2
   segments:
   - segmentId: not null
     spans:
@@ -76,7 +76,7 @@ segmentItems:
       spanType: Entry
       startTime: gt 0
   serviceName: producer
-- segmentSize: 1
+- segmentSize: ge 1
   segments:
   - segmentId: not null
     spans:
@@ -101,3 +101,48 @@ segmentItems:
       spanType: Entry
       startTime: gt 0
   serviceName: consumer
+
+logItems:
+- logSize: ge 2
+  logs:
+  - body:
+      content:
+        json: '{"message": "handle ping"}'
+      type: ''
+    endpoint: /ping
+    layer: ''
+    tags:
+      data:
+      - key: level
+        value: DEBUG
+    timestamp: gt 0
+  - body:
+      content:
+        text: do http request
+      type: ''
+    endpoint: /ping
+    layer: ''
+    tags:
+      data:
+      - key: level
+        value: INFO
+    timestamp: gt 0
+    traceContext:
+      spanId: 1
+      traceId: not null
+      traceSegmentId: not null
+  serviceName: producer
+- logSize: ge 1
+  logs:
+  - body:
+      content:
+        json: '{"message": "handle pong"}'
+      type: ''
+    endpoint: /pong
+    layer: ''
+    tags:
+      data:
+      - key: level
+        value: DEBUG
+    timestamp: gt 0
+  serviceName: consumer
diff --git a/e2e/rust-toolchain.toml b/e2e/rust-toolchain.toml
deleted file mode 100644
index 09cf6d7..0000000
--- a/e2e/rust-toolchain.toml
+++ /dev/null
@@ -1,20 +0,0 @@
-# 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.
-#
-[toolchain]
-channel = "1.57.0"
-components = ["rustfmt", "clippy"]
diff --git a/e2e/src/main.rs b/e2e/src/main.rs
index 2b1780c..f3ffd57 100644
--- a/e2e/src/main.rs
+++ b/e2e/src/main.rs
@@ -21,6 +21,10 @@ use hyper::{
     Body, Client, Method, Request, Response, Server, StatusCode,
 };
 use skywalking::{
+    logging::{
+        logger::{self, Logger},
+        record::{LogRecord, RecordType},
+    },
     reporter::grpc::GrpcReporter,
     trace::{
         propagation::{
@@ -30,7 +34,7 @@ use skywalking::{
         tracer::{self, Tracer},
     },
 };
-use std::{convert::Infallible, error::Error, future, net::SocketAddr};
+use std::{convert::Infallible, error::Error, net::SocketAddr};
 use structopt::StructOpt;
 
 static NOT_FOUND_MSG: &str = "not found";
@@ -40,10 +44,18 @@ async fn handle_ping(
     _req: Request<Body>,
     client: Client<HttpConnector>,
 ) -> Result<Response<Body>, Infallible> {
+    logger::log(
+        LogRecord::new()
+            .add_tag("level", "DEBUG")
+            .endpoint("/ping")
+            .record_type(RecordType::Json)
+            .content(r#"{"message": "handle ping"}"#),
+    );
+
     let mut context = tracer::create_trace_context();
     let _span = context.create_entry_span("/ping");
     {
-        let _span2 = context.create_exit_span("/pong", "consumer:8082");
+        let span2 = context.create_exit_span("/pong", "consumer:8082");
         let header = encode_propagation(&context, "/pong", "consumer:8082");
         let req = Request::builder()
             .method(Method::GET)
@@ -52,6 +64,15 @@ async fn handle_ping(
             .body(Body::from(""))
             .unwrap();
 
+        logger::log(
+            LogRecord::new()
+                .add_tag("level", "INFO")
+                .endpoint("/ping")
+                .with_tracing_context(&context)
+                .with_span(&span2)
+                .record_type(RecordType::Text)
+                .content("do http request"),
+        );
         client.request(req).await.unwrap();
     }
     {
@@ -106,6 +127,14 @@ async fn run_producer_service(host: [u8; 4]) {
 }
 
 async fn handle_pong(_req: Request<Body>) -> Result<Response<Body>, Infallible> {
+    logger::log(
+        LogRecord::new()
+            .add_tag("level", "DEBUG")
+            .endpoint("/pong")
+            .record_type(RecordType::Json)
+            .content(r#"{"message": "handle pong"}"#),
+    );
+
     let ctx = decode_propagation(
         _req.headers()[SKYWALKING_HTTP_CONTEXT_HEADER_KEY]
             .to_str()
@@ -154,20 +183,19 @@ struct Opt {
 async fn main() -> Result<(), Box<dyn Error>> {
     let opt = Opt::from_args();
     let reporter = GrpcReporter::connect("http://collector:19876").await?;
+    let handle = reporter.reporting().await.spawn();
 
-    let handle = if opt.mode == "consumer" {
-        tracer::set_global_tracer(Tracer::new("consumer", "node_0", reporter));
-        let handle = tracer::reporting(future::pending());
+    if opt.mode == "consumer" {
+        tracer::set_global_tracer(Tracer::new("consumer", "node_0", reporter.clone()));
+        logger::set_global_logger(Logger::new("consumer", "node_0", reporter));
         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(future::pending());
+        tracer::set_global_tracer(Tracer::new("producer", "node_0", reporter.clone()));
+        logger::set_global_logger(Logger::new("producer", "node_0", reporter));
         run_producer_service([0, 0, 0, 0]).await;
-        handle
     } else {
         unreachable!()
-    };
+    }
 
     handle.await?;
 
diff --git a/examples/simple_log_report.rs b/examples/simple_log_report.rs
new file mode 100644
index 0000000..0466e9a
--- /dev/null
+++ b/examples/simple_log_report.rs
@@ -0,0 +1,44 @@
+// 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 skywalking::{
+    logging::{logger::Logger, record::LogRecord},
+    reporter::grpc::GrpcReporter,
+};
+use std::{error::Error, future};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    // Connect to skywalking oap server.
+    let reporter = GrpcReporter::connect("http://0.0.0.0:11800").await?;
+
+    // Do logging.
+    let logger = Logger::new("service", "instance", reporter.clone());
+    logger.log(LogRecord::new().content("something to log"));
+
+    // Start reporting and quit immediately when have completed the existing
+    // collection.
+    reporter
+        .reporting()
+        .await
+        .with_graceful_shutdown(future::ready(()))
+        .start()
+        .await?;
+
+    Ok(())
+}
diff --git a/examples/simple_trace_report.rs b/examples/simple_trace_report.rs
index af382f2..68f6e7d 100644
--- a/examples/simple_trace_report.rs
+++ b/examples/simple_trace_report.rs
@@ -48,17 +48,25 @@ async fn handle_request(tracer: Tracer) {
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn Error>> {
+    // Connect to skywalking oap server.
     let reporter = GrpcReporter::connect("http://0.0.0.0:11800").await?;
-    let tracer = Tracer::new("service", "instance", reporter);
-
-    tokio::spawn(handle_request(tracer.clone()));
 
-    // Start to report.
-    tracer
-        .reporting(async move {
-            let _ = signal::ctrl_c().await.unwrap();
+    // Spawn the reporting in background, with listening the graceful shutdown
+    // signal.
+    let handle = reporter
+        .reporting()
+        .await
+        .with_graceful_shutdown(async move {
+            signal::ctrl_c().await.expect("failed to listen for event");
         })
-        .await?;
+        .spawn();
+
+    // Do tracing.
+    let tracer = Tracer::new("service", "instance", reporter);
+    handle_request(tracer).await;
+
+    // Wait the reporting to quit.
+    handle.await?;
 
     Ok(())
 }
diff --git a/src/common/mod.rs b/src/common/mod.rs
index 7bb6d17..7c59317 100644
--- a/src/common/mod.rs
+++ b/src/common/mod.rs
@@ -15,3 +15,4 @@
 //
 
 pub mod random_generator;
+pub(crate) mod system_time;
diff --git a/src/trace/system_time.rs b/src/common/system_time.rs
similarity index 100%
rename from src/trace/system_time.rs
rename to src/common/system_time.rs
diff --git a/src/logging/logger.rs b/src/logging/logger.rs
index 8cda4c1..2571ea7 100644
--- a/src/logging/logger.rs
+++ b/src/logging/logger.rs
@@ -14,4 +14,70 @@
 // limitations under the License.
 //
 
-pub struct Logger {}
+use super::record::LogRecord;
+use crate::reporter::{CollectItem, DynReport, Report};
+use std::sync::Arc;
+use tokio::sync::OnceCell;
+
+static GLOBAL_LOGGER: OnceCell<Logger> = OnceCell::const_new();
+
+/// Set the global logger.
+pub fn set_global_logger(tracer: Logger) {
+    if GLOBAL_LOGGER.set(tracer).is_err() {
+        panic!("global logger has setted")
+    }
+}
+
+/// Get the global logger.
+pub fn global_logger() -> &'static Logger {
+    GLOBAL_LOGGER.get().expect("global logger haven't setted")
+}
+
+/// Log by global logger.
+pub fn log(record: LogRecord) {
+    global_logger().log(record);
+}
+
+pub struct Inner {
+    service_name: String,
+    instance_name: String,
+    reporter: Box<DynReport>,
+}
+
+#[derive(Clone)]
+pub struct Logger {
+    inner: Arc<Inner>,
+}
+
+impl Logger {
+    /// New with service info and reporter.
+    pub fn new(
+        service_name: impl ToString,
+        instance_name: impl ToString,
+        reporter: impl Report + Send + Sync + 'static,
+    ) -> Self {
+        Self {
+            inner: Arc::new(Inner {
+                service_name: service_name.to_string(),
+                instance_name: instance_name.to_string(),
+                reporter: Box::new(reporter),
+            }),
+        }
+    }
+
+    pub fn service_name(&self) -> &str {
+        &self.inner.service_name
+    }
+
+    pub fn instance_name(&self) -> &str {
+        &self.inner.instance_name
+    }
+
+    pub fn log(&self, record: LogRecord) {
+        let data = record.convert_to_log_data(
+            self.service_name().to_owned(),
+            self.instance_name().to_owned(),
+        );
+        self.inner.reporter.report(CollectItem::Log(data));
+    }
+}
diff --git a/src/logging/record.rs b/src/logging/record.rs
index ecad7e6..87f22cf 100644
--- a/src/logging/record.rs
+++ b/src/logging/record.rs
@@ -14,4 +14,152 @@
 // limitations under the License.
 //
 
-pub struct LogRecord {}
+use crate::{
+    common::system_time::{fetch_time, TimePeriod},
+    skywalking_proto::v3::{
+        log_data_body::Content, JsonLog, KeyStringValuePair, LogData, LogDataBody, LogTags,
+        TextLog, TraceContext, YamlLog,
+    },
+    trace::{span::Span, trace_context::TracingContext},
+};
+use std::time::{SystemTime, UNIX_EPOCH};
+
+pub enum RecordType {
+    Text,
+    Json,
+    Yaml,
+}
+
+impl Default for RecordType {
+    fn default() -> Self {
+        Self::Text
+    }
+}
+
+#[derive(Default)]
+pub struct LogRecord {
+    time: Option<SystemTime>,
+    is_ignore_time: bool,
+    endpoint: String,
+    tags: Vec<(String, String)>,
+    trace_id: Option<String>,
+    trace_segment_id: Option<String>,
+    span_id: Option<i32>,
+    record_type: RecordType,
+    content: String,
+}
+
+impl LogRecord {
+    #[inline]
+    pub fn new() -> Self {
+        Default::default()
+    }
+
+    pub fn custome_time(mut self, time: SystemTime) -> Self {
+        self.time = Some(time);
+        self
+    }
+
+    pub fn ignore_time(mut self) -> Self {
+        self.is_ignore_time = true;
+        self
+    }
+
+    pub fn endpoint(mut self, endpoint: impl ToString) -> Self {
+        self.endpoint = endpoint.to_string();
+        self
+    }
+
+    pub fn add_tag(mut self, key: impl ToString, value: impl ToString) -> Self {
+        self.tags.push((key.to_string(), value.to_string()));
+        self
+    }
+
+    pub fn add_tags<K, V, I>(mut self, tags: I) -> Self
+    where
+        K: ToString,
+        V: ToString,
+        I: IntoIterator<Item = (K, V)>,
+    {
+        self.tags.extend(
+            tags.into_iter()
+                .map(|(k, v)| (k.to_string(), v.to_string())),
+        );
+        self
+    }
+
+    pub fn with_tracing_context(mut self, tracing_context: &TracingContext) -> Self {
+        self.trace_id = Some(tracing_context.trace_id().to_owned());
+        self.trace_segment_id = Some(tracing_context.trace_segment_id().to_owned());
+        self
+    }
+
+    pub fn with_span(mut self, span: &Span) -> Self {
+        self.span_id = Some(span.with_span_object(|span| span.span_id));
+        self
+    }
+
+    pub fn record_type(mut self, record_type: RecordType) -> Self {
+        self.record_type = record_type;
+        self
+    }
+
+    pub fn content(mut self, content: impl ToString) -> Self {
+        self.content = content.to_string();
+        self
+    }
+
+    pub(crate) fn convert_to_log_data(
+        self,
+        service_name: String,
+        instance_name: String,
+    ) -> LogData {
+        let timestamp = if self.is_ignore_time {
+            0
+        } else {
+            match self.time {
+                Some(time) => time
+                    .duration_since(UNIX_EPOCH)
+                    .map(|dur| dur.as_millis() as i64)
+                    .unwrap_or_default(),
+                None => fetch_time(TimePeriod::Log),
+            }
+        };
+        let trace_context = match (self.trace_id, self.trace_segment_id, self.span_id) {
+            (Some(trace_id), Some(trace_segment_id), Some(span_id)) => Some(TraceContext {
+                trace_id,
+                trace_segment_id,
+                span_id,
+            }),
+            _ => None,
+        };
+        let tags = if self.tags.is_empty() {
+            None
+        } else {
+            let data = self
+                .tags
+                .into_iter()
+                .map(|(key, value)| KeyStringValuePair { key, value })
+                .collect();
+            Some(LogTags { data })
+        };
+
+        LogData {
+            timestamp,
+            service: service_name,
+            service_instance: instance_name,
+            endpoint: self.endpoint,
+            body: Some(LogDataBody {
+                r#type: "".to_owned(),
+                content: match self.record_type {
+                    RecordType::Text => Some(Content::Text(TextLog { text: self.content })),
+                    RecordType::Json => Some(Content::Json(JsonLog { json: self.content })),
+                    RecordType::Yaml => Some(Content::Yaml(YamlLog { yaml: self.content })),
+                },
+            }),
+            trace_context,
+            tags,
+            layer: Default::default(),
+        }
+    }
+}
diff --git a/src/reporter/grpc.rs b/src/reporter/grpc.rs
index fb04364..eb92517 100644
--- a/src/reporter/grpc.rs
+++ b/src/reporter/grpc.rs
@@ -14,42 +14,332 @@
 // limitations under the License.
 //
 
-use super::Reporter;
-use crate::skywalking_proto::v3::{
-    trace_segment_report_service_client::TraceSegmentReportServiceClient, SegmentObject,
+use crate::{
+    reporter::{CollectItem, Report},
+    skywalking_proto::v3::{
+        log_report_service_client::LogReportServiceClient,
+        trace_segment_report_service_client::TraceSegmentReportServiceClient, LogData,
+        SegmentObject,
+    },
 };
 use futures_util::stream;
-use std::{collections::LinkedList, error::Error};
+use std::{
+    collections::LinkedList,
+    error::Error,
+    future::{pending, Future},
+    mem::take,
+    pin::Pin,
+    sync::{
+        atomic::{AtomicBool, Ordering},
+        Arc,
+    },
+    task::{Context, Poll},
+};
+use tokio::{
+    select,
+    sync::{mpsc, Mutex},
+    task::JoinHandle,
+    try_join,
+};
 use tonic::{
     async_trait,
     transport::{self, Channel, Endpoint},
 };
 
-type ReporterClient = TraceSegmentReportServiceClient<Channel>;
+pub trait CollectItemProduce: Send + Sync + 'static {
+    fn produce(&self, item: CollectItem) -> Result<(), Box<dyn Error>>;
+}
+
+impl CollectItemProduce for () {
+    fn produce(&self, _item: CollectItem) -> Result<(), Box<dyn Error>> {
+        Ok(())
+    }
+}
+
+impl CollectItemProduce for mpsc::UnboundedSender<CollectItem> {
+    fn produce(&self, item: CollectItem) -> Result<(), Box<dyn Error>> {
+        Ok(self.send(item)?)
+    }
+}
+
+#[async_trait]
+pub trait ColletcItemConsume: Send + Sync + 'static {
+    async fn consume(&mut self) -> Result<Option<CollectItem>, Box<dyn Error + Send>>;
+
+    async fn try_consume(&mut self) -> Result<Option<CollectItem>, Box<dyn Error + Send>>;
+}
+
+#[async_trait]
+impl ColletcItemConsume for () {
+    async fn consume(&mut self) -> Result<Option<CollectItem>, Box<dyn Error + Send>> {
+        Ok(None)
+    }
+
+    async fn try_consume(&mut self) -> Result<Option<CollectItem>, Box<dyn Error + Send>> {
+        Ok(None)
+    }
+}
+
+#[async_trait]
+impl ColletcItemConsume for mpsc::UnboundedReceiver<CollectItem> {
+    async fn consume(&mut self) -> Result<Option<CollectItem>, Box<dyn Error + Send>> {
+        Ok(self.recv().await)
+    }
+
+    async fn try_consume(&mut self) -> Result<Option<CollectItem>, Box<dyn Error + Send>> {
+        use mpsc::error::TryRecvError;
+
+        match self.try_recv() {
+            Ok(item) => Ok(Some(item)),
+            Err(e) => match e {
+                TryRecvError::Empty => Ok(None),
+                TryRecvError::Disconnected => Err(Box::new(e)),
+            },
+        }
+    }
+}
 
-pub struct GrpcReporter {
-    client: ReporterClient,
+struct Inner<P, C> {
+    trace_client: Mutex<TraceSegmentReportServiceClient<Channel>>,
+    log_client: Mutex<LogReportServiceClient<Channel>>,
+    producer: P,
+    consumer: Mutex<Option<C>>,
+    is_reporting: AtomicBool,
+    is_closed: AtomicBool,
 }
 
-impl GrpcReporter {
+pub type DynErrHandle = dyn Fn(Box<dyn Error>) + Send + Sync + 'static;
+
+pub struct GrpcReporter<P, C> {
+    inner: Arc<Inner<P, C>>,
+    err_handle: Arc<Option<Box<DynErrHandle>>>,
+}
+
+impl GrpcReporter<mpsc::UnboundedSender<CollectItem>, mpsc::UnboundedReceiver<CollectItem>> {
     pub fn new(channel: Channel) -> Self {
-        let client = ReporterClient::new(channel);
-        Self { client }
+        let (p, c) = mpsc::unbounded_channel();
+        Self::new_with_pc(channel, p, c)
     }
 
     pub async fn connect(
         address: impl TryInto<Endpoint, Error = transport::Error>,
     ) -> crate::Result<Self> {
-        let client = ReporterClient::connect(address.try_into()?).await?;
-        Ok(Self { client })
+        let endpoint = address.try_into()?;
+        let channel = endpoint.connect().await?;
+        Ok(Self::new(channel))
     }
 }
 
-#[async_trait]
-impl Reporter for GrpcReporter {
-    async fn collect(&mut self, segments: LinkedList<SegmentObject>) -> Result<(), Box<dyn Error>> {
-        let stream = stream::iter(segments);
-        self.client.collect(stream).await?;
+impl<P: CollectItemProduce, C: ColletcItemConsume> GrpcReporter<P, C> {
+    pub fn new_with_pc(channel: Channel, producer: P, consumer: C) -> Self {
+        Self {
+            inner: Arc::new(Inner {
+                trace_client: Mutex::new(TraceSegmentReportServiceClient::new(channel.clone())),
+                log_client: Mutex::new(LogReportServiceClient::new(channel)),
+                producer,
+                consumer: Mutex::new(Some(consumer)),
+                is_reporting: Default::default(),
+                is_closed: Default::default(),
+            }),
+            err_handle: Default::default(),
+        }
+    }
+
+    pub fn with_err_handle(
+        mut self,
+        handle: impl Fn(Box<dyn Error>) + Send + Sync + 'static,
+    ) -> Self {
+        self.err_handle = Arc::new(Some(Box::new(handle)));
+        self
+    }
+
+    /// Start to reporting, quit when shutdown_signal received.
+    ///
+    /// Accept a `shutdown_signal` argument as a graceful shutdown signal.
+    ///
+    /// # Panics
+    ///
+    /// Panic if call more than once.
+    pub async fn reporting(&self) -> Reporting<P, C> {
+        if self.inner.is_reporting.swap(true, Ordering::Relaxed) {
+            panic!("reporting already called");
+        }
+
+        Reporting {
+            rb: ReporterAndBuffer {
+                inner: Arc::clone(&self.inner),
+                trace_buffer: Default::default(),
+                log_buffer: Default::default(),
+                status_handle: None,
+            },
+            shutdown_signal: Box::pin(pending()),
+            consumer: self.inner.consumer.lock().await.take().unwrap(),
+        }
+    }
+}
+
+impl<P, C> Clone for GrpcReporter<P, C> {
+    fn clone(&self) -> Self {
+        Self {
+            inner: self.inner.clone(),
+            err_handle: self.err_handle.clone(),
+        }
+    }
+}
+
+impl<P: CollectItemProduce, C: ColletcItemConsume> Report for GrpcReporter<P, C> {
+    fn report(&self, item: CollectItem) {
+        if !self.inner.is_closed.load(Ordering::Relaxed) {
+            if let Err(e) = self.inner.producer.produce(item) {
+                if let Some(handle) = self.err_handle.as_deref() {
+                    handle(e);
+                }
+            }
+        }
+    }
+}
+
+struct ReporterAndBuffer<P, C> {
+    inner: Arc<Inner<P, C>>,
+    trace_buffer: LinkedList<SegmentObject>,
+    log_buffer: LinkedList<LogData>,
+    status_handle: Option<Box<dyn Fn(tonic::Status) + Send + 'static>>,
+}
+
+impl<P: CollectItemProduce, C: ColletcItemConsume> ReporterAndBuffer<P, C> {
+    async fn report(&mut self, item: CollectItem) {
+        // TODO Implement batch collect in future.
+        match item {
+            CollectItem::Trace(item) => {
+                self.trace_buffer.push_back(item);
+            }
+            CollectItem::Log(item) => {
+                self.log_buffer.push_back(item);
+            }
+        }
+
+        if !self.trace_buffer.is_empty() {
+            let buffer = take(&mut self.trace_buffer);
+            if let Err(e) = self
+                .inner
+                .trace_client
+                .lock()
+                .await
+                .collect(stream::iter(buffer))
+                .await
+            {
+                if let Some(status_handle) = &self.status_handle {
+                    status_handle(e);
+                }
+            }
+        }
+        if !self.log_buffer.is_empty() {
+            let buffer = take(&mut self.log_buffer);
+            if let Err(e) = self
+                .inner
+                .log_client
+                .lock()
+                .await
+                .collect(stream::iter(buffer))
+                .await
+            {
+                if let Some(status_handle) = &self.status_handle {
+                    status_handle(e);
+                }
+            }
+        }
+    }
+}
+
+/// Created by [GrpcReporter::reporting].
+pub struct Reporting<P, C> {
+    rb: ReporterAndBuffer<P, C>,
+    consumer: C,
+    shutdown_signal: Pin<Box<dyn Future<Output = ()> + Send + Sync + 'static>>,
+}
+
+impl<P: CollectItemProduce, C: ColletcItemConsume> Reporting<P, C> {
+    pub fn with_graceful_shutdown(
+        mut self,
+        shutdown_signal: impl Future<Output = ()> + Send + Sync + 'static,
+    ) -> Self {
+        self.shutdown_signal = Box::pin(shutdown_signal);
+        self
+    }
+
+    pub fn with_staus_handle(mut self, handle: impl Fn(tonic::Status) + Send + 'static) -> Self {
+        self.rb.status_handle = Some(Box::new(handle));
+        self
+    }
+
+    pub fn spawn(self) -> ReportingJoinHandle {
+        ReportingJoinHandle {
+            handle: tokio::spawn(self.start()),
+        }
+    }
+
+    pub async fn start(self) -> crate::Result<()> {
+        let (shutdown_tx, mut shutdown_rx) = mpsc::unbounded_channel();
+        let Reporting {
+            mut rb,
+            mut consumer,
+            shutdown_signal,
+        } = self;
+
+        let work_fut = async move {
+            loop {
+                select! {
+                    item = consumer.consume() => {
+                        match item {
+                            Ok(Some(item)) => {
+                                rb.report(item).await;
+                            }
+                            Ok(None) => break,
+                            Err(err) => return Err(crate::Error::Other(err)),
+                        }
+                    }
+                    _ =  shutdown_rx.recv() => break,
+                }
+            }
+
+            rb.inner.is_closed.store(true, Ordering::Relaxed);
+
+            // Flush.
+            loop {
+                match consumer.try_consume().await {
+                    Ok(Some(item)) => {
+                        rb.report(item).await;
+                    }
+                    Ok(None) => break,
+                    Err(err) => return Err(err.into()),
+                }
+            }
+
+            Ok::<_, crate::Error>(())
+        };
+
+        let shutdown_fut = async move {
+            shutdown_signal.await;
+            shutdown_tx
+                .send(())
+                .map_err(|e| crate::Error::Other(Box::new(e)))?;
+            Ok(())
+        };
+
+        try_join!(work_fut, shutdown_fut)?;
+
         Ok(())
     }
 }
+
+pub struct ReportingJoinHandle {
+    handle: JoinHandle<crate::Result<()>>,
+}
+
+impl Future for ReportingJoinHandle {
+    type Output = crate::Result<()>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        Pin::new(&mut self.handle).poll(cx).map(|r| r?)
+    }
+}
diff --git a/src/reporter/log.rs b/src/reporter/log.rs
deleted file mode 100644
index 2e0a98a..0000000
--- a/src/reporter/log.rs
+++ /dev/null
@@ -1,74 +0,0 @@
-// 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 super::Reporter;
-use crate::skywalking_proto::v3::SegmentObject;
-use std::{collections::LinkedList, error::Error};
-use tonic::async_trait;
-
-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>) -> Result<(), Box<dyn Error>> {
-        for segment in segments {
-            match self.used {
-                Used::Println => println!("{} segment={:?}", self.tip, segment),
-                Used::Tracing => tracing::info!(?segment, "{}", self.tip),
-            }
-        }
-        Ok(())
-    }
-}
diff --git a/src/reporter/mod.rs b/src/reporter/mod.rs
index bb52a16..534647d 100644
--- a/src/reporter/mod.rs
+++ b/src/reporter/mod.rs
@@ -15,25 +15,39 @@
 //
 
 pub mod grpc;
-pub mod log;
+pub mod once_cell;
+pub mod print;
 
-use crate::skywalking_proto::v3::SegmentObject;
-use std::{collections::LinkedList, error::Error, result::Result};
-use tonic::async_trait;
+use crate::skywalking_proto::v3::{LogData, SegmentObject};
+use std::{ops::Deref, sync::Arc};
 
-pub(crate) type DynReporter = dyn Reporter + Send + Sync + 'static;
+#[derive(Debug)]
+#[non_exhaustive]
+pub enum CollectItem {
+    Trace(SegmentObject),
+    Log(LogData),
+}
+
+pub(crate) type DynReport = dyn Report + Send + Sync + 'static;
+
+/// Report provide non-blocking report method for trace, log and metric object.
+pub trait Report {
+    fn report(&self, item: CollectItem);
+}
 
-#[async_trait]
-pub trait Reporter {
-    async fn collect(&mut self, segments: LinkedList<SegmentObject>) -> Result<(), Box<dyn Error>>;
+/// Noop reporter.
+impl Report for () {
+    fn report(&self, _item: CollectItem) {}
+}
+
+impl<T: Report> Report for Box<T> {
+    fn report(&self, item: CollectItem) {
+        Report::report(self.deref(), item)
+    }
 }
 
-#[async_trait]
-impl Reporter for () {
-    async fn collect(
-        &mut self,
-        _segments: LinkedList<SegmentObject>,
-    ) -> Result<(), Box<dyn Error>> {
-        Ok(())
+impl<T: Report> Report for Arc<T> {
+    fn report(&self, item: CollectItem) {
+        Report::report(self.deref(), item)
     }
 }
diff --git a/src/reporter/once_cell.rs b/src/reporter/once_cell.rs
new file mode 100644
index 0000000..acd8d73
--- /dev/null
+++ b/src/reporter/once_cell.rs
@@ -0,0 +1,38 @@
+// 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::reporter::{CollectItem, Report};
+use tokio::sync::OnceCell;
+
+#[derive(Clone, Default)]
+pub struct OnceCellReporter<R: Report> {
+    report: OnceCell<R>,
+}
+
+impl<R: Report> OnceCellReporter<R> {
+    pub fn set(&self, report: R) -> Option<()> {
+        self.report.set(report).ok()
+    }
+}
+
+impl<R: Report> Report for OnceCellReporter<R> {
+    fn report(&self, item: CollectItem) {
+        self.report
+            .get()
+            .expect("Report hasn't setted")
+            .report(item)
+    }
+}
diff --git a/src/reporter/print.rs b/src/reporter/print.rs
new file mode 100644
index 0000000..6b355e5
--- /dev/null
+++ b/src/reporter/print.rs
@@ -0,0 +1,55 @@
+// 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::reporter::{CollectItem, Report};
+
+#[derive(Default, Clone)]
+pub struct PrintReporter {
+    use_stderr: bool,
+}
+
+impl PrintReporter {
+    #[inline]
+    pub fn new() -> Self {
+        Default::default()
+    }
+
+    pub fn use_stderr(mut self, use_stderr: bool) -> Self {
+        self.use_stderr = use_stderr;
+        self
+    }
+}
+
+impl Report for PrintReporter {
+    fn report(&self, items: CollectItem) {
+        match items {
+            CollectItem::Trace(segment) => {
+                if self.use_stderr {
+                    eprintln!("trace segment={:?}", segment);
+                } else {
+                    println!("trace segment={:?}", segment);
+                }
+            }
+            CollectItem::Log(data) => {
+                if self.use_stderr {
+                    eprintln!("log data={:?}", data);
+                } else {
+                    println!("log data={:?}", data);
+                }
+            }
+        }
+    }
+}
diff --git a/src/skywalking_proto/v3/mod.rs b/src/skywalking_proto/v3/mod.rs
index 7a2a268..8623eaa 100644
--- a/src/skywalking_proto/v3/mod.rs
+++ b/src/skywalking_proto/v3/mod.rs
@@ -13,7 +13,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 //
-use crate::trace::system_time::{fetch_time, TimePeriod};
+use crate::common::system_time::{fetch_time, TimePeriod};
 
 tonic::include_proto!("skywalking.v3");
 
diff --git a/src/trace/mod.rs b/src/trace/mod.rs
index 28d7ace..c194875 100644
--- a/src/trace/mod.rs
+++ b/src/trace/mod.rs
@@ -16,6 +16,5 @@
 
 pub mod propagation;
 pub mod span;
-pub(crate) mod system_time;
 pub mod trace_context;
 pub mod tracer;
diff --git a/src/trace/span.rs b/src/trace/span.rs
index c9c6294..ce49c6b 100644
--- a/src/trace/span.rs
+++ b/src/trace/span.rs
@@ -14,13 +14,11 @@
 // limitations under the License.
 //
 
-use super::{
-    system_time::{fetch_time, TimePeriod},
-    trace_context::SpanStack,
-};
 use crate::{
+    common::system_time::{fetch_time, TimePeriod},
     error::LOCK_MSG,
     skywalking_proto::v3::{SpanLayer, SpanObject, SpanType},
+    trace::trace_context::SpanStack,
 };
 use std::{
     fmt::Formatter,
diff --git a/src/trace/trace_context.rs b/src/trace/trace_context.rs
index 6f23adb..3710199 100644
--- a/src/trace/trace_context.rs
+++ b/src/trace/trace_context.rs
@@ -14,18 +14,20 @@
 // limitations under the License.
 //
 
-use super::{
-    span::Span,
-    system_time::{fetch_time, TimePeriod},
-    tracer::{Tracer, WeakTracer},
-};
 use crate::{
-    common::random_generator::RandomGenerator,
+    common::{
+        random_generator::RandomGenerator,
+        system_time::{fetch_time, TimePeriod},
+    },
     error::LOCK_MSG,
     skywalking_proto::v3::{
         RefType, SegmentObject, SegmentReference, SpanLayer, SpanObject, SpanType,
     },
-    trace::propagation::context::PropagationContext,
+    trace::{
+        propagation::context::PropagationContext,
+        span::Span,
+        tracer::{Tracer, WeakTracer},
+    },
 };
 use std::{
     fmt::Formatter,
@@ -332,7 +334,7 @@ impl TracingContext {
     /// This conversion should be done before sending segments into OAP.
     ///
     /// Notice: The spans will taked, so this method shouldn't be called twice.
-    pub(crate) fn convert_segment_object(&mut self) -> SegmentObject {
+    pub(crate) fn convert_to_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();
diff --git a/src/trace/tracer.rs b/src/trace/tracer.rs
index 9ea9696..963d270 100644
--- a/src/trace/tracer.rs
+++ b/src/trace/tracer.rs
@@ -15,29 +15,11 @@
 //
 
 use crate::{
-    reporter::{DynReporter, Reporter},
-    skywalking_proto::v3::SegmentObject,
+    reporter::{CollectItem, DynReport, Report},
     trace::trace_context::TracingContext,
 };
-use std::{
-    collections::LinkedList,
-    error::Error,
-    future::Future,
-    pin::Pin,
-    sync::{
-        atomic::{AtomicBool, Ordering},
-        Arc, Weak,
-    },
-    task::{Context, Poll},
-};
-use tokio::{
-    sync::{
-        mpsc::{self},
-        Mutex, OnceCell,
-    },
-    task::JoinHandle,
-};
-use tonic::async_trait;
+use std::sync::{Arc, Weak};
+use tokio::sync::OnceCell;
 
 static GLOBAL_TRACER: OnceCell<Tracer> = OnceCell::const_new();
 
@@ -58,74 +40,10 @@ pub fn create_trace_context() -> TracingContext {
     global_tracer().create_trace_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<Output = ()> + Send + Sync + 'static) -> Reporting {
-    global_tracer().reporting(shutdown_signal)
-}
-
-pub trait SegmentSender: Send + Sync + 'static {
-    fn send(&self, segment: SegmentObject) -> Result<(), Box<dyn Error>>;
-}
-
-impl SegmentSender for () {
-    fn send(&self, _segment: SegmentObject) -> Result<(), Box<dyn Error>> {
-        Ok(())
-    }
-}
-
-impl SegmentSender for mpsc::UnboundedSender<SegmentObject> {
-    fn send(&self, segment: SegmentObject) -> Result<(), Box<dyn Error>> {
-        Ok(self.send(segment)?)
-    }
-}
-
-#[async_trait]
-pub trait SegmentReceiver: Send + Sync + 'static {
-    async fn recv(&self) -> Result<Option<SegmentObject>, Box<dyn Error + Send>>;
-
-    async fn try_recv(&self) -> Result<Option<SegmentObject>, Box<dyn Error + Send>>;
-}
-
-#[async_trait]
-impl SegmentReceiver for () {
-    async fn recv(&self) -> Result<Option<SegmentObject>, Box<dyn Error + Send>> {
-        Ok(None)
-    }
-
-    async fn try_recv(&self) -> Result<Option<SegmentObject>, Box<dyn Error + Send>> {
-        Ok(None)
-    }
-}
-
-#[async_trait]
-impl SegmentReceiver for Mutex<mpsc::UnboundedReceiver<SegmentObject>> {
-    async fn recv(&self) -> Result<Option<SegmentObject>, Box<dyn Error + Send>> {
-        Ok(self.lock().await.recv().await)
-    }
-
-    async fn try_recv(&self) -> Result<Option<SegmentObject>, Box<dyn Error + Send>> {
-        use mpsc::error::TryRecvError;
-
-        match self.lock().await.try_recv() {
-            Ok(segment) => Ok(Some(segment)),
-            Err(e) => match e {
-                TryRecvError::Empty => Ok(None),
-                TryRecvError::Disconnected => Err(Box::new(e)),
-            },
-        }
-    }
-}
-
 struct Inner {
     service_name: String,
     instance_name: String,
-    segment_sender: Box<dyn SegmentSender>,
-    segment_receiver: Box<dyn SegmentReceiver>,
-    reporter: Mutex<Box<DynReporter>>,
-    is_reporting: AtomicBool,
-    is_closed: AtomicBool,
+    reporter: Box<DynReport>,
 }
 
 /// Skywalking tracer.
@@ -139,33 +57,13 @@ impl Tracer {
     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::new_with_channel(
-            service_name,
-            instance_name,
-            reporter,
-            (segment_sender, Mutex::new(segment_receiver)),
-        )
-    }
-
-    /// New with service info, reporter, and custom channel.
-    pub fn new_with_channel(
-        service_name: impl ToString,
-        instance_name: impl ToString,
-        reporter: impl Reporter + Send + Sync + 'static,
-        channel: (impl SegmentSender, impl SegmentReceiver),
+        reporter: impl Report + Send + Sync + 'static,
     ) -> Self {
         Self {
             inner: Arc::new(Inner {
                 service_name: service_name.to_string(),
                 instance_name: instance_name.to_string(),
-                segment_sender: Box::new(channel.0),
-                segment_receiver: Box::new(channel.1),
-                reporter: Mutex::new(Box::new(reporter)),
-                is_reporting: Default::default(),
-                is_closed: Default::default(),
+                reporter: Box::new(reporter),
             }),
         }
     }
@@ -178,15 +76,6 @@ impl Tracer {
         &self.inner.instance_name
     }
 
-    /// Set the reporter, only valid if [`Tracer::reporting`] not started.
-    pub fn set_reporter(&self, reporter: impl Reporter + Send + Sync + 'static) {
-        if !self.inner.is_reporting.load(Ordering::Relaxed) {
-            if let Ok(mut lock) = self.inner.reporter.try_lock() {
-                *lock = Box::new(reporter);
-            }
-        }
-    }
-
     /// Create trace conetxt.
     pub fn create_trace_context(&self) -> TracingContext {
         TracingContext::new(
@@ -198,98 +87,10 @@ impl Tracer {
 
     /// Finalize the trace context.
     pub(crate) fn finalize_context(&self, context: &mut TracingContext) {
-        if self.inner.is_closed.load(Ordering::Relaxed) {
-            tracing::warn!("tracer closed");
-            return;
-        }
-
-        let segment_object = context.convert_segment_object();
-        if let Err(err) = self.inner.segment_sender.send(segment_object) {
-            tracing::error!(?err, "send segment object failed");
-        }
-    }
-
-    /// Start to reporting, quit when shutdown_signal received.
-    ///
-    /// Accept a `shutdown_signal` argument as a graceful shutdown signal.
-    ///
-    /// # Panics
-    ///
-    /// Panic if call more than once.
-    pub fn reporting(
-        &self,
-        shutdown_signal: impl Future<Output = ()> + Send + Sync + 'static,
-    ) -> Reporting {
-        if self.inner.is_reporting.swap(true, Ordering::Relaxed) {
-            panic!("reporting already called");
-        }
-
-        Reporting {
-            handle: tokio::spawn(self.clone().do_reporting(shutdown_signal)),
-        }
-    }
-
-    async fn do_reporting(
-        self,
-        shutdown_signal: impl Future<Output = ()> + Send + Sync + 'static,
-    ) -> crate::Result<()> {
-        let (shutdown_tx, mut shutdown_rx) = mpsc::unbounded_channel();
-
-        let handle = tokio::spawn(async move {
-            loop {
-                tokio::select! {
-                    segment = self.inner.segment_receiver.recv() => {
-                        match segment {
-                            Ok(Some(segment)) => {
-                                // TODO Implement batch collect in future.
-                                let mut segments = LinkedList::new();
-                                segments.push_back(segment);
-                                Self::report_segment_object(&self.inner.reporter, segments).await;
-                            }
-                            Ok(None) => break,
-                            Err(err) => return Err(err.into()),
-                        }
-                    }
-                    _ =  shutdown_rx.recv() => break,
-                }
-            }
-
-            self.inner.is_closed.store(true, Ordering::Relaxed);
-
-            // Flush.
-            let mut segments = LinkedList::new();
-            loop {
-                match self.inner.segment_receiver.try_recv().await {
-                    Ok(Some(segment)) => {
-                        segments.push_back(segment);
-                    }
-                    Ok(None) => break,
-                    Err(err) => return Err(err.into()),
-                }
-            }
-            Self::report_segment_object(&self.inner.reporter, segments).await;
-
-            Ok::<_, crate::Error>(())
-        });
-
-        shutdown_signal.await;
-
-        if shutdown_tx.send(()).is_err() {
-            tracing::error!("shutdown signal send failed");
-        }
-
-        handle.await??;
-
-        Ok(())
-    }
-
-    async fn report_segment_object(
-        reporter: &Mutex<Box<DynReporter>>,
-        segments: LinkedList<SegmentObject>,
-    ) {
-        if let Err(err) = reporter.lock().await.collect(segments).await {
-            tracing::error!(?err, "collect failed");
-        }
+        let segment_object = context.convert_to_segment_object();
+        self.inner
+            .reporter
+            .report(CollectItem::Trace(segment_object));
     }
 
     fn downgrade(&self) -> WeakTracer {
@@ -310,31 +111,11 @@ impl WeakTracer {
     }
 }
 
-/// Created by [Tracer::reporting].
-pub struct Reporting {
-    handle: JoinHandle<crate::Result<()>>,
-}
-
-impl Future for Reporting {
-    type Output = crate::Result<()>;
-
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        Pin::new(&mut self.handle).poll(cx).map(|r| r?)
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
-    use std::future;
 
     trait AssertSend: Send {}
 
     impl AssertSend for Tracer {}
-
-    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
-    async fn custom_channel() {
-        let tracer = Tracer::new_with_channel("service_name", "instance_name", (), ((), ()));
-        tracer.reporting(future::ready(())).await.unwrap();
-    }
 }
diff --git a/tests/logging.rs b/tests/logging.rs
new file mode 100644
index 0000000..e909bf6
--- /dev/null
+++ b/tests/logging.rs
@@ -0,0 +1,169 @@
+// 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 skywalking::{
+    logging::{
+        logger::Logger,
+        record::{LogRecord, RecordType},
+    },
+    reporter::{CollectItem, Report},
+    skywalking_proto::v3::{
+        log_data_body::Content, JsonLog, KeyStringValuePair, LogData, LogDataBody, LogTags,
+        TextLog, TraceContext,
+    },
+    trace::tracer::Tracer,
+};
+use std::{
+    collections::LinkedList,
+    sync::{Arc, Mutex},
+};
+
+#[test]
+fn log() {
+    let reporter = Arc::new(MockReporter::default());
+    let logger = Logger::new("service_name", "instance_name", reporter.clone());
+
+    {
+        logger.log(LogRecord::new());
+        assert_eq!(
+            reporter.pop(),
+            LogData {
+                timestamp: 10,
+                service: "service_name".to_owned(),
+                service_instance: "instance_name".to_owned(),
+                body: Some(LogDataBody {
+                    r#type: "".to_owned(),
+                    content: Some(Content::Text(TextLog {
+                        text: "".to_owned()
+                    }))
+                }),
+                ..Default::default()
+            }
+        );
+    }
+
+    {
+        logger.log(LogRecord::new().ignore_time());
+        assert_eq!(
+            reporter.pop(),
+            LogData {
+                timestamp: 0,
+                service: "service_name".to_owned(),
+                service_instance: "instance_name".to_owned(),
+                body: Some(LogDataBody {
+                    r#type: "".to_owned(),
+                    content: Some(Content::Text(TextLog {
+                        text: "".to_owned()
+                    }))
+                }),
+                ..Default::default()
+            }
+        );
+    }
+
+    {
+        logger.log(
+            LogRecord::new()
+                .endpoint("endpoint")
+                .add_tag("foo", "foo")
+                .add_tags([("bar", "bar")])
+                .record_type(RecordType::Json)
+                .content(r#"{"content": "something to log"}"#),
+        );
+        assert_eq!(
+            reporter.pop(),
+            LogData {
+                timestamp: 10,
+                service: "service_name".to_owned(),
+                service_instance: "instance_name".to_owned(),
+                endpoint: "endpoint".to_owned(),
+                tags: Some(LogTags {
+                    data: vec![
+                        KeyStringValuePair {
+                            key: "foo".to_owned(),
+                            value: "foo".to_owned()
+                        },
+                        KeyStringValuePair {
+                            key: "bar".to_owned(),
+                            value: "bar".to_owned()
+                        },
+                    ],
+                }),
+                body: Some(LogDataBody {
+                    r#type: "".to_owned(),
+                    content: Some(Content::Json(JsonLog {
+                        json: r#"{"content": "something to log"}"#.to_owned()
+                    }))
+                }),
+                ..Default::default()
+            }
+        );
+    }
+}
+
+#[test]
+fn integrate_trace() {
+    let reporter = Arc::new(MockReporter::default());
+    let tracer = Tracer::new("service_name", "instance_name", reporter.clone());
+    let logger = Logger::new("service_name", "instance_name", reporter.clone());
+
+    let mut ctx = tracer.create_trace_context();
+    let span = ctx.create_entry_span("operation_name");
+
+    logger.log(LogRecord::new().with_tracing_context(&ctx).with_span(&span));
+    assert_eq!(
+        reporter.pop(),
+        LogData {
+            timestamp: 10,
+            service: "service_name".to_owned(),
+            service_instance: "instance_name".to_owned(),
+            trace_context: Some(TraceContext {
+                trace_id: ctx.trace_id().to_owned(),
+                trace_segment_id: ctx.trace_segment_id().to_owned(),
+                span_id: span.span_id()
+            }),
+            body: Some(LogDataBody {
+                r#type: "".to_owned(),
+                content: Some(Content::Text(TextLog {
+                    text: "".to_owned()
+                }))
+            }),
+            ..Default::default()
+        }
+    );
+}
+
+#[derive(Default, Clone)]
+struct MockReporter {
+    items: Arc<Mutex<LinkedList<LogData>>>,
+}
+
+impl MockReporter {
+    fn pop(&self) -> LogData {
+        self.items.try_lock().unwrap().pop_back().unwrap()
+    }
+}
+
+impl Report for MockReporter {
+    fn report(&self, item: CollectItem) {
+        match item {
+            CollectItem::Log(data) => {
+                self.items.try_lock().unwrap().push_back(data);
+            }
+            _ => {}
+        }
+    }
+}
diff --git a/tests/propagation.rs b/tests/propagation.rs
index 1b95812..684b60e 100644
--- a/tests/propagation.rs
+++ b/tests/propagation.rs
@@ -14,18 +14,13 @@
 // limitations under the License.
 //
 
-#![allow(unused_imports)]
 use skywalking::{
-    reporter::log::LogReporter,
+    reporter::print::PrintReporter,
     trace::{
-        propagation::{
-            context::PropagationContext, decoder::decode_propagation, encoder::encode_propagation,
-        },
-        trace_context::TracingContext,
+        propagation::{decoder::decode_propagation, encoder::encode_propagation},
         tracer::Tracer,
     },
 };
-use std::sync::Arc;
 
 #[test]
 fn basic() {
@@ -68,7 +63,7 @@ fn invalid_sample() {
 
 #[test]
 fn basic_encode() {
-    let tracer = Tracer::new("mesh", "instance", LogReporter::new());
+    let tracer = Tracer::new("mesh", "instance", PrintReporter::new());
     let tc = tracer.create_trace_context();
     let res = encode_propagation(&tc, "/api/v1/health", "example.com:8080");
     let res2 = decode_propagation(&res).unwrap();
diff --git a/tests/trace_context.rs b/tests/trace_context.rs
index 7ca80d2..efae1e8 100644
--- a/tests/trace_context.rs
+++ b/tests/trace_context.rs
@@ -16,7 +16,7 @@
 
 use prost::Message;
 use skywalking::{
-    reporter::{log::LogReporter, Reporter},
+    reporter::{print::PrintReporter, CollectItem, Report},
     skywalking_proto::v3::{
         KeyStringValuePair, Log, RefType, SegmentObject, SegmentReference, SpanLayer, SpanObject,
         SpanType,
@@ -28,8 +28,6 @@ use skywalking::{
 };
 use std::{
     collections::LinkedList,
-    error::Error,
-    future,
     sync::{Arc, Mutex},
     thread,
 };
@@ -182,7 +180,7 @@ async fn create_span() {
 #[test]
 #[should_panic]
 fn create_local_span_failed() {
-    let tracer = Tracer::new("service", "instance", LogReporter::new());
+    let tracer = Tracer::new("service", "instance", PrintReporter::new());
     let mut context = tracer.create_trace_context();
     let _span1 = context.create_local_span("op1");
 }
@@ -190,7 +188,7 @@ fn create_local_span_failed() {
 #[test]
 #[should_panic]
 fn create_exit_span_failed() {
-    let tracer = Tracer::new("service", "instance", LogReporter::new());
+    let tracer = Tracer::new("service", "instance", PrintReporter::new());
     let mut context = tracer.create_trace_context();
     let _span1 = context.create_exit_span("op1", "example.com/test");
 }
@@ -220,7 +218,7 @@ async fn create_span_from_context() {
 
 #[test]
 fn crossprocess_test() {
-    let tracer = Tracer::new("service", "instance", LogReporter::new());
+    let tracer = Tracer::new("service", "instance", PrintReporter::new());
     let mut context1 = tracer.create_trace_context();
     assert_eq!(context1.service(), "service");
     assert_eq!(context1.service_instance(), "instance");
@@ -233,7 +231,7 @@ fn crossprocess_test() {
             let enc_prop = encode_propagation(&context1, "endpoint", "address");
             let dec_prop = decode_propagation(&enc_prop).unwrap();
 
-            let tracer = Tracer::new("service2", "instance2", LogReporter::new());
+            let tracer = Tracer::new("service2", "instance2", PrintReporter::new());
             let mut context2 = tracer.create_trace_context();
 
             let span3 = context2.create_entry_span_with_propagation("op2", &dec_prop);
@@ -323,20 +321,18 @@ impl MockReporter {
         f2: impl FnOnce(&LinkedList<SegmentObject>),
     ) {
         let reporter = MockReporter::default();
-
-        let tracer = f1(reporter.clone());
-
-        tracer.reporting(future::ready(())).await.unwrap();
-
+        f1(reporter.clone());
         let segments = reporter.segments.try_lock().unwrap();
         f2(&*segments);
     }
 }
 
-#[tonic::async_trait]
-impl Reporter for MockReporter {
-    async fn collect(&mut self, segments: LinkedList<SegmentObject>) -> Result<(), Box<dyn Error>> {
-        self.segments.try_lock().unwrap().extend(segments);
-        Ok(())
+impl Report for MockReporter {
+    fn report(&self, item: CollectItem) {
+        let segment = match item {
+            CollectItem::Trace(segment) => segment,
+            _ => unreachable!(),
+        };
+        self.segments.try_lock().unwrap().push_back(segment);
     }
 }
