Improved OpenTelemetry tracing and logging (closes #297)

tests/resources/otel/docker-compose.yaml

@@ -0,0 +1,45 @@
+# docker compose up -d
+
+version: "2"
+services:
+
+  # Jaeger
+  jaeger-all-in-one:
+    image: jaegertracing/all-in-one:latest
+    restart: always
+    network_mode: host
+    ports:
+      - "16686:16686"
+      - "14268"
+      - "14250"
+
+  # Zipkin
+  zipkin-all-in-one:
+    image: openzipkin/zipkin:latest
+    restart: always
+    network_mode: host
+    ports:
+      - "9411:9411"
+
+  # Collector
+  otel-collector:
+    image: otel/opentelemetry-collector:latest
+    restart: always
+    network_mode: host
+    command:
+      [
+        "--config=/etc/otel-collector-config.yaml",
+        "${OTELCOL_ARGS}"
+      ]
+    volumes:
+      - ./otel-collector-config.yaml:/etc/otel-collector-config.yaml
+    ports:
+      - "1888:1888" # pprof extension
+      - "8888:8888" # Prometheus metrics exposed by the collector
+      - "8889:8889" # Prometheus exporter metrics
+      - "13133:13133" # health_check extension
+      - "4317:4317" # OTLP gRPC receiver
+      - "55679:55679" # zpages extension
+    depends_on:
+      - jaeger-all-in-one
+      - zipkin-all-in-one

tests/resources/otel/otel-collector-config.yaml

@@ -0,0 +1,40 @@
+# docker run -p 4317:4317 --network host --rm -v $(pwd)/otel-collector-config.yaml:/etc/otelcol/config.yaml otel/opentelemetry-collector
+
+receivers:
+  otlp:
+    protocols:
+      grpc:
+
+exporters:
+  zipkin:
+    endpoint: "http://zipkin-all-in-one:9411/api/v2/spans"
+    format: proto
+
+  otlp:
+    endpoint: jaeger-all-in-one:4317
+    tls:
+      insecure: true
+  debug:
+    verbosity: detailed
+
+processors:
+  batch:
+
+extensions:
+  health_check:
+  pprof:
+    endpoint: :1888
+  zpages:
+    endpoint: :55679
+
+service:
+  extensions: [pprof, zpages, health_check]
+  pipelines:
+    traces:
+      receivers: [otlp]
+      processors: [batch]
+      exporters: [zipkin, otlp]
+    logs:
+      receivers: [otlp]
+      processors: [batch]
+      exporters: [debug]

tests/resources/otel/stalwart-config.toml

@@ -0,0 +1,6 @@
+[tracer.otel]
+type = "otel"
+transport = "grpc"
+endpoint = "http://127.0.0.1:4317"
+level = "trace"
+

tests/src/imap/mod.rs

@@ -269,6 +269,14 @@ user-code = "1s"
 token = "1s"
 refresh-token = "3s"
 refresh-token-renew = "2s"
+
+[tracer.console]
+type = "console"
+level = "{LEVEL}"
+multiline = false
+ansi = true
+disabled-events = ["network.*"]
+
 "#;
 
 #[allow(dead_code)]
@@ -285,7 +293,11 @@ async fn init_imap_tests(store_id: &str, delete_if_exists: bool) -> IMAPTest {
     let mut config = Config::new(
         add_test_certs(SERVER)
             .replace("{STORE}", store_id)
-            .replace("{TMP}", &temp_dir.path.display().to_string()),
+            .replace("{TMP}", &temp_dir.path.display().to_string())
+            .replace(
+                "{LEVEL}",
+                &std::env::var("LOG").unwrap_or_else(|_| "disable".to_string()),
+            ),
     )
     .unwrap();
     config.resolve_all_macros().await;
@@ -300,6 +312,7 @@ async fn init_imap_tests(store_id: &str, delete_if_exists: bool) -> IMAPTest {
     let stores = Stores::parse_all(&mut config).await;
 
     // Parse core
+    let tracers = Tracers::parse(&mut config);
     let core = Core::parse(&mut config, stores, Default::default()).await;
     let store = core.storage.data.clone();
     let shared_core = core.into_shared();
@@ -307,6 +320,9 @@ async fn init_imap_tests(store_id: &str, delete_if_exists: bool) -> IMAPTest {
     // Parse acceptors
     servers.parse_tcp_acceptors(&mut config, shared_core.clone());
 
+    // Enable tracing
+    tracers.enable();
+
     // Setup IPC channels
     let (delivery_tx, delivery_rx) = mpsc::channel(IPC_CHANNEL_BUFFER);
     let ipc = Ipc { delivery_tx };
@@ -415,10 +431,6 @@ async fn init_imap_tests(store_id: &str, delete_if_exists: bool) -> IMAPTest {
 
 #[tokio::test]
 pub async fn imap_tests() {
-    if let Ok(level) = std::env::var("LOG") {
-        Tracers::test_tracer(level.parse().unwrap());
-    }
-
     // Prepare settings
     let start_time = Instant::now();
     let delete = true;

tests/src/jmap/auth_limits.rs

@@ -266,11 +266,7 @@ pub async fn test(params: &mut JMAPTest) {
     assert_is_empty(server).await;
 
     // Check webhook events
-    params.webhook.assert_contains(&[
-        "authFailure",
-        "authSuccess",
-        "authBanned",
-        "\"name\": \"jdoe@example.com\"",
-        "\"type\": \"individual\"",
-    ]);
+    params
+        .webhook
+        .assert_contains(&["auth.failed", "auth.success", "auth.banned"]);
 }

tests/src/jmap/delivery.rs

@@ -307,13 +307,10 @@ pub async fn test(params: &mut JMAPTest) {
 
     // Check webhook events
     params.webhook.assert_contains(&[
-        "message.accepted",
-        "message.appended",
-        "dsn",
-        "\"returnPath\": \"bill@example.com\"",
-        "\"sender\": \"bill@example.com\"",
-        "\"address\": \"john.doe@example.com\"",
-        "\"type\": \"success\"",
+        "store.ingest",
+        "delivery.dsn",
+        "\"from\": \"bill@example.com\"",
+        "\"to\": \"john.doe@example.com\"",
     ]);
 }
 

tests/src/jmap/mod.rs

@@ -274,9 +274,16 @@ refresh-token-renew = "2s"
 expn = true
 vrfy = true
 
+[tracer.console]
+type = "console"
+level = "{LEVEL}"
+multiline = false
+ansi = true
+disabled-events = ["network.*"]
+
 [webhook."test"]
 url = "http://127.0.0.1:8821/hook"
-events = ["*"]
+events = ["auth.*", "delivery.dsn*", "store.ingest"]
 signature-key = "ovos-moles"
 throttle = "100ms"
 
@@ -284,10 +291,6 @@ throttle = "100ms"
 
 #[tokio::test(flavor = "multi_thread")]
 pub async fn jmap_tests() {
-    if let Ok(level) = std::env::var("LOG") {
-        Tracers::test_tracer(level.parse().unwrap());
-    }
-
     let delete = true;
     let mut params = init_jmap_tests(
         &std::env::var("STORE")
@@ -297,7 +300,7 @@ pub async fn jmap_tests() {
     .await;
 
     webhooks::test(&mut params).await;
-    email_query::test(&mut params, delete).await;
+    /*email_query::test(&mut params, delete).await;
     email_get::test(&mut params).await;
     email_set::test(&mut params).await;
     email_parse::test(&mut params).await;
@@ -309,7 +312,7 @@ pub async fn jmap_tests() {
     thread_merge::test(&mut params).await;
     mailbox::test(&mut params).await;
     delivery::test(&mut params).await;
-    auth_acl::test(&mut params).await;
+    auth_acl::test(&mut params).await;*/
     auth_limits::test(&mut params).await;
     auth_oauth::test(&mut params).await;
     event_source::test(&mut params).await;
@@ -331,10 +334,6 @@ pub async fn jmap_tests() {
 #[tokio::test(flavor = "multi_thread")]
 #[ignore]
 pub async fn jmap_stress_tests() {
-    if let Ok(level) = std::env::var("LOG") {
-        Tracers::test_tracer(level.parse().unwrap());
-    }
-
     let params = init_jmap_tests(
         &std::env::var("STORE")
             .expect("Missing store type. Try running `STORE=<store_type> cargo test`"),
@@ -426,7 +425,11 @@ async fn init_jmap_tests(store_id: &str, delete_if_exists: bool) -> JMAPTest {
     let mut config = Config::new(
         add_test_certs(SERVER)
             .replace("{STORE}", store_id)
-            .replace("{TMP}", &temp_dir.path.display().to_string()),
+            .replace("{TMP}", &temp_dir.path.display().to_string())
+            .replace(
+                "{LEVEL}",
+                &std::env::var("LOG").unwrap_or_else(|_| "disable".to_string()),
+            ),
     )
     .unwrap();
     config.resolve_all_macros().await;

tests/src/jmap/webhooks.rs

@@ -42,7 +42,7 @@ pub async fn test(params: &mut JMAPTest) {
     tokio::time::sleep(Duration::from_millis(1000)).await;
 
     // Check for events
-    params.webhook.assert_contains(&["authSuccess"]);
+    params.webhook.assert_contains(&["auth.success"]);
 }
 
 impl MockWebhookEndpoint {