Merge pull request #605 from lmnr-ai/dev

Span level errors, clickhouse migrations

Author: dinmukhamedm

.env

@@ -7,7 +7,8 @@ SHARED_SECRET_TOKEN=some_secret
 RABBITMQ_DEFAULT_USER=admin
 RABBITMQ_DEFAULT_PASS=adminpasswd
 
-CLICKHOUSE_USER=default
+CLICKHOUSE_USER=ch_user
+CLICKHOUSE_PASSWORD=ch_passwd
 POSTGRES_PORT=5433
 # must be exactly 32 bytes (64 hex characters)
 AEAD_SECRET_KEY=0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef

.github/workflows/backend-build.yml

@@ -7,7 +7,13 @@ on:
       - opened
       - reopened
     paths:
-      - "app-server/**"
+      # Explicit list prevents triggering build on changes such as
+      # README, .env.example.
+      - "app-server/src/**"
+      - "app-server/proto/**"
+      - "app-server/**/*.rs"
+      - "app-server/Cargo.lock"
+      - "app-server/Cargo.toml"
 
 jobs:
   build:

.github/workflows/fe-tests-check.yml

@@ -0,0 +1,38 @@
+name: Check Frontend Tests
+
+on:
+  pull_request:
+    types:
+      - synchronize
+      - opened
+      - reopened
+    paths:
+      - 'frontend/**' 
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+
+      - name: Install pnpm
+        uses: pnpm/action-setup@v4
+        with:
+          version: 9
+
+      - name: Set up Node.js
+        uses: actions/setup-node@v4
+        with:
+          node-version: 'lts/*'
+          cache: 'pnpm'
+          cache-dependency-path: ./frontend/pnpm-lock.yaml
+
+      - name: Install dependencies
+        working-directory: ./frontend
+        run: pnpm install
+
+      - name: Run tests
+        working-directory: ./frontend
+        run: pnpm test

app-server/.env.example

@@ -9,7 +9,8 @@ GRPC_PORT=8001
 RABBITMQ_URL=amqp://admin:adminpasswd@localhost:5672/%2f
 
 CLICKHOUSE_URL=http://localhost:8123
-CLICKHOUSE_USER=default
+CLICKHOUSE_USER=ch_user
+CLICKHOUSE_PASSWORD=ch_passwd
 
 SHARED_SECRET_TOKEN=some_secret
 # must be exactly 32 bytes (64 hex characters)

app-server/src/ch/spans.rs

@@ -64,6 +64,7 @@ pub struct CHSpan {
     pub path: String,
     pub input: String,
     pub output: String,
+    pub status: String,
     #[serde(default)]
     pub size_bytes: u64,
 }
@@ -117,6 +118,7 @@ impl CHSpan {
                 .unwrap_or(String::from("<null>")),
             input: span_input_string,
             output: span_output_string,
+            status: span.status.clone().unwrap_or(String::from("<null>")),
             size_bytes: size_bytes as u64,
         }
     }

app-server/src/db/spans.rs

@@ -55,6 +55,7 @@ pub struct Span {
     pub start_time: DateTime<Utc>,
     pub end_time: DateTime<Utc>,
     pub events: Option<Value>,
+    pub status: Option<String>,
     pub labels: Option<Value>,
     pub input_url: Option<String>,
     pub output_url: Option<String>,
@@ -108,6 +109,7 @@ pub async fn record_span(pool: &PgPool, span: &Span, project_id: &Uuid) -> Resul
             output_preview,
             input_url,
             output_url,
+            status,
             project_id
         )
         VALUES(
@@ -125,7 +127,8 @@ pub async fn record_span(pool: &PgPool, span: &Span, project_id: &Uuid) -> Resul
             $12,
             $13,
             $14,
-            $15)
+            $15,
+            $16)
         ON CONFLICT (span_id, project_id) DO UPDATE SET
             trace_id = EXCLUDED.trace_id,
             parent_span_id = EXCLUDED.parent_span_id,
@@ -139,7 +142,8 @@ pub async fn record_span(pool: &PgPool, span: &Span, project_id: &Uuid) -> Resul
             input_preview = EXCLUDED.input_preview,
             output_preview = EXCLUDED.output_preview,
             input_url = EXCLUDED.input_url,
-            output_url = EXCLUDED.output_url
+            output_url = EXCLUDED.output_url,
+            status = EXCLUDED.status
     ",
     )
     .bind(&span.span_id)
@@ -156,6 +160,7 @@ pub async fn record_span(pool: &PgPool, span: &Span, project_id: &Uuid) -> Resul
     .bind(&output_preview)
     .bind(&span.input_url as &Option<String>)
     .bind(&span.output_url as &Option<String>)
+    .bind(&span.status)
     .bind(&project_id)
     .execute(pool)
     .await?;

app-server/src/db/stats.rs

@@ -140,62 +140,6 @@ pub struct WorkspaceStats {
     pub storage_limit: i64,
 }
 
-pub async fn get_workspace_stats(pool: &PgPool, workspace_id: &Uuid) -> Result<WorkspaceStats> {
-    let workspace_stats = sqlx::query_as::<_, WorkspaceStats>(
-        "WITH members_per_workspace AS (
-            SELECT
-                workspace_id,
-                count(user_id)::int8 as members
-            FROM
-                members_of_workspaces
-            WHERE
-                workspace_id = $1
-            GROUP BY workspace_id
-        )
-        SELECT
-            subscription_tiers.name as tier_name,
-            subscription_tiers.members_per_workspace as seats_included_in_tier,
-            workspace_usage.span_count as total_spans,
-            workspace_usage.span_count_since_reset as spans_this_month,
-            workspace_usage.step_count as total_steps,
-            workspace_usage.step_count_since_reset as steps_this_month,
-            subscription_tiers.spans as spans_limit,
-            subscription_tiers.steps as steps_limit,
-            GREATEST(
-                workspace_usage.span_count_since_reset - subscription_tiers.spans, 
-                0
-            ) as spans_over_limit,
-            GREATEST(
-                workspace_usage.span_count_since_reset - subscription_tiers.spans,
-                0
-            )::float8 * subscription_tiers.extra_span_price as spans_over_limit_cost,
-            GREATEST(
-                workspace_usage.step_count_since_reset - subscription_tiers.steps,
-                0
-            ) as steps_over_limit,
-            GREATEST(
-                workspace_usage.step_count_since_reset - subscription_tiers.steps,
-                0
-            )::float8 * subscription_tiers.extra_step_price as steps_over_limit_cost,
-            members_per_workspace.members,
-            subscription_tiers.members_per_workspace +
-                workspaces.additional_seats as members_limit,
-            subscription_tiers.storage_mib as storage_limit,
-            workspace_usage.reset_time
-        FROM
-            workspace_usage
-            JOIN members_per_workspace
-                ON members_per_workspace.workspace_id = workspace_usage.workspace_id
-            JOIN workspaces ON workspaces.id = workspace_usage.workspace_id
-            JOIN subscription_tiers ON subscription_tiers.id = workspaces.tier_id",
-    )
-    .bind(workspace_id)
-    .fetch_one(pool)
-    .await?;
-
-    Ok(workspace_stats)
-}
-
 pub async fn is_workspace_over_limit(
     pool: &PgPool,
     workspace_id: &Uuid,

app-server/src/evaluators/mod.rs

@@ -61,20 +61,14 @@ pub async fn inner_process_evaluators(
     client: Arc<reqwest::Client>,
     python_online_evaluator_url: &str,
 ) {
-    let mut receiver = match queue
+    let mut receiver = queue
         .get_receiver(
             EVALUATORS_QUEUE,
             EVALUATORS_EXCHANGE,
             EVALUATORS_ROUTING_KEY,
         )
         .await
-    {
-        Ok(receiver) => receiver,
-        Err(e) => {
-            log::error!("Failed to get receiver for evaluator queue: {:?}", e);
-            return;
-        }
-    };
+        .unwrap();
 
     while let Some(delivery) = receiver.receive().await {
         if let Err(e) = delivery {

app-server/src/main.rs

@@ -533,11 +533,6 @@ fn main() -> anyhow::Result<()> {
                                 .service(routes::workspace::get_workspace)
                                 .service(routes::workspace::create_workspace),
                         )
-                        .service(
-                            web::scope("/api/v1/limits")
-                                .wrap(auth.clone())
-                                .service(routes::limits::get_workspace_stats),
-                        )
                         .service(
                             // auth on path projects/{project_id} is handled by middleware on Next.js
                             web::scope("/api/v1/projects/{project_id}")