Apache SkyWalking Rust Agent

SkyWalking Rust Agent 为 Rust 应用和库提供可观察性功能，包括跟踪、指标、分布式系统的拓扑图和警报。它使用 SkyWalking 原生格式和核心概念，以保持最佳兼容性和性能。

概念

所有概念均来自官方 SkyWalking 定义。

跟踪

Span

Span 是分布式跟踪系统中的一个重要且常见概念。从 Google Dapper 论文中学习 Span。为了更好的性能，我们将 span 扩展为 3 种类型。

1. EntrySpan EntrySpan 代表服务提供者，也是服务器端端点。作为一个 APM 系统，我们针对的是应用服务器。所以几乎所有的服务和 MQ 消费者都是 EntrySpan。
2. LocalSpan LocalSpan 代表一个普通的 Java 方法，它不与远程服务相关，也不是 MQ 生产者/消费者或服务（例如 HTTP 服务）提供者/消费者。
3. ExitSpan ExitSpan 代表服务或 MQ 生产者的客户端，早期 SkyWalking 中称为 LeafSpan。例如，通过 JDBC 访问数据库、读取 Redis/Memcached 被归类为 ExitSpan。

标签和日志是 span 的类似属性。

• 标签是一个键值对，用于指示具有字符串值的属性。
• 日志比标签更复杂，包含一个时间戳和多个键值对。日志表示一个事件，通常是错误发生。

跟踪上下文

跟踪上下文是跟踪过程的上下文。Span 应该仅通过上下文创建，并在 span 完成后存档到上下文。

日志记录

LogRecord

LogRecord 是 LogData 的简单构建器，它是 Skywalking 的日志格式。

指标

Meter

• Counter API 代表一个单调递增的单个计数器，它自动收集数据并向后端报告。
• Gauge API 代表一个单个数值。
• 直方图 API 表示带有自定义桶的样本观察结果的摘要。

管理

报告实例的额外信息。

报告实例属性

方法 insert_os_info 在 skywalking::management::instance::Properties 中用于插入预定义的操作系统信息。此外，您还可以使用 insert、update 和 remove 来自定义实例信息。

预定义的操作系统信息

保持存活

在后台分析中保持实例存活。仅当不需要报告跟踪和指标时才建议单独进行 keepAlive 报告。否则，它将是重复的。

示例

use skywalking::{
    logging::{logger::Logger, record::{LogRecord, RecordType}},
    reporter::grpc::GrpcReporter,
    trace::tracer::Tracer,
    metrics::{meter::Counter, metricer::Metricer},
};
use std::error::Error;
use tokio::signal;

async fn handle_request(tracer: Tracer, logger: Logger) {
    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.
        // 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");

            // 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.
        }

        // Auto close span when dropped.
    }

    // Auto report ctx when dropped.
}

async fn handle_metric(mut metricer: Metricer) {
    let counter = metricer.register(
        Counter::new("instance_trace_count")
            .add_label("region", "us-west")
            .add_label("az", "az-1"),
    );

    metricer.boot().await;

    counter.increment(10.);
}

#[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?;
    // Optional authentication, based on backend setting.
    let reporter = reporter.with_authentication("<TOKEN>");

    // 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");
        })
        .spawn();

    let tracer = Tracer::new("service", "instance", reporter.clone());
    let logger = Logger::new("service", "instance", reporter.clone());
    let metricer = Metricer::new("service", "instance", reporter);

    handle_metric(metricer).await;

    handle_request(tracer, logger).await;

    handle.await?;

    Ok(())
}

高级 API

异步 Span API

Span::prepare_for_async 设计用于异步用例。当需要将 Span 的标签、日志和属性（包括结束时间）设置在另一个线程或协程中时。

TracingContext::wait 等待所有 AsyncSpan 完成。

use skywalking::{
    trace::tracer::Tracer,
    trace::span::HandleSpanObject,
};

async fn handle(tracer: Tracer) {
    let mut ctx = tracer.create_trace_context();

    {
        let span = ctx.create_entry_span("op1");

        // Create AsyncSpan and drop span.
        // Internally, span will occupy the position of finalized span stack.
        let mut async_span = span.prepare_for_async();

        // Start async route, catch async_span with `move` keyword.
        tokio::spawn(async move {

            async_span.add_tag("foo", "bar");

            // Something...

            // async_span will drop here, submit modifications to finalized spans stack.
        });
    }

    // Wait for all `AsyncSpan` finished.
    ctx.wait();
}

高级报告器

高级报告提供了一种将代理收集的数据提交到后端的替代方法。

kafka 报告器

Kafka 报告器插件支持将跟踪、指标、日志、实例属性报告到 Kafka 集群。

确保已启用功能 kafka-reporter。

#[cfg(feature = "kafka-reporter")]
mod example {
    use skywalking::reporter::Report;
    use skywalking::reporter::kafka::{KafkaReportBuilder, KafkaReporter, RDKafkaClientConfig};

    async fn do_something(reporter: &impl Report) {
        // ....
    }

    async fn foo() {
        let mut client_config = RDKafkaClientConfig::new();
        client_config
            .set("bootstrap.servers", "broker:9092")
            .set("message.timeout.ms", "6000");

        let (reporter, reporting) = KafkaReportBuilder::new(client_config).build().await.unwrap();
        let handle = reporting.spawn();

        do_something(&reporter);

        handle.await.unwrap();
    }
}

如何编译？

如果您有 skywalking-(VERSION).crate，您可以使用以下方式解包

tar -xvzf skywalking-(VERSION).crate