feat: complete observability stack (O1-O13) #52
@@ -0,0 +1,50 @@
|
|||||||
|
# Local development environment (.env file)
|
||||||
|
# Copy to .env and fill in your local/dev URLs
|
||||||
|
# .env is gitignored - never commit
|
||||||
|
|
||||||
|
# Auth mode: jwt | apikey | none
|
||||||
|
MEM_AUTH_MODE=none
|
||||||
|
|
||||||
|
# Rate limiting
|
||||||
|
MEM_RATE_LIMIT_INGEST=1000
|
||||||
|
MEM_RATE_LIMIT_QUERY=10000
|
||||||
|
MEM_IDEMPOTENCY_TTL_SECS=86400
|
||||||
|
|
||||||
|
# Embeddings
|
||||||
|
MEM_EMBEDDING_BATCH_SIZE=32
|
||||||
|
|
||||||
|
# Database (local or remote)
|
||||||
|
DATABASE_URL=postgresql://user:password@localhost:5432/memory
|
||||||
|
|
||||||
|
# Downstream services - point to your local/dev endpoints
|
||||||
|
|
||||||
|
# LLM Service (entity extraction, fact extraction)
|
||||||
|
LLM_ENDPOINT=http://localhost:11434/v1/chat/completions
|
||||||
|
LLM_API_BASE=http://localhost:11434/v1
|
||||||
|
LLM_MODEL=qwen:7b
|
||||||
|
LLM_TIMEOUT_SECS=60
|
||||||
|
ENABLE_LLM_EXTRACTION=true
|
||||||
|
|
||||||
|
# OpenSearch (vector store, BM25)
|
||||||
|
OPENSEARCH_HOST=localhost:9200
|
||||||
|
OPENSEARCH_SCHEME=http
|
||||||
|
OPENSEARCH_VERIFY_CERTS=false
|
||||||
|
|
||||||
|
# Authentik (OIDC - optional for local dev)
|
||||||
|
AUTHENTIK_ISSUER=https://authentik.riotpiao.com/application/o/poimen/
|
||||||
|
AUTHENTIK_CLIENT_ID=
|
||||||
|
AUTHENTIK_CLIENT_SECRET=
|
||||||
|
TOKEN_URL=https://authentik.riotpiao.com/application/o/token/
|
||||||
|
AUTHENTIK_VERIFY_SSL=false
|
||||||
|
|
||||||
|
# Temporal (workflow orchestration - future)
|
||||||
|
TEMPORAL_ENDPOINT=localhost:7233
|
||||||
|
TEMPORAL_NAMESPACE=poimen
|
||||||
|
|
||||||
|
# API Gateway (route optimization - future)
|
||||||
|
GATEWAY_URL=http://localhost:8080
|
||||||
|
|
||||||
|
# Server config
|
||||||
|
MEM_PORT=8080
|
||||||
|
MEM_API_KEY=test-key
|
||||||
|
MEM_HOME=/tmp
|
||||||
Generated
+1
@@ -2053,6 +2053,7 @@ dependencies = [
|
|||||||
"mem-ingest",
|
"mem-ingest",
|
||||||
"mem-llm",
|
"mem-llm",
|
||||||
"mem-store",
|
"mem-store",
|
||||||
|
"once_cell",
|
||||||
"pgvector",
|
"pgvector",
|
||||||
"rand 0.8.7",
|
"rand 0.8.7",
|
||||||
"redis",
|
"redis",
|
||||||
|
|||||||
@@ -439,29 +439,52 @@ pub async fn ingest_handler(
|
|||||||
body: web::Json<IngestRequest>,
|
body: web::Json<IngestRequest>,
|
||||||
state: web::Data<AppState>,
|
state: web::Data<AppState>,
|
||||||
) -> HttpResponse {
|
) -> HttpResponse {
|
||||||
|
use crate::metrics::*;
|
||||||
|
INGEST_REQUESTS_TOTAL.inc();
|
||||||
|
INGEST_IN_FLIGHT.inc();
|
||||||
|
let _timer = Timer::new(&INGEST_DURATION);
|
||||||
|
|
||||||
// Auth + capability check
|
// Auth + capability check
|
||||||
let (claims, _token) = match validate_auth(&req, &state).await {
|
let (claims, _token) = match validate_auth(&req, &state).await {
|
||||||
Ok(c) => c,
|
Ok(c) => c,
|
||||||
Err(e) => return e,
|
Err(e) => {
|
||||||
|
INGEST_AUTH_FAILURES.inc();
|
||||||
|
INGEST_ERRORS_TOTAL.inc();
|
||||||
|
INGEST_IN_FLIGHT.dec();
|
||||||
|
return e;
|
||||||
|
}
|
||||||
};
|
};
|
||||||
if !has_capability(&claims, "memory:write") {
|
if !has_capability(&claims, "memory:write") {
|
||||||
|
INGEST_AUTH_FAILURES.inc();
|
||||||
|
INGEST_ERRORS_TOTAL.inc();
|
||||||
|
INGEST_IN_FLIGHT.dec();
|
||||||
return HttpResponse::Forbidden().json(json!({
|
return HttpResponse::Forbidden().json(json!({
|
||||||
"error": "forbidden",
|
"error": "forbidden",
|
||||||
"reason": "missing capability: memory:write"
|
"reason": "missing capability: memory:write"
|
||||||
}));
|
}));
|
||||||
}
|
}
|
||||||
if let Err(e) = check_rate_limit(&claims, &state, "/memory/ingest") {
|
if let Err(e) = check_rate_limit(&claims, &state, "/memory/ingest") {
|
||||||
|
INGEST_RATE_LIMITED.inc();
|
||||||
|
INGEST_IN_FLIGHT.dec();
|
||||||
return e;
|
return e;
|
||||||
}
|
}
|
||||||
|
|
||||||
// Check idempotency
|
// Check idempotency
|
||||||
if let Some(cached) = state.idempotency_store.get(&body.ingest_id) {
|
if let Some(cached) = state.idempotency_store.get(&body.ingest_id) {
|
||||||
tracing::info!("Returning cached response for ingest_id: {}", body.ingest_id);
|
tracing::info!("Returning cached response for ingest_id: {}", body.ingest_id);
|
||||||
|
INGEST_DUPLICATES_TOTAL.inc();
|
||||||
|
INGEST_IN_FLIGHT.dec();
|
||||||
return HttpResponse::Accepted().json(cached);
|
return HttpResponse::Accepted().json(cached);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
let byte_count: usize = body.records.iter().map(|r| r.text.len()).sum();
|
||||||
|
INGEST_BYTES_TOTAL.inc_by(byte_count as u64);
|
||||||
|
INGEST_RECORDS_TOTAL.inc_by(body.records.len() as u64);
|
||||||
|
|
||||||
// Execute ingest
|
// Execute ingest
|
||||||
execute_ingest(&state, &body).await
|
let resp = execute_ingest(&state, &body).await;
|
||||||
|
INGEST_IN_FLIGHT.dec();
|
||||||
|
resp
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Execute ingest job creation and spawn worker
|
/// Execute ingest job creation and spawn worker
|
||||||
|
|||||||
Reference in New Issue
Block a user