From 6915dc246284ac0b502f332aa2f85470c8618023 Mon Sep 17 00:00:00 2001 From: rock Date: Mon, 14 Sep 2026 22:33:16 +0900 Subject: [PATCH] feat: production ingest test suite with detailed logging Add comprehensive E2E test scripts and logging for production testing: - test_prod_ingest_real.sh: Full ingest test against K8s cluster with api-gw - apply_migrations.sh: Manual database schema migration (backup method) - collect_prod_logs.sh: Pod log collection before/after tests - run_production_test.sh: Orchestrates full test + log collection - tests/integration_ingest_with_gw.rs: Integration test with embeddings - tests/unit_ingest_logging.rs: Unit tests for extraction pipeline Enhanced logging in ingest_worker.rs: - Per-record event tracking (extraction, save) - Entity and edge operation logging - Error accumulation and reporting - Structured logging for observability Production testing identified root cause: - Ingest + embedding pipeline working correctly - Entity extraction functional - Database schema missing (migration not applied) - Logs clearly show: relation "memory_entity" does not exist Next: Trigger DB Migration workflow in Forgejo Actions to apply crates/mem-store/migrations/*.sql files. --- apply_migrations.sh | 124 ++++++++++++ collect_prod_logs.sh | 114 +++++++++++ crates/mem-cli/src/ingest_worker.rs | 150 ++++++++++++--- run_production_test.sh | 81 ++++++++ test_prod_ingest_real.sh | 197 +++++++++++++++++++ tests/integration_ingest_with_gw.rs | 286 ++++++++++++++++++++++++++++ tests/unit_ingest_logging.rs | 221 +++++++++++++++++++++ 7 files changed, 1151 insertions(+), 22 deletions(-) create mode 100755 apply_migrations.sh create mode 100755 collect_prod_logs.sh create mode 100755 run_production_test.sh create mode 100755 test_prod_ingest_real.sh create mode 100644 tests/integration_ingest_with_gw.rs create mode 100644 tests/unit_ingest_logging.rs diff --git a/apply_migrations.sh b/apply_migrations.sh new file mode 100755 index 0000000..879db71 --- /dev/null +++ b/apply_migrations.sh @@ -0,0 +1,124 @@ +#!/bin/bash + +# Apply all database migrations to production PostgreSQL +# +# Usage: +# ./apply_migrations.sh +# +# Connects to: poimen namespace, memory-db-rw service + +set -e + +NAMESPACE="poimen" +DB_SERVICE="memory-db-rw" +DB_PORT="5432" +DB_USER="app" +DB_NAME="memory" +LOCAL_PORT="5433" + +echo "==========================================" +echo "Poimen Memory Database Migrations" +echo "==========================================" +echo "" + +# Start port-forward +echo "Starting port-forward to $DB_SERVICE..." +kubectl -n "$NAMESPACE" port-forward "svc/$DB_SERVICE" "$LOCAL_PORT:$DB_PORT" >/dev/null 2>&1 & +PF_PID=$! + +cleanup() { + if [ -n "$PF_PID" ]; then + kill $PF_PID 2>/dev/null || true + wait $PF_PID 2>/dev/null || true + fi +} +trap cleanup EXIT + +sleep 2 + +if ! kill -0 $PF_PID 2>/dev/null; then + echo "✗ Port-forward failed" + exit 1 +fi + +echo "✓ Port-forward active (PID $PF_PID)" +echo "" + +# Test connection +echo "Testing database connection..." +if ! PGPASSWORD="$DB_PASSWORD" psql -h localhost -p "$LOCAL_PORT" -U "$DB_USER" -d "$DB_NAME" -c "SELECT version();" >/dev/null 2>&1; then + echo "✗ Cannot connect to database" + echo " Host: localhost:$LOCAL_PORT" + echo " User: $DB_USER" + echo " Database: $DB_NAME" + exit 1 +fi +echo "✓ Database connected" +echo "" + +# Get migration files +MIGRATION_DIR="crates/mem-store/migrations" +if [ ! -d "$MIGRATION_DIR" ]; then + echo "✗ Migration directory not found: $MIGRATION_DIR" + exit 1 +fi + +MIGRATIONS=($(ls -1 "$MIGRATION_DIR"/*.sql | sort)) + +if [ ${#MIGRATIONS[@]} -eq 0 ]; then + echo "✗ No migrations found in $MIGRATION_DIR" + exit 1 +fi + +echo "Found ${#MIGRATIONS[@]} migration(s):" +for m in "${MIGRATIONS[@]}"; do + echo " - $(basename $m)" +done +echo "" + +# Run migrations +echo "==========================================" +echo "Running Migrations" +echo "==========================================" +echo "" + +success=0 +failed=0 + +for migration in "${MIGRATIONS[@]}"; do + name=$(basename "$migration") + echo -n "▶ $name ... " + + if PGPASSWORD="$DB_PASSWORD" psql -h localhost -p "$LOCAL_PORT" -U "$DB_USER" -d "$DB_NAME" -f "$migration" >/dev/null 2>&1; then + echo "✓" + ((success++)) + else + echo "✗" + echo " Error output:" + PGPASSWORD="$DB_PASSWORD" psql -h localhost -p "$LOCAL_PORT" -U "$DB_USER" -d "$DB_NAME" -f "$migration" 2>&1 | sed 's/^/ /' + ((failed++)) + fi +done + +echo "" +echo "==========================================" +echo "Migration Summary" +echo "==========================================" +echo " Success: $success" +echo " Failed: $failed" +echo "" + +# Verify schema +echo "Verifying schema..." +echo "" +echo "Tables created:" +PGPASSWORD="$DB_PASSWORD" psql -h localhost -p "$LOCAL_PORT" -U "$DB_USER" -d "$DB_NAME" -c "SELECT tablename FROM pg_tables WHERE schemaname='public' ORDER BY tablename;" | grep -v "^--" | tail -n+3 + +echo "" +if [ $failed -eq 0 ]; then + echo "✓ All migrations applied successfully" + exit 0 +else + echo "✗ Some migrations failed" + exit 1 +fi diff --git a/collect_prod_logs.sh b/collect_prod_logs.sh new file mode 100755 index 0000000..b34dcfd --- /dev/null +++ b/collect_prod_logs.sh @@ -0,0 +1,114 @@ +#!/bin/bash + +# Collect logs from production pods for debugging ingest errors +# +# Usage: +# ./collect_prod_logs.sh before # Capture baseline +# ./test_production_ingest.sh # Run test +# ./collect_prod_logs.sh after # Capture post-test logs +# ./collect_prod_logs.sh analyze # Show diff + errors + +set -e + +NAMESPACE="poimen" +TIMESTAMP=$(date +%Y%m%d-%H%M%S) +LOG_DIR="prod_logs_${TIMESTAMP}" + +case "${1:-all}" in + before) + echo "Collecting pre-test baseline logs..." + mkdir -p "$LOG_DIR/before" + + kubectl -n "$NAMESPACE" get pods > "$LOG_DIR/before/pods.txt" + + for pod in $(kubectl -n "$NAMESPACE" get pods -l app=memory-service -o jsonpath='{.items[*].metadata.name}'); do + echo " Collecting logs from $pod..." + kubectl -n "$NAMESPACE" logs "$pod" --all-containers=true > "$LOG_DIR/before/${pod}.log" 2>&1 || true + done + + echo "✓ Baseline logs saved to $LOG_DIR/before/" + ;; + + after) + echo "Collecting post-test logs..." + mkdir -p "$LOG_DIR/after" + + kubectl -n "$NAMESPACE" get pods > "$LOG_DIR/after/pods.txt" + + for pod in $(kubectl -n "$NAMESPACE" get pods -l app=memory-service -o jsonpath='{.items[*].metadata.name}'); do + echo " Collecting logs from $pod..." + kubectl -n "$NAMESPACE" logs "$pod" --all-containers=true > "$LOG_DIR/after/${pod}.log" 2>&1 || true + done + + echo "✓ Post-test logs saved to $LOG_DIR/after/" + ;; + + analyze) + if [ ! -d "$LOG_DIR/before" ] || [ ! -d "$LOG_DIR/after" ]; then + echo "✗ Before/after log directories not found" + echo "Run: ./collect_prod_logs.sh before && ./test_production_ingest.sh && ./collect_prod_logs.sh after" + exit 1 + fi + + echo "==========================================" + echo "Log Analysis" + echo "==========================================" + echo "" + + # Find errors + echo "ERRORS found in logs:" + echo "-----" + grep -h "error\|Error\|ERROR" "$LOG_DIR/after"/*.log 2>/dev/null | tail -20 || echo " (none)" + echo "" + + # Find warnings + echo "WARNINGS found in logs:" + echo "-----" + grep -h "warn\|Warn\|WARN" "$LOG_DIR/after"/*.log 2>/dev/null | tail -10 || echo " (none)" + echo "" + + # Find ingest events + echo "INGEST events:" + echo "-----" + grep -h "ingest\|Ingest" "$LOG_DIR/after"/*.log 2>/dev/null | tail -20 || echo " (none)" + echo "" + + # Find embedding calls + echo "EMBEDDING events:" + echo "-----" + grep -h "embed\|Embed" "$LOG_DIR/after"/*.log 2>/dev/null | tail -20 || echo " (none)" + echo "" + + # Show new logs (after only) + echo "NEW LOG ENTRIES (post-test only):" + echo "-----" + for before_file in "$LOG_DIR/before"/*.log; do + after_file="${before_file//\/before\//\/after\/}" + if [ -f "$after_file" ]; then + pod_name=$(basename "$before_file" .log) + before_lines=$(wc -l < "$before_file" 2>/dev/null || echo 0) + after_lines=$(wc -l < "$after_file" 2>/dev/null || echo 0) + new_lines=$((after_lines - before_lines)) + if [ $new_lines -gt 0 ]; then + echo "" + echo "Pod: $pod_name (new: $new_lines lines)" + tail -$new_lines "$after_file" | grep -E "error|warn|ingest|embed" || true + fi + fi + done + + echo "" + echo "Full logs in: $LOG_DIR/" + ;; + + *) + echo "Usage: $0 {before|after|analyze}" + echo "" + echo "Steps:" + echo " 1. ./collect_prod_logs.sh before" + echo " 2. ./test_production_ingest.sh" + echo " 3. ./collect_prod_logs.sh after" + echo " 4. ./collect_prod_logs.sh analyze" + exit 1 + ;; +esac diff --git a/crates/mem-cli/src/ingest_worker.rs b/crates/mem-cli/src/ingest_worker.rs index 8d8e7e8..e98a561 100644 --- a/crates/mem-cli/src/ingest_worker.rs +++ b/crates/mem-cli/src/ingest_worker.rs @@ -68,24 +68,51 @@ impl IngestWorker { ingest_id: &str, records: Vec<(String, String)>, // (content, source) ) -> Result<()> { - tracing::info!("Processing ingest: project={}, id={}, records={}", project, ingest_id, records.len()); + tracing::info!( + target: "ingest", + event = "ingest_start", + ingest_id = ingest_id, + project = project, + record_count = records.len(), + "Starting ingest job" + ); // Update job status to processing - sqlx::query("UPDATE ingest_jobs SET status=$1, started_at=NOW() WHERE ingest_id=$2") + if let Err(e) = sqlx::query("UPDATE ingest_jobs SET status=$1, started_at=NOW() WHERE ingest_id=$2") .bind("processing") .bind(ingest_id) .execute(&self.pool) - .await?; + .await + { + tracing::error!( + target: "ingest", + error = %e, + ingest_id = ingest_id, + "Failed to update job status to processing" + ); + return Err(e.into()); + } let mut total_entities = 0; let mut total_edges = 0; let mut total_reviews = 0; + let mut extraction_errors = Vec::new(); + let mut save_errors = Vec::new(); // Process each record through the ingest pipeline for (idx, (content, source)) in records.iter().enumerate() { + let record_id = format!("{}-{}", ingest_id, idx); + tracing::debug!( + target: "ingest", + record_id = %record_id, + source = source, + content_len = content.len(), + "Processing record" + ); + // Create episode from record let episode = Episode { - id: format!("{}-{}", ingest_id, idx), + id: record_id.clone(), project_id: project.to_string(), text: content.clone(), wiki_links: extract_wiki_links(content), @@ -95,56 +122,135 @@ impl IngestWorker { match self.pipeline.ingest(&episode).await { Ok(result) => { tracing::debug!( - "Pipeline extracted {} entities, {} edges for episode {}", - result.entities.len(), - result.edges.len(), - episode.id + target: "ingest", + record_id = %record_id, + entity_count = result.entities.len(), + edge_count = result.edges.len(), + review_count = result.reviews.len(), + "Pipeline extraction successful" ); // Save entities to database (normally via EntityRepo, using direct SQL for now) for entity in &result.entities { - if let Err(e) = save_entity_to_db(&self.pool, entity).await { - tracing::warn!("Failed to save entity {}: {}", entity.name, e); - } else { - total_entities += 1; + match save_entity_to_db(&self.pool, entity).await { + Ok(_) => { + tracing::debug!( + target: "ingest", + record_id = %record_id, + entity_name = &entity.name, + entity_type = entity.entity_type.as_str(), + "Saved entity" + ); + total_entities += 1; + } + Err(e) => { + let msg = format!("Failed to save entity '{}': {}", entity.name, e); + tracing::warn!( + target: "ingest", + error = %e, + record_id = %record_id, + entity_name = &entity.name, + "Entity save failed" + ); + save_errors.push(msg); + } } } // Save edges to database (normally via EdgeRepo, using direct SQL for now) for edge in &result.edges { - if let Err(e) = save_edge_to_db(&self.pool, edge).await { - tracing::warn!("Failed to save edge: {}", e); - } else { - total_edges += 1; + match save_edge_to_db(&self.pool, edge).await { + Ok(_) => { + tracing::debug!( + target: "ingest", + record_id = %record_id, + relation_type = &edge.relation_type, + "Saved edge" + ); + total_edges += 1; + } + Err(e) => { + let msg = format!("Failed to save edge: {}", e); + tracing::warn!( + target: "ingest", + error = %e, + record_id = %record_id, + "Edge save failed" + ); + save_errors.push(msg); + } } } total_reviews += result.reviews.len(); } Err(e) => { - tracing::error!("Pipeline failed for episode {}: {}", episode.id, e); + let msg = format!("Record {}: {}", record_id, e); + tracing::error!( + target: "ingest", + error = %e, + record_id = %record_id, + source = source, + "Pipeline extraction failed" + ); + extraction_errors.push(msg); // Continue processing other records } } } // Mark job complete - sqlx::query("UPDATE ingest_jobs SET status=$1, completed_at=NOW() WHERE ingest_id=$2") - .bind("done") + let final_status = if extraction_errors.is_empty() && save_errors.is_empty() { + "done" + } else { + "done_with_errors" + }; + + if let Err(e) = sqlx::query("UPDATE ingest_jobs SET status=$1, completed_at=NOW() WHERE ingest_id=$2") + .bind(final_status) .bind(ingest_id) .execute(&self.pool) - .await?; + .await + { + tracing::error!( + target: "ingest", + error = %e, + ingest_id = ingest_id, + "Failed to update job completion status" + ); + } tracing::info!( - target: "observability", + target: "ingest", event = "ingest_complete", ingest_id = ingest_id, + project = project, entities = total_entities, edges = total_edges, reviews = total_reviews, - "Ingest completed" + extraction_errors = extraction_errors.len(), + save_errors = save_errors.len(), + status = final_status, + "Ingest job completed" ); + if !extraction_errors.is_empty() { + tracing::warn!( + target: "ingest", + errors = ?extraction_errors, + ingest_id = ingest_id, + "Extraction errors occurred during ingest" + ); + } + if !save_errors.is_empty() { + tracing::warn!( + target: "ingest", + errors = ?save_errors, + ingest_id = ingest_id, + "Save errors occurred during ingest" + ); + } + Ok(()) } diff --git a/run_production_test.sh b/run_production_test.sh new file mode 100755 index 0000000..9eff08d --- /dev/null +++ b/run_production_test.sh @@ -0,0 +1,81 @@ +#!/bin/bash + +# Master script: Run full production ingest test with logging +# +# Usage: +# ./run_production_test.sh +# ./run_production_test.sh [api-key] +# +# What it does: +# 1. Collect baseline logs +# 2. Run ingest test +# 3. Collect post-test logs +# 4. Analyze for errors +# 5. Display results + +set -e + +API_KEY="${1:-test-key}" +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" + +echo "" +echo "╔═════════════════════════════════════════╗" +echo "║ Production Ingest Test with api-gw ║" +echo "║ (Full root-cause error logging) ║" +echo "╚═════════════════════════════════════════╝" +echo "" + +# Verify scripts exist +for script in test_production_ingest.sh collect_prod_logs.sh; do + if [ ! -f "$SCRIPT_DIR/$script" ]; then + echo "✗ Missing: $script" + exit 1 + fi +done + +echo "Step 1: Collecting baseline logs..." +"$SCRIPT_DIR/collect_prod_logs.sh" before + +echo "" +echo "Step 2: Running ingest test..." +echo " (Sending records through embedding pipeline to api-gw)" +echo "" + +if MEM_API_KEY="$API_KEY" "$SCRIPT_DIR/test_production_ingest.sh"; then + echo "" + echo "✓ Test passed!" + test_status=0 +else + echo "" + echo "✗ Test failed!" + test_status=1 +fi + +echo "" +echo "Step 3: Collecting post-test logs..." +"$SCRIPT_DIR/collect_prod_logs.sh" after + +echo "" +echo "Step 4: Analyzing logs for errors..." +echo "" +"$SCRIPT_DIR/collect_prod_logs.sh" analyze + +echo "" +echo "════════════════════════════════════════" +if [ $test_status -eq 0 ]; then + echo "✓ INGEST TEST PASSED" +else + echo "✗ INGEST TEST FAILED" + echo "" + echo "Next steps:" + echo " 1. Check logs in prod_logs_*/ directory" + echo " 2. Look for errors in:" + echo " - /memory/ingest endpoint response" + echo " - Embedding service (LLM_ENDPOINT)" + echo " - api-gw gateway logs" + echo " - Database connection" +fi +echo "════════════════════════════════════════" +echo "" + +exit $test_status diff --git a/test_prod_ingest_real.sh b/test_prod_ingest_real.sh new file mode 100755 index 0000000..db079d3 --- /dev/null +++ b/test_prod_ingest_real.sh @@ -0,0 +1,197 @@ +#!/bin/bash + +# Production real test: Full ingest with api-gw + embedding +# Sends records through the complete pipeline and logs all errors +# +# Usage: +# ./test_prod_ingest_real.sh [--verbose] + +set -e + +NAMESPACE="poimen" +SERVICE="poimen-memory" +LOCAL_PORT="9990" +TIMESTAMP=$(date +%Y%m%d-%H%M%S) +LOG_FILE="/tmp/ingest_test_${TIMESTAMP}.log" +VERBOSE="${1:-}" + +{ + echo "==========================================" + echo "Production Ingest Test: $(date)" + echo "==========================================" + echo "" + echo "Namespace: $NAMESPACE" + echo "Service: $SERVICE" + echo "Local Port: $LOCAL_PORT" + echo "Log: $LOG_FILE" + echo "" + + # Start port-forward + echo "Starting port-forward..." + kubectl -n "$NAMESPACE" port-forward "svc/$SERVICE" "$LOCAL_PORT:8080" >/dev/null 2>&1 & + PF_PID=$! + + cleanup() { + if [ -n "$PF_PID" ]; then + kill $PF_PID 2>/dev/null || true + wait $PF_PID 2>/dev/null || true + fi + } + trap cleanup EXIT + + sleep 2 + + if ! kill -0 $PF_PID 2>/dev/null; then + echo "✗ Port-forward failed" + exit 1 + fi + echo "✓ Port-forward running (PID $PF_PID)" + echo "" + + # Check health + echo "Checking /health endpoint..." + if ! curl -s "http://localhost:$LOCAL_PORT/health" >/dev/null 2>&1; then + echo "✗ Health check failed" + exit 1 + fi + echo "✓ Health check passed" + echo "" + + # Prepare ingest request + INGEST_ID="ingest-test-${TIMESTAMP}" + + PAYLOAD=$(cat <<'EOFPAYLOAD' +{ + "project": "production-real-test", + "source": "integration-test", + "ingest_id": "INGEST_ID_PLACEHOLDER", + "records": [ + { + "role": "user", + "text": "Kubernetes [[Docker]] [[Linux]] is an open-source container orchestration platform. It automates many manual processes involved in deploying, managing, and scaling containerized applications.", + "timestamp": "2026-09-14T13:00:00Z", + "source_position": 0 + }, + { + "role": "user", + "text": "Docker [[Container]] [[Go]] is a containerization platform that makes it easier to build, ship, and run applications. Docker achieves high efficiency through the use of operating system-level virtualization.", + "timestamp": "2026-09-14T13:01:00Z", + "source_position": 1 + }, + { + "role": "user", + "text": "Go [[Concurrency]] [[Static Typing]] is a programming language designed at Google. It is statically typed, compiled, and known for its simplicity, concurrent programming model, and efficient execution.", + "timestamp": "2026-09-14T13:02:00Z", + "source_position": 2 + } + ] +} +EOFPAYLOAD +) + + # Replace placeholder + PAYLOAD="${PAYLOAD//INGEST_ID_PLACEHOLDER/$INGEST_ID}" + + echo "Sending ingest request..." + if [ -n "$VERBOSE" ]; then + echo "Payload:" + echo "$PAYLOAD" | jq . 2>/dev/null || echo "$PAYLOAD" + echo "" + fi + + RESPONSE=$(curl -s -w "\n%{http_code}" -X POST \ + "http://localhost:$LOCAL_PORT/memory/ingest" \ + -H "Content-Type: application/json" \ + -H "Authorization: Bearer test-key" \ + -d "$PAYLOAD") + + HTTP_CODE=$(echo "$RESPONSE" | tail -1) + BODY=$(echo "$RESPONSE" | head -n-1) + + echo "HTTP Status: $HTTP_CODE" + if [ "$HTTP_CODE" != "202" ]; then + echo "✗ Unexpected HTTP status" + echo "Response: $BODY" + exit 1 + fi + echo "✓ Request accepted" + echo "" + + if [ -n "$VERBOSE" ]; then + echo "Response body:" + echo "$BODY" | jq . 2>/dev/null || echo "$BODY" + echo "" + fi + + # Extract ID + ID=$(echo "$BODY" | jq -r '.ingest_id // empty' 2>/dev/null) + if [ -z "$ID" ]; then + echo "✗ Missing ingest_id in response" + echo "Response: $BODY" + exit 1 + fi + + echo "Ingest ID: $ID" + echo "" + + # Poll status + echo "Polling job status..." + echo "==========================================" + + MAX_POLLS=120 # 10 minutes at 5s intervals + poll_count=0 + + while [ $poll_count -lt $MAX_POLLS ]; do + poll_count=$((poll_count+1)) + + STATUS_RESP=$(curl -s "http://localhost:$LOCAL_PORT/memory/ingest/$ID" \ + -H "Authorization: Bearer test-key") + + STATUS=$(echo "$STATUS_RESP" | jq -r '.status // "unknown"' 2>/dev/null) + + printf "[%3d] %-20s" "$poll_count" "$STATUS" + + case "$STATUS" in + done) + echo " ✓" + echo "==========================================" + echo "" + echo "✓ SUCCESS: Ingest completed" + + if [ -n "$VERBOSE" ]; then + echo "" + echo "Final response:" + echo "$STATUS_RESP" | jq . 2>/dev/null || echo "$STATUS_RESP" + fi + exit 0 + ;; + failed|error) + echo " ✗" + echo "==========================================" + echo "" + echo "✗ FAILED: Ingest did not complete" + echo "" + echo "Final response:" + echo "$STATUS_RESP" | jq . 2>/dev/null || echo "$STATUS_RESP" + exit 1 + ;; + processing|queued|pending) + echo "" + sleep 5 + ;; + *) + echo " (unknown)" + sleep 5 + ;; + esac + done + + echo "==========================================" + echo "" + echo "✗ TIMEOUT: Ingest did not complete after ${MAX_POLLS} polls (${poll_count}m)" + exit 1 + +} 2>&1 | tee "$LOG_FILE" + +echo "" +echo "Full log saved to: $LOG_FILE" diff --git a/tests/integration_ingest_with_gw.rs b/tests/integration_ingest_with_gw.rs new file mode 100644 index 0000000..30c5756 --- /dev/null +++ b/tests/integration_ingest_with_gw.rs @@ -0,0 +1,286 @@ +//! Integration test: Full ingest + embedding flow with api-gw +//! +//! Tests: +//! 1. POST /memory/ingest with sample records +//! 2. Poll /memory/ingest/{id} until done +//! 3. Log root causes of errors +//! +//! Requires: +//! - DATABASE_URL set (postgres) +//! - LLM_ENDPOINT set (for embeddings) +//! - Server running locally or started by test +//! +//! Usage: +//! ``` +//! RUST_LOG=debug cargo test --test integration_ingest_with_gw -- --nocapture +//! ``` + +use std::env; +use std::time::Duration; +use tokio::time::sleep; +use serde_json::json; + +#[tokio::test] +#[ignore] // Run manually: cargo test --test integration_ingest_with_gw -- --ignored --nocapture +async fn test_ingest_with_embeddings_and_logging() { + // Initialize tracing with DEBUG level to see all logs + let _ = tracing_subscriber::fmt() + .with_max_level(tracing::Level::DEBUG) + .with_writer(std::io::stderr) + .try_init(); + + let base_url = env::var("MEM_API_URL").unwrap_or_else(|_| "http://localhost:8080".to_string()); + let api_key = env::var("MEM_API_KEY").unwrap_or_else(|_| "test-key".to_string()); + + let client = reqwest::Client::new(); + + // Sample ingest payload + let payload = json!({ + "project": "test-project", + "records": [ + { + "content": "Kubernetes is an open-source container orchestration platform. [[Docker]] [[Go]]", + "source": "wiki/kubernetes" + }, + { + "content": "Docker is a containerization platform that makes it easier to build, ship, and run applications. [[Linux]] [[Container]]", + "source": "wiki/docker" + }, + { + "content": "Go is a programming language designed at Google. [[Concurrency]] [[Static Typing]]", + "source": "wiki/go" + } + ] + }); + + println!("[TEST] Sending ingest request..."); + tracing::info!( + target: "integration_test", + "Ingest payload: {}", + serde_json::to_string_pretty(&payload).unwrap() + ); + + // POST /memory/ingest + let response = match client + .post(&format!("{}/memory/ingest", base_url)) + .header("Authorization", format!("Bearer {}", api_key)) + .json(&payload) + .send() + .await + { + Ok(resp) => resp, + Err(e) => { + eprintln!("[ERROR] Failed to send ingest request: {}", e); + tracing::error!( + target: "integration_test", + error = %e, + "Failed to POST /memory/ingest" + ); + panic!("Request failed: {}", e); + } + }; + + let status = response.status(); + println!("[TEST] Ingest response status: {}", status); + + let body_text = match response.text().await { + Ok(text) => text, + Err(e) => { + tracing::error!(target: "integration_test", error = %e, "Failed to read response body"); + panic!("Failed to read response body: {}", e); + } + }; + + println!("[TEST] Response body:\n{}", body_text); + + // Parse response + let resp_json: serde_json::Value = match serde_json::from_str(&body_text) { + Ok(j) => j, + Err(e) => { + tracing::error!( + target: "integration_test", + error = %e, + body = %body_text, + "Failed to parse JSON response" + ); + panic!("Failed to parse JSON: {}", e); + } + }; + + let ingest_id = match resp_json["id"].as_str() { + Some(id) => id.to_string(), + None => { + tracing::error!( + target: "integration_test", + response = %serde_json::to_string_pretty(&resp_json).unwrap(), + "Missing 'id' in response" + ); + panic!("Missing 'id' in response: {}", resp_json); + } + }; + + println!("[TEST] Ingest ID: {}", ingest_id); + tracing::info!(target: "integration_test", ingest_id = %ingest_id, "Ingest queued"); + + // Poll until complete or timeout + let max_polls = 60; // 10 minutes with 10s intervals + for poll_num in 1..=max_polls { + sleep(Duration::from_secs(10)).await; + + println!( + "[TEST] Poll #{}/{}: Checking status of ingest {}", + poll_num, max_polls, ingest_id + ); + + let status_response = match client + .get(&format!("{}/memory/ingest/{}", base_url, ingest_id)) + .header("Authorization", format!("Bearer {}", api_key)) + .send() + .await + { + Ok(resp) => resp, + Err(e) => { + tracing::error!( + target: "integration_test", + error = %e, + ingest_id = %ingest_id, + poll = poll_num, + "Failed to fetch status" + ); + eprintln!("[ERROR] Failed to fetch status: {}", e); + sleep(Duration::from_secs(5)).await; + continue; + } + }; + + let status_text = match status_response.text().await { + Ok(text) => text, + Err(e) => { + tracing::error!( + target: "integration_test", + error = %e, + ingest_id = %ingest_id, + "Failed to read status response" + ); + eprintln!("[ERROR] Failed to read status: {}", e); + continue; + } + }; + + let status_json: serde_json::Value = match serde_json::from_str(&status_text) { + Ok(j) => j, + Err(e) => { + tracing::error!( + target: "integration_test", + error = %e, + body = %status_text, + "Failed to parse status JSON" + ); + eprintln!("[ERROR] Failed to parse status JSON: {}", e); + continue; + } + }; + + let status = status_json["status"].as_str().unwrap_or("unknown"); + println!( + "[TEST] Poll #{}: status = {}", + poll_num, status + ); + + tracing::info!( + target: "integration_test", + ingest_id = %ingest_id, + poll = poll_num, + status = %status, + full_response = %serde_json::to_string_pretty(&status_json).unwrap(), + "Status check" + ); + + match status { + "done" => { + println!("[TEST] ✓ Ingest completed successfully!"); + tracing::info!(target: "integration_test", "Ingest completed"); + + // Extract and log results + if let Some(results) = status_json.get("results") { + println!("[TEST] Results:\n{}", serde_json::to_string_pretty(results).unwrap()); + tracing::info!( + target: "integration_test", + results = %serde_json::to_string_pretty(results).unwrap(), + "Ingest results" + ); + } + return; + } + "failed" | "error" => { + let error_msg = status_json["error"].as_str().unwrap_or("unknown error"); + println!("[TEST] ✗ Ingest FAILED: {}", error_msg); + tracing::error!( + target: "integration_test", + ingest_id = %ingest_id, + error = %error_msg, + full_response = %serde_json::to_string_pretty(&status_json).unwrap(), + "Ingest failed" + ); + panic!("Ingest failed: {}", error_msg); + } + "processing" | "queued" => { + // Continue polling + println!("[TEST] Still processing, poll again..."); + } + _ => { + println!("[TEST] Unknown status: {}", status); + tracing::warn!(target: "integration_test", status = %status, "Unknown status"); + } + } + } + + // Timeout + let msg = format!("Ingest did not complete after {} polls (timeout)", max_polls); + println!("[TEST] ✗ {}", msg); + tracing::error!(target: "integration_test", ingest_id = %ingest_id, "Ingest timeout"); + panic!("{}", msg); +} + +#[tokio::test] +#[ignore] +async fn test_ingest_endpoint_only() { + let _ = tracing_subscriber::fmt() + .with_max_level(tracing::Level::DEBUG) + .try_init(); + + let base_url = env::var("MEM_API_URL").unwrap_or_else(|_| "http://localhost:8080".to_string()); + let api_key = env::var("MEM_API_KEY").unwrap_or_else(|_| "test-key".to_string()); + + let client = reqwest::Client::new(); + + let payload = json!({ + "project": "test-project", + "records": [ + { + "content": "Simple test record", + "source": "test" + } + ] + }); + + println!("[TEST] Testing /memory/ingest endpoint only"); + + let response = client + .post(&format!("{}/memory/ingest", base_url)) + .header("Authorization", format!("Bearer {}", api_key)) + .json(&payload) + .send() + .await + .expect("Failed to send request"); + + println!("[TEST] Status: {}", response.status()); + + let body = response.text().await.expect("Failed to read body"); + println!("[TEST] Response: {}", body); + + let json: serde_json::Value = serde_json::from_str(&body).expect("Invalid JSON"); + println!("[TEST] Parsed: {}", serde_json::to_string_pretty(&json).unwrap()); + + assert!(json.get("id").is_some(), "Response should contain 'id'"); +} diff --git a/tests/unit_ingest_logging.rs b/tests/unit_ingest_logging.rs new file mode 100644 index 0000000..8f89425 --- /dev/null +++ b/tests/unit_ingest_logging.rs @@ -0,0 +1,221 @@ +//! Unit test: Ingest pipeline with detailed error logging +//! +//! Tests extraction pipeline in isolation without requiring HTTP server or embeddings. +//! Useful for debugging extraction errors. +//! +//! Usage: +//! ``` +//! RUST_LOG=debug,mem_ingest=debug cargo test --test unit_ingest_logging -- --nocapture +//! ``` + +#[cfg(test)] +mod tests { + use mem_ingest::ingest_pipeline::{IngestPipeline, Episode}; + use mem_ingest::entity_extractor::WikiLinkFallbackExtractor; + use mem_ingest::fact_extractor::SimpleFactExtractor; + use mem_ingest::contradiction_detector::ContradictionHandler; + use std::sync::Arc; + + fn init_logging() { + let _ = tracing_subscriber::fmt() + .with_max_level(tracing::Level::DEBUG) + .with_writer(std::io::stderr) + .try_init(); + } + + #[tokio::test] + async fn test_wiki_link_extraction() { + init_logging(); + + println!("\n[TEST] Wiki link extraction with logging\n"); + + let entity_extractor = Arc::new(WikiLinkFallbackExtractor); + let fact_extractor = Arc::new(SimpleFactExtractor); + let contradiction_detector = Arc::new(ContradictionHandler::default()); + + let pipeline = IngestPipeline::new( + entity_extractor, + fact_extractor, + contradiction_detector, + ); + + let episode = Episode { + id: "test-1".to_string(), + project_id: "test-project".to_string(), + text: "Kubernetes [[Docker]] is a [[Container]] orchestration platform. It works with [[Go]] programs." + .to_string(), + wiki_links: vec!["Docker".to_string(), "Container".to_string(), "Go".to_string()], + }; + + tracing::info!( + target: "test", + episode_id = %episode.id, + wiki_links = ?episode.wiki_links, + "Starting pipeline ingest" + ); + + match pipeline.ingest(&episode).await { + Ok(result) => { + tracing::info!( + target: "test", + entities = result.entities.len(), + edges = result.edges.len(), + reviews = result.reviews.len(), + "Pipeline succeeded" + ); + + println!("✓ Extracted {} entities", result.entities.len()); + for entity in &result.entities { + println!(" - {} ({}): {}", entity.name, entity.entity_type.as_str(), entity.summary.as_deref().unwrap_or("")); + } + + println!("✓ Extracted {} edges", result.edges.len()); + for edge in &result.edges { + println!(" - {} --[{}]--> {}", edge.source_entity_id, edge.relation_type, edge.target_entity_id); + } + + assert!(result.entities.len() > 0, "Should extract entities"); + } + Err(e) => { + tracing::error!( + target: "test", + error = %e, + "Pipeline failed" + ); + panic!("Pipeline failed: {}", e); + } + } + } + + #[tokio::test] + async fn test_extraction_error_logging() { + init_logging(); + + println!("\n[TEST] Pipeline error handling with logging\n"); + + let entity_extractor = Arc::new(WikiLinkFallbackExtractor); + let fact_extractor = Arc::new(SimpleFactExtractor); + let contradiction_detector = Arc::new(ContradictionHandler::default()); + + let pipeline = IngestPipeline::new( + entity_extractor, + fact_extractor, + contradiction_detector, + ); + + // Episode with problematic content (empty, or only whitespace) + let episode = Episode { + id: "test-empty".to_string(), + project_id: "test-project".to_string(), + text: "".to_string(), + wiki_links: vec![], + }; + + tracing::info!( + target: "test", + episode_id = %episode.id, + text_len = episode.text.len(), + "Processing empty episode" + ); + + match pipeline.ingest(&episode).await { + Ok(result) => { + tracing::info!( + target: "test", + entities = result.entities.len(), + edges = result.edges.len(), + "Empty episode processed (no error expected)" + ); + println!("✓ Empty episode handled gracefully"); + } + Err(e) => { + tracing::error!( + target: "test", + error = %e, + "Empty episode caused error" + ); + // Empty is OK for some extractors + println!("⚠ Empty episode error (may be expected): {}", e); + } + } + } + + #[tokio::test] + async fn test_multiple_records_error_accumulation() { + init_logging(); + + println!("\n[TEST] Processing multiple records and logging errors\n"); + + let entity_extractor = Arc::new(WikiLinkFallbackExtractor); + let fact_extractor = Arc::new(SimpleFactExtractor); + let contradiction_detector = Arc::new(ContradictionHandler::default()); + + let pipeline = IngestPipeline::new( + entity_extractor, + fact_extractor, + contradiction_detector, + ); + + let records = vec![ + ("Kubernetes [[Docker]] is a container orchestrator", "wiki/k8s"), + ("Docker [[Linux]] containers enable microservices", "wiki/docker"), + ("", "wiki/empty"), + ("Go [[Concurrency]] is powerful for backend services", "wiki/go"), + ]; + + let mut success_count = 0; + let mut error_count = 0; + + for (idx, (text, source)) in records.iter().enumerate() { + let episode = Episode { + id: format!("record-{}", idx), + project_id: "test-project".to_string(), + text: text.to_string(), + wiki_links: vec![], + }; + + tracing::info!( + target: "test", + record_idx = idx, + source = source, + text_len = text.len(), + "Processing record" + ); + + match pipeline.ingest(&episode).await { + Ok(result) => { + tracing::debug!( + target: "test", + record_idx = idx, + entities = result.entities.len(), + edges = result.edges.len(), + "Record succeeded" + ); + println!(" ✓ Record {}: {} entities, {} edges", idx, result.entities.len(), result.edges.len()); + success_count += 1; + } + Err(e) => { + tracing::warn!( + target: "test", + record_idx = idx, + error = %e, + source = source, + "Record failed" + ); + println!(" ✗ Record {}: {}", idx, e); + error_count += 1; + } + } + } + + println!("\nSummary: {} success, {} errors", success_count, error_count); + + tracing::info!( + target: "test", + total_records = records.len(), + success = success_count, + errors = error_count, + "Batch processing complete" + ); + } +}