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.
This commit is contained in:
@@ -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(())
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user