Merge pull request #13 from CleverCloud/feat/tracing

Implements tracing trough opentelemetry

Author: FlorentinDUBOIS

Cargo.lock

@@ -13,49 +13,86 @@ keywords = ["kubernetes", "operator", "clevercloud", "openshift"]
 [dependencies]
 async-trait = "^0.1.51"
 chrono = "^0.4.19"
-clevercloud-sdk = { version = "^0.1.1", features = ["jsonschemas", "logging", "metrics"] }
+clevercloud-sdk = { version = "^0.2.0", features = ["jsonschemas"] }
 config = "^0.11.0"
 futures = "^0.3.17"
 hostname = "^0.3.1"
 hyper = { version = "^0.14.13", features = ["server", "tcp", "http1"] }
 json-patch = "^0.2.6"
-kube = { version = "^0.61.0", default-features = false, features = ["client", "rustls-tls", "ws", "gzip", "derive", "jsonpatch"] }
+kube = { version = "^0.61.0", default-features = false, features = [
+    "client",
+    "rustls-tls",
+    "ws",
+    "gzip",
+    "derive",
+    "jsonpatch",
+] }
 kube-derive = "^0.61.0"
 kube-runtime = "^0.61.0"
-k8s-openapi = { version = "^0.13.0", default-features = false, features = ["v1_21"] }
+k8s-openapi = { version = "^0.13.0", default-features = false, features = [
+    "v1_21",
+] }
 lazy_static = { version = "^1.4.0", optional = true }
+opentelemetry = { version = "^0.16.0", features = [
+    "rt-tokio",
+], optional = true }
+opentelemetry-jaeger = { version = "^0.15.0", features = [
+    "rt-tokio",
+    "collector_client",
+], optional = true }
 paw = "^1.0.0"
 prometheus = { version = "^0.13.0", optional = true }
-schemars = { version = "^0.8.6", features = [ "chrono", "indexmap", "uuid", "bytes", "url" ] }
+schemars = { version = "^0.8.6", features = [
+    "chrono",
+    "indexmap",
+    "uuid",
+    "bytes",
+    "url",
+] }
 sentry = { version = "^0.23.0", optional = true }
 serde = { version = "^1.0.130", features = ["derive"] }
-serde_json = { version = "^1.0.68", features = ["preserve_order", "float_roundtrip"] }
+serde_json = { version = "^1.0.68", features = [
+    "preserve_order",
+    "float_roundtrip",
+] }
 serde_yaml = "^0.8.21"
 slog = { version = "^2.7.0" }
 slog-async = "^2.7.0"
 slog-term = "^2.8.0"
 slog-scope = "^4.4.0"
 slog-stdlog = { version = "^4.1.0", optional = true }
-structopt = { version = "^0.3.23", features = ["paw"] }
+structopt = { version = "^0.3.25", features = ["paw"] }
 thiserror = "^1.0.29"
 tokio = { version = "^1.12.0", features = ["full"] }
+tracing = { version = "^0.1.29", optional = true }
+tracing-futures = { version = "^0.2.5", features = ["tokio"], optional = true }
+tracing-subscriber = { version = "^0.2.25", features = [
+    "chrono",
+], optional = true }
+tracing-opentelemetry = { version = "^0.15.0", optional = true }
 
 [features]
-default = ["metrics", "release-log-debug", "debug-log-trace"]
-logging = ["slog-stdlog"]
-metrics = ["lazy_static", "prometheus"]
+default = [
+    "metrics",
+    "trace",
+    "tracker",
+    "slog/release_max_level_debug",
+    "slog/max_level_trace",
+]
+logging = [
+    "clevercloud-sdk/logging",
+    "tracing-subscriber/tracing-log",
+    "slog-stdlog",
+]
+metrics = ["clevercloud-sdk/metrics", "lazy_static", "prometheus"]
 tracker = ["sentry"]
-
-release-log-off     = ["slog/release_max_level_off"]
-release-log-error   = ["slog/release_max_level_error"]
-release-log-warning = ["slog/release_max_level_warn"]
-release-log-info    = ["slog/release_max_level_info"]
-release-log-debug   = ["slog/release_max_level_debug"]
-release-log-trace   = ["slog/release_max_level_trace"]
-
-debug-log-off     = ["slog/max_level_off"]
-debug-log-error   = ["slog/max_level_error"]
-debug-log-warning = ["slog/max_level_warn"]
-debug-log-info    = ["slog/max_level_info"]
-debug-log-debug   = ["slog/max_level_debug"]
-debug-log-trace   = ["slog/max_level_trace"]
+trace = [
+    "clevercloud-sdk/trace",
+    "clevercloud-sdk/tokio",
+    "tracing",
+    "tracing-futures",
+    "tracing-subscriber",
+    "tracing-opentelemetry",
+    "opentelemetry",
+    "opentelemetry-jaeger",
+]

Cargo.toml

@@ -25,6 +25,7 @@ pub enum CustomResource {
 impl FromStr for CustomResource {
     type Err = Box<dyn Error + Send + Sync>;
 
+    #[cfg_attr(feature = "trace", tracing::instrument)]
     fn from_str(s: &str) -> Result<Self, Self::Err> {
         match s.to_lowercase().as_str() {
             "postgresql" => Ok(Self::PostgreSql),
@@ -59,6 +60,7 @@ pub enum CustomResourceDefinition {
 impl Executor for CustomResourceDefinition {
     type Error = CustomResourceDefinitionError;
 
+    #[cfg_attr(feature = "trace", tracing::instrument)]
     async fn execute(&self, config: Arc<Configuration>) -> Result<(), Self::Error> {
         match self {
             Self::View { custom_resource } => view(config, custom_resource).await,
@@ -69,6 +71,7 @@ impl Executor for CustomResourceDefinition {
 // -----------------------------------------------------------------------------
 // view function
 
+#[cfg_attr(feature = "trace", tracing::instrument)]
 pub async fn view(
     _config: Arc<Configuration>,
     custom_resource: &Option<CustomResource>,

src/cmd/crd.rs

@@ -59,6 +59,7 @@ pub enum Command {
 impl Executor for Command {
     type Error = CommandError;
 
+    #[cfg_attr(feature = "trace", tracing::instrument)]
     async fn execute(&self, config: Arc<Configuration>) -> Result<(), Self::Error> {
         match self {
             Self::CustomResourceDefinition(crd) => crd
@@ -110,6 +111,7 @@ pub enum DaemonError {
 // -----------------------------------------------------------------------------
 // daemon function
 
+#[cfg_attr(feature = "trace", tracing::instrument)]
 pub async fn daemon(
     kubeconfig: Option<PathBuf>,
     config: Arc<Configuration>,

src/cmd/mod.rs

@@ -4,10 +4,16 @@
 //! resource definition
 use std::{convert::TryFrom, error::Error, sync::Arc};
 
+#[cfg(feature = "trace")]
+use opentelemetry::global;
+#[cfg(feature = "trace")]
+use opentelemetry_jaeger::Propagator;
 use slog::{o, Drain, Level, LevelFilter, Logger};
 use slog_async::Async;
 use slog_scope::{crit, debug, info, set_global_logger, GlobalLoggerGuard as Guard};
 use slog_term::{FullFormat, TermDecorator};
+#[cfg(feature = "trace")]
+use tracing_subscriber::{layer::SubscriberExt, Registry};
 
 use crate::{
     cmd::{daemon, Args, Executor},
@@ -51,14 +57,36 @@ pub(crate) async fn main(args: Args) -> Result<(), Box<dyn Error + Send + Sync>>
     }
 
     #[cfg(feature = "tracker")]
-    if let Some(dsn) = config.sentry.dsn.to_owned() {
-        let _sguard = sentry::init((
-            dsn,
+    let _sguard = config.sentry.dsn.as_ref().map(|dsn| {
+        sentry::init((
+            dsn.to_owned(),
             sentry::ClientOptions {
                 release: sentry::release_name!(),
                 ..Default::default()
             },
-        ));
+        ))
+    });
+
+    #[cfg(feature = "trace")]
+    if let Some(jaeger) = &config.jaeger {
+        info!("Start to trace using jaeger with opentelemetry compatibility"; "endpoint" => jaeger.endpoint.to_owned());
+        global::set_text_map_propagator(Propagator::new());
+
+        let mut builder = opentelemetry_jaeger::new_pipeline()
+            .with_collector_endpoint(jaeger.endpoint.to_owned())
+            .with_service_name(env!("CARGO_PKG_NAME"));
+
+        if let Some(user) = &jaeger.user {
+            builder = builder.with_collector_username(user);
+        }
+
+        if let Some(password) = &jaeger.password {
+            builder = builder.with_collector_password(password);
+        }
+
+        let layer = tracing_opentelemetry::layer().with_tracer(builder.install_simple()?);
+
+        tracing::subscriber::set_global_default(Registry::default().with(layer))?;
     }
 
     let result: Result<_, Box<dyn Error + Send + Sync>> = match &args.command {
@@ -71,6 +99,9 @@ pub(crate) async fn main(args: Args) -> Result<(), Box<dyn Error + Send + Sync>>
         return Err(err);
     }
 
+    #[cfg(feature = "trace")]
+    global::shutdown_tracer_provider();
+
     info!("{} halted!", env!("CARGO_PKG_NAME"));
     Ok(())
 }

src/main.rs

@@ -41,6 +41,7 @@ pub struct Api {
 
 #[allow(clippy::from_over_into)]
 impl Into<Credentials> for Api {
+    #[cfg_attr(feature = "trace", tracing::instrument)]
     fn into(self) -> Credentials {
         Credentials {
             token: self.token.to_owned(),
@@ -76,6 +77,17 @@ pub struct Sentry {
     pub dsn: Option<String>,
 }
 
+// -----------------------------------------------------------------------------
+// Jaeger structure
+
+#[cfg(feature = "trace")]
+#[derive(Serialize, Deserialize, PartialEq, Eq, Clone, Debug, Default)]
+pub struct Jaeger {
+    pub endpoint: String,
+    pub user: Option<String>,
+    pub password: Option<String>,
+}
+
 // -----------------------------------------------------------------------------
 // Configuration structures
 
@@ -88,11 +100,15 @@ pub struct Configuration {
     #[cfg(feature = "tracker")]
     #[serde(rename = "sentry", default = "Default::default")]
     pub sentry: Sentry,
+    #[cfg(feature = "trace")]
+    #[serde(rename = "jaeger")]
+    pub jaeger: Option<Jaeger>,
 }
 
 impl TryFrom<PathBuf> for Configuration {
     type Error = ConfigurationError;
 
+    #[cfg_attr(feature = "trace", tracing::instrument)]
     fn try_from(path: PathBuf) -> Result<Self, Self::Error> {
         let mut config = Config::default();
 
@@ -119,6 +135,7 @@ impl TryFrom<PathBuf> for Configuration {
 }
 
 impl Configuration {
+    #[cfg_attr(feature = "trace", tracing::instrument)]
     pub fn try_default() -> Result<Self, ConfigurationError> {
         let mut config = Config::default();
 

src/svc/cfg.rs

@@ -3,7 +3,7 @@
 //! This module provide extensions to help building custom resource reconciler
 //! loop
 
-use std::collections::BTreeMap;
+use std::{collections::BTreeMap, fmt::Debug};
 
 use async_trait::async_trait;
 use clevercloud_sdk::{
@@ -18,7 +18,7 @@ use slog_scope::{debug, trace};
 // AddonExt trait
 
 #[async_trait]
-pub trait AddonExt: Into<CreateAddonOpts> + Clone + Sync + Send {
+pub trait AddonExt: Into<CreateAddonOpts> + Clone + Debug + Sync + Send {
     type Error: From<ClientError> + Sync + Send;
 
     fn id(&self) -> Option<String>;
@@ -27,6 +27,7 @@ pub trait AddonExt: Into<CreateAddonOpts> + Clone + Sync + Send {
 
     fn name(&self) -> String;
 
+    #[cfg_attr(feature = "trace", tracing::instrument)]
     async fn get(&self, client: &Client) -> Result<Option<Addon>, Self::Error> {
         if let Some(id) = &self.id() {
             trace!("Retrieve the addon from the identifier"; "id" => &id, "name" => self.name());
@@ -56,6 +57,7 @@ pub trait AddonExt: Into<CreateAddonOpts> + Clone + Sync + Send {
         Ok(None)
     }
 
+    #[cfg_attr(feature = "trace", tracing::instrument)]
     async fn upsert(&self, client: &Client) -> Result<Addon, Self::Error> {
         debug!("Try to retrieve the addon, before creating a new one"; "id" => &self.id(), "name" => self.name());
         if let Some(addon) = self.get(client).await? {
@@ -66,6 +68,7 @@ pub trait AddonExt: Into<CreateAddonOpts> + Clone + Sync + Send {
         Ok(addon::create(client, &self.organisation(), &self.to_owned().into()).await?)
     }
 
+    #[cfg_attr(feature = "trace", tracing::instrument)]
     async fn delete(&self, client: &Client) -> Result<(), Self::Error> {
         if let Some(a) = self.get(client).await? {
             addon::delete(client, &self.organisation(), &a.id).await?;
@@ -74,6 +77,7 @@ pub trait AddonExt: Into<CreateAddonOpts> + Clone + Sync + Send {
         Ok(())
     }
 
+    #[cfg_attr(feature = "trace", tracing::instrument)]
     async fn secrets(
         &self,
         client: &Client,

src/svc/clevercloud/ext.rs

@@ -98,6 +98,7 @@ pub struct PostgreSqlStatus {
 
 #[allow(clippy::from_over_into)]
 impl Into<CreateAddonOpts> for PostgreSql {
+    #[cfg_attr(feature = "trace", tracing::instrument)]
     fn into(self) -> CreateAddonOpts {
         CreateAddonOpts {
             name: AddonExt::name(&self),
@@ -112,6 +113,7 @@ impl Into<CreateAddonOpts> for PostgreSql {
 impl AddonExt for PostgreSql {
     type Error = ReconcilerError;
 
+    #[cfg_attr(feature = "trace", tracing::instrument)]
     fn id(&self) -> Option<String> {
         if let Some(status) = &self.status {
             return status.addon.to_owned();
@@ -120,10 +122,12 @@ impl AddonExt for PostgreSql {
         None
     }
 
+    #[cfg_attr(feature = "trace", tracing::instrument)]
     fn organisation(&self) -> String {
         self.spec.organisation.to_owned()
     }
 
+    #[cfg_attr(feature = "trace", tracing::instrument)]
     fn name(&self) -> String {
         "kubernetes_".to_string()
             + &self
@@ -133,13 +137,15 @@ impl AddonExt for PostgreSql {
 }
 
 impl PostgreSql {
+    #[cfg_attr(feature = "trace", tracing::instrument)]
     pub fn set_addon_id(&mut self, id: Option<String>) {
         let mut status = self.status.get_or_insert_with(PostgreSqlStatus::default);
 
         status.addon = id;
         self.status = Some(status.to_owned());
     }
 
+    #[cfg_attr(feature = "trace", tracing::instrument)]
     pub fn get_addon_id(&self) -> Option<String> {
         self.status
             .to_owned()
@@ -190,18 +196,21 @@ pub enum ReconcilerError {
 }
 
 impl From<kube::Error> for ReconcilerError {
+    #[cfg_attr(feature = "trace", tracing::instrument)]
     fn from(err: kube::Error) -> Self {
         Self::KubeClient(err)
     }
 }
 
 impl From<ClientError> for ReconcilerError {
+    #[cfg_attr(feature = "trace", tracing::instrument)]
     fn from(err: ClientError) -> Self {
         Self::CleverClient(err)
     }
 }
 
 impl From<controller::Error<Self, watcher::Error>> for ReconcilerError {
+    #[cfg_attr(feature = "trace", tracing::instrument)]
     fn from(err: controller::Error<ReconcilerError, watcher::Error>) -> Self {
         Self::Reconcile(err.to_string())
     }

src/svc/crd/postgresql.rs

@@ -9,6 +9,7 @@ use kube::{
     Config,
 };
 
+#[cfg_attr(feature = "trace", tracing::instrument)]
 /// returns a new kubernetes client from the given path if defined
 /// or retrieve it from environment or defaults paths
 pub async fn try_new(path: Option<PathBuf>) -> Result<kube::Client, kube::Error> {
@@ -20,5 +21,5 @@ pub async fn try_new(path: Option<PathBuf>) -> Result<kube::Client, kube::Error>
     let opts = KubeConfigOptions::default();
     let config = Config::from_custom_kubeconfig(kubeconfig, &opts).await?;
 
-    Ok(kube::Client::try_from(config)?)
+    kube::Client::try_from(config)
 }