chore: instrument

Author: ymgyt

crates/synd_api/src/args.rs

@@ -1,6 +1,6 @@
 use std::path::PathBuf;
 
-use clap::Parser;
+use clap::{ArgAction, Parser};
 
 #[derive(Parser, Debug)]
 #[command(version, propagate_version = true, disable_help_subcommand = true)]
@@ -9,6 +9,8 @@ pub struct Args {
     pub kvsd: KvsdOptions,
     #[command(flatten)]
     pub tls: TlsOptions,
+    #[command(flatten)]
+    pub o11y: ObservabilityOptions,
 }
 
 #[derive(clap::Args, Debug)]
@@ -35,6 +37,26 @@ pub struct TlsOptions {
     pub private_key: PathBuf,
 }
 
+#[derive(clap::Args, Debug)]
+#[command(next_help_heading = "Observability options")]
+pub struct ObservabilityOptions {
+    /// Show code location(file, line number) in logs
+    #[arg(long, env = "SYND_LOG_SHOW_LOCATION", default_value_t = false, action = ArgAction::Set )]
+    pub show_code_location: bool,
+
+    /// Show event target(module in default) in logs
+    #[arg(long, env = "SYND_LOG_SHOW_TARGET", default_value_t = true, action = ArgAction::Set)]
+    pub show_target: bool,
+
+    /// Opentelemetry otlp exporter endpoint
+    #[arg(long, env = "OTEL_EXPORTER_OTLP_ENDPOINT")]
+    pub otlp_endpoint: Option<String>,
+
+    /// Opentelemetry trace sampler ratio
+    #[arg(long, env = "OTEL_TRACES_SAMPLER_ARG", default_value_t = 1.0)]
+    pub trace_sampler_ratio: f64,
+}
+
 #[must_use]
 pub fn parse() -> Args {
     Args::parse()

crates/synd_api/src/client/github/client.rs

@@ -37,6 +37,7 @@ impl GithubClient {
         }
     }
 
+    #[tracing::instrument(name = "github::authenticate", skip_all)]
     pub async fn authenticate(&self, access_token: &str) -> anyhow::Result<String> {
         let variables = query::authenticate::Variables {};
         let request = query::Authenticate::build_query(variables);

crates/synd_api/src/gql/mod.rs

@@ -2,7 +2,7 @@ mod query;
 pub use query::Query;
 
 mod mutation;
-use async_graphql::{extensions::Tracing, EmptySubscription, Schema, SchemaBuilder};
+use async_graphql::{EmptySubscription, Schema, SchemaBuilder};
 pub use mutation::Mutation;
 
 use crate::{principal::Principal, usecase};
@@ -45,7 +45,9 @@ pub fn schema_builder() -> SchemaBuilder<Query, Mutation, EmptySubscription> {
     #[cfg(not(feature = "introspection"))]
     let schema = Schema::build(Query, Mutation, EmptySubscription).disable_introspection();
 
-    schema.extension(Tracing)
+    // disabled
+    // schema.extension(Tracing)
+    schema
 }
 
 impl<'a> usecase::Context for &async_graphql::Context<'a> {

crates/synd_api/src/main.rs

@@ -5,15 +5,15 @@ use synd_o11y::{
 use tracing::{error, info};
 
 use synd_api::{
-    args::{self, Args},
+    args::{self, Args, ObservabilityOptions},
     config,
     dependency::Dependency,
     repository::kvsd::ConnectKvsdFailed,
     serve::listen_and_serve,
     shutdown::Shutdown,
 };
 
-fn init_tracing() -> Option<OpenTelemetryGuard> {
+fn init_tracing(options: &ObservabilityOptions) -> Option<OpenTelemetryGuard> {
     use synd_o11y::{
         opentelemetry::init_propagation,
         tracing_subscriber::{audit, otel_log, otel_trace},
@@ -27,21 +27,19 @@ fn init_tracing() -> Option<OpenTelemetryGuard> {
         use supports_color::Stream;
         supports_color::on(Stream::Stdout).is_some()
     };
-    let show_src = true;
-    let show_target = !show_src;
+    let show_src = options.show_code_location;
+    let show_target = options.show_target;
 
     let (opentelemetry_layers, guard) = {
-        match std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").ok() {
-            None => (None, None),
-            Some(endpoint) if endpoint.is_empty() => (None, None),
+        match options.otlp_endpoint.as_deref() {
+            None | Some("") => (None, None),
             Some(endpoint) => {
                 let resource = synd_o11y::opentelemetry::resource(config::NAME, config::VERSION);
 
-                tracing::info!(endpoint, ?resource, "Export opentelemetry signals");
-
-                let trace_layer = otel_trace::layer(resource.clone());
+                let trace_layer =
+                    otel_trace::layer(endpoint, resource.clone(), options.trace_sampler_ratio);
                 let log_layer = otel_log::layer(endpoint, resource.clone());
-                let metrics_layer = otel_metrics::layer(resource);
+                let metrics_layer = otel_metrics::layer(endpoint, resource);
 
                 (
                     Some(trace_layer.and_then(log_layer).and_then(metrics_layer)),
@@ -62,7 +60,7 @@ fn init_tracing() -> Option<OpenTelemetryGuard> {
                 .with_filter(metrics_event_filter())
                 .and_then(opentelemetry_layers)
                 .with_filter(
-                    EnvFilter::try_from_default_env()
+                    EnvFilter::try_from_env("SYND_LOG")
                         .or_else(|_| EnvFilter::try_new("info"))
                         .unwrap()
                         .add_directive(audit::Audit::directive()),
@@ -77,18 +75,18 @@ fn init_tracing() -> Option<OpenTelemetryGuard> {
     guard
 }
 
-async fn run(Args { kvsd, tls }: Args, shutdown: Shutdown) -> anyhow::Result<()> {
+async fn run(Args { kvsd, tls, o11y }: Args, shutdown: Shutdown) -> anyhow::Result<()> {
     let dep = Dependency::new(kvsd, tls).await?;
 
-    info!(version = config::VERSION, "Runinng...");
+    info!(version = config::VERSION, otlp_endpoint=?o11y.otlp_endpoint, "Runinng...");
 
     listen_and_serve(dep, shutdown).await
 }
 
 #[tokio::main]
 async fn main() {
     let args = args::parse();
-    let _guard = init_tracing();
+    let _guard = init_tracing(&args.o11y);
     let shutdown = Shutdown::watch_signal();
 
     if let Err(err) = run(args, shutdown).await {

crates/synd_api/src/repository/kvsd.rs

@@ -109,6 +109,7 @@ impl KvsdClient {
 
 #[async_trait]
 impl SubscriptionRepository for KvsdClient {
+    #[tracing::instrument(skip_all)]
     async fn put_feed_subscription(
         &self,
         feed: repository::types::FeedSubscription,
@@ -131,6 +132,7 @@ impl SubscriptionRepository for KvsdClient {
         Self::set(&mut client, key, urls).await
     }
 
+    #[tracing::instrument(skip_all)]
     async fn delete_feed_subscription(
         &self,
         feed: repository::types::FeedSubscription,
@@ -148,6 +150,7 @@ impl SubscriptionRepository for KvsdClient {
         Self::set(&mut client, key, urls).await
     }
 
+    #[tracing::instrument(skip_all)]
     async fn fetch_subscribed_feed_urls(&self, user_id: &str) -> RepositoryResult<Vec<String>> {
         let key = Self::feed_subscription_key(user_id);
 

crates/synd_api/src/serve/auth.rs

@@ -35,6 +35,7 @@ impl Authenticator {
     }
 
     /// Authenticate from given token
+    #[tracing::instrument(skip_all)]
     pub async fn authenticate<S>(&self, token: S) -> Result<Principal, ()>
     where
         S: AsRef<str>,

crates/synd_api/src/serve/layer/request_metrics.rs

@@ -2,9 +2,10 @@ use std::{
     convert::Infallible,
     pin::Pin,
     task::{Context, Poll},
+    time::Instant,
 };
 
-use axum::{extract::Request, response::Response};
+use axum::{extract::Request, http::Method, response::Response};
 use futures_util::Future;
 use synd_o11y::metric;
 use tower::{Layer, Service};
@@ -46,15 +47,31 @@ where
     }
 
     fn call(&mut self, req: Request) -> Self::Future {
+        let start = Instant::now();
         let path = req.uri().path().to_owned();
+        let method = req.method().clone();
 
         let mut this = self.clone();
         Box::pin(async move {
             let response = this.inner.call(req).await.unwrap();
             let status = response.status().as_u16();
 
+            // https://opentelemetry.io/docs/specs/semconv/http/http-metrics/
+            // should be http.server.request ?
             metric!(monotonic_counter.request = 1, path, status);
 
+            // instrument graphql latency
+            if path == "/graphql" && method == Method::POST {
+                // f64 is matter
+                // The type of the field that MetricsVisitor visits when on_event() is called is pre defined.
+                // If u128 which is returned from elapsed() is used, it will not be visited, resulting in no metrics recorded.
+                // Spec say "When instruments are measuring durations, seconds SHOULD be used"
+                // https://opentelemetry.io/docs/specs/semconv/general/metrics/#instrument-units
+                let elapsed: f64 = start.elapsed().as_secs_f64();
+                // is there any semantic conventions?
+                metric!(histogram.graphql.duration = elapsed);
+            }
+
             Ok(response)
         })
     }

crates/synd_api/src/shutdown.rs

@@ -22,7 +22,7 @@ impl Shutdown {
                 Err(err) => tracing::error!("Failed to handle signal {err}"),
             }
             // Signal graceful shutdown to axum_server
-            handle2.graceful_shutdown(Some(Duration::from_secs(10)));
+            handle2.graceful_shutdown(Some(Duration::from_secs(3)));
             tx2.send(()).ok();
         });
 

crates/synd_api/src/usecase/fetch_entries.rs

@@ -50,6 +50,7 @@ impl Usecase for FetchEntries {
         Ok(principal)
     }
 
+    #[tracing::instrument(name = "fetch_entries", skip(self, principal))]
     async fn usecase(
         &self,
         Input {

crates/synd_api/src/usecase/subscribe_feed.rs

@@ -1,6 +1,7 @@
 use std::sync::Arc;
 
 use synd_feed::{feed::cache::FetchCachedFeed, types::Feed};
+use synd_o11y::metric;
 
 use crate::{
     principal::Principal,
@@ -70,6 +71,8 @@ impl Usecase for SubscribeFeed {
             })
             .await?;
 
+        metric!(counter.feed.subscription = 1);
+
         Ok(Output {
             output: SubscribeFeedOutput { feed },
         })

crates/synd_api/src/usecase/unsubscribe_feed.rs

@@ -1,5 +1,7 @@
 use std::sync::Arc;
 
+use synd_o11y::metric;
+
 use crate::{
     principal::Principal,
     repository::{self, SubscriptionRepository},
@@ -56,6 +58,8 @@ impl Usecase for UnsubscribeFeed {
             })
             .await?;
 
+        metric!(counter.feed.subscription = -1);
+
         Ok(Output {
             output: UnsubscribeFeedOutput {},
         })