Dev (#662)

* estimate span size before any processing (#654)

* Feat/lam 682 (#655)

* feat: update delete project, fix spans filters

* fix: rendering, model fallback

* feat: move delete endpoint to next

* feat: fix margins

* feat: update delete project data in ch, fix comments

* fix: command

* feat: add debug console

* feat: move helper function

* allow CTES in SQL editor (#653)

* allow CTES in SQL editor

* remove debug console.log

* fix reset usage when upgrading (#657)

* Hotfix/ack browser events (#660)

* first ack browser events, then increment usage

* don't wait for async insert on browser events

* feat: fix decode key by passing in docker file (#661)

* feat: fix decode key by passing in docker file

* feat: add key to full build

---------

Co-authored-by: Dinmukhamed Mailibay <47117969+dinmukhamedm@users.noreply.github.com>

Author: olzhik11

app-server/src/browser_events/mod.rs

@@ -92,6 +92,9 @@ async fn inner_process_browser_events(
 
         match backoff::future::retry(exponential_backoff, insert_browser_events).await {
             Ok(recorded_bytes) => {
+                if let Err(e) = acker.ack().await {
+                    log::error!("Failed to ack MQ delivery (browser events): {:?}", e);
+                }
                 if let Err(e) = increment_project_browser_events_bytes_ingested(
                     &db.pool,
                     &project_id,
@@ -104,9 +107,6 @@ async fn inner_process_browser_events(
                         e
                     );
                 }
-                if let Err(e) = acker.ack().await {
-                    log::error!("Failed to ack MQ delivery (browser events): {:?}", e);
-                }
             }
             Err(e) => {
                 log::error!(

app-server/src/ch/browser_events.rs

@@ -38,8 +38,7 @@ pub async fn insert_browser_events(
             );
             e
         })?
-        .with_option("async_insert", "1")
-        .with_option("wait_for_async_insert", "1");
+        .with_option("async_insert", "1");
 
     let mut total_size_bytes = 0;
 

app-server/src/db/projects.rs

@@ -48,12 +48,3 @@ pub async fn create_project(
 
     Ok(project)
 }
-
-pub async fn delete_project(pool: &PgPool, project_id: &Uuid) -> Result<()> {
-    sqlx::query("DELETE FROM projects WHERE id = $1")
-        .bind(project_id)
-        .execute(pool)
-        .await?;
-
-    Ok(())
-}

app-server/src/main.rs

@@ -550,7 +550,7 @@ fn main() -> anyhow::Result<()> {
                                 .service(routes::labels::update_label_class)
                                 .service(routes::traces::get_traces_metrics)
                                 .service(routes::provider_api_keys::save_api_key)
-                                .service(routes::spans::create_span),
+                                .service(routes::spans::create_span)
                         )
                         .service(routes::probes::check_health)
                         .service(routes::probes::check_ready)

app-server/src/routes/mod.rs

@@ -5,7 +5,6 @@ pub mod error;
 pub mod evaluations;
 pub mod labels;
 pub mod probes;
-pub mod projects;
 pub mod provider_api_keys;
 pub mod spans;
 pub mod traces;

app-server/src/routes/projects.rs

@@ -13,6 +13,8 @@ use crate::{
     features::{Feature, is_feature_enabled},
     mq::{MessageQueue, MessageQueueDeliveryTrait, MessageQueueReceiverTrait, MessageQueueTrait},
     storage::Storage,
+    traces::IngestedBytes,
+    utils::estimate_json_size,
 };
 
 pub async fn process_queue_spans(
@@ -100,8 +102,15 @@ async fn inner_process_queue_spans(
                 }
             }
         }
-
         let mut span: Span = rabbitmq_span_message.span;
+        let events = rabbitmq_span_message.events;
+
+        // Make sure we count the sizes before any processing, as soon as
+        // we pick up the span from the queue.
+        let span_bytes = estimate_json_size(
+            &serde_json::to_value(&span.get_attributes().attributes).unwrap_or_default(),
+        );
+        let events_bytes = estimate_json_size(&serde_json::to_value(&events).unwrap_or_default());
 
         // Parse and enrich span attributes for input/output extraction
         // This heavy processing is done on the consumer side
@@ -121,12 +130,14 @@ async fn inner_process_queue_spans(
             }
         }
 
-        let events = rabbitmq_span_message.events;
-
         process_spans_and_events(
             &mut span,
             events,
             &rabbitmq_span_message.project_id,
+            &IngestedBytes {
+                span_bytes,
+                events_bytes,
+            },
             db.clone(),
             clickhouse.clone(),
             cache.clone(),

app-server/src/traces/consumer.rs

@@ -21,7 +21,6 @@ use crate::{
         provider::convert_span_to_provider_format,
         utils::{get_llm_usage_for_span, record_labels_to_db_and_ch, record_span_to_db},
     },
-    utils::estimate_json_size,
 };
 
 pub mod attributes;
@@ -39,10 +38,16 @@ pub const OBSERVATIONS_QUEUE: &str = "observations_queue";
 pub const OBSERVATIONS_EXCHANGE: &str = "observations_exchange";
 pub const OBSERVATIONS_ROUTING_KEY: &str = "observations_routing_key";
 
+pub struct IngestedBytes {
+    pub span_bytes: usize,
+    pub events_bytes: usize,
+}
+
 pub async fn process_spans_and_events(
     span: &mut Span,
     events: Vec<Event>,
     project_id: &Uuid,
+    ingested_bytes: &IngestedBytes,
     db: Arc<DB>,
     clickhouse: clickhouse::Client,
     cache: Arc<Cache>,
@@ -120,9 +125,7 @@ pub async fn process_spans_and_events(
                 let _ = acker.ack().await.map_err(|e| {
                     log::error!("Failed to ack MQ delivery (span): {:?}", e);
                 });
-                estimate_json_size(
-                    &serde_json::to_value(&span.get_attributes().attributes).unwrap_or_default(),
-                )
+                ingested_bytes.span_bytes
             }
             Err(e) => {
                 log::error!(
@@ -147,7 +150,7 @@ pub async fn process_spans_and_events(
     }
 
     let recorded_events_bytes = match record_events(db.clone(), clickhouse.clone(), &events).await {
-        Ok(_) => estimate_json_size(&serde_json::to_value(&events).unwrap_or_default()),
+        Ok(_) => ingested_bytes.events_bytes,
         Err(e) => {
             log::error!("Failed to record events: {:?}", e);
             0
@@ -157,7 +160,7 @@ pub async fn process_spans_and_events(
     if let Err(e) = increment_project_spans_bytes_ingested(
         &db.pool,
         &project_id,
-        recorded_span_bytes + recorded_events_bytes,
+        recorded_events_bytes + recorded_span_bytes,
     )
     .await
     {

app-server/src/traces/mod.rs

@@ -128,6 +128,7 @@ services:
       CLICKHOUSE_USER: ${CLICKHOUSE_USER}
       CLICKHOUSE_PASSWORD: ${CLICKHOUSE_PASSWORD}
       OPENAI_API_KEY: ${OPENAI_API_KEY}
+      AEAD_SECRET_KEY: ${AEAD_SECRET_KEY}
 
 volumes:
   clickhouse-data:

docker-compose-full.yml

@@ -25,10 +25,9 @@ services:
       - type: volume
         source: clickhouse-logs
         target: /var/log/clickhouse-server/
-      - type: bind
-        source: ./clickhouse/config/users.xml
-        target: /etc/clickhouse-server/users.d/users.xml
-        read_only: true
+    environment:
+        CLICKHOUSE_USER: ${CLICKHOUSE_USER}
+        CLICKHOUSE_PASSWORD: ${CLICKHOUSE_PASSWORD}
     ulimits:
       nofile:
         soft: 262144
@@ -80,8 +79,6 @@ services:
       - "${APP_SERVER_GRPC_HOST_PORT:-8001}:8001"
     build:
       context: ./app-server
-      args:
-        DATABASE_URL: postgres://${POSTGRES_USER}:${POSTGRES_PASSWORD}@postgres:5432/${POSTGRES_DB}
     container_name: app-server
     depends_on:
       postgres:
@@ -103,6 +100,8 @@ services:
       CLICKHOUSE_PASSWORD: ${CLICKHOUSE_PASSWORD}
       ENVIRONMENT: FULL
       AGENT_MANAGER_URL: http://agent-manager:8901
+      AEAD_SECRET_KEY: ${AEAD_SECRET_KEY}
+
   frontend:
     build:
       context: ./frontend
@@ -124,6 +123,7 @@ services:
       CLICKHOUSE_USER: ${CLICKHOUSE_USER}
       CLICKHOUSE_PASSWORD: ${CLICKHOUSE_PASSWORD}
       OPENAI_API_KEY: ${OPENAI_API_KEY}
+      AEAD_SECRET_KEY: ${AEAD_SECRET_KEY}
 
 volumes:
   clickhouse-data: