telemetry/
telemetry.rs

1// Copyright 2024 ScopeDB, Inc.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15mod common;
16
17use std::time::Duration;
18
19use scopedb_client::AppendCircuitBreakerOptions;
20use scopedb_client::AppendDeliveryOutcome;
21use scopedb_client::AppendFailurePolicy;
22use scopedb_client::AppendStream;
23use uuid::Uuid;
24
25fn track(stream: &AppendStream, request: u64) -> bool {
26    match stream.try_send(&serde_json::json!({
27        "event_id": Uuid::now_v7(),
28        "occurred_at": jiff::Timestamp::now(),
29        "name": "request.completed",
30        "attributes": {"request": request, "status": 200},
31    })) {
32        // This means local admission only, not a remote commit.
33        Ok(()) => true,
34        Err(error) => {
35            // Route this diagnostic to another sink, never back into `stream`.
36            eprintln!("telemetry row dropped locally: {error}");
37            false
38        }
39    }
40}
41
42#[tokio::main(flavor = "current_thread")]
43async fn main() -> Result<(), Box<dyn std::error::Error>> {
44    let client = common::client()?;
45    let table = common::write_table(&client)?;
46
47    // Continue mode is explicitly best effort: failed or ambiguous batches are
48    // accounted for, released, and do not stop later telemetry batches.
49    let telemetry = table
50        .append_stream()
51        .failure_policy(AppendFailurePolicy::Continue)
52        .circuit_breaker(Some(AppendCircuitBreakerOptions::default()))
53        .target_batch_bytes(1024 * 1024)
54        .max_batch_rows(10_000)
55        .flush_interval(Duration::from_secs(1))
56        .max_buffered_bytes(32 * 1024 * 1024)
57        // This limit applies to each stream; include every application replica.
58        .max_concurrent_batches(1)
59        .attempt_timeout(Duration::from_secs(10))
60        .on_batch_failure(|event| {
61            // A synchronous observer; use a separate diagnostics sink.
62            eprintln!("telemetry append failure: {event:?}");
63        })
64        .build()?;
65
66    let mut dropped_locally = 0_u64;
67    for request in 1..=100 {
68        if !track(&telemetry, request) {
69            dropped_locally += 1;
70        }
71    }
72    println!("locally dropped: {dropped_locally}");
73
74    // In a service, stop and join producer tasks before this settlement barrier.
75    let report = telemetry.shutdown().await?;
76    if report.outcome != AppendDeliveryOutcome::Ok {
77        eprintln!("telemetry loss or ambiguity: {report:?}");
78    }
79    println!("final stream stats: {:?}", telemetry.stats());
80
81    Ok(())
82}
telemetry/telemetry.rs

telemetry/
telemetry.rs
