When our e-commerce platform launched an AI-powered customer service system handling 50,000 daily queries, we faced a critical challenge: understanding why certain RAG (Retrieval-Augmented Generation) responses were slow, expensive, or failing silently. Traditional logging gave us raw data, but we needed observability. This tutorial walks through the complete ELK (Elasticsearch, Logstash, Kibana) pipeline we built to monitor, analyze, and alert on every AI API call using HolySheep AI as our core LLM provider—with pricing at just $1 per million tokens versus the industry standard of $7.30, the cost savings alone justified building proper observability.
The Use Case: E-Commerce RAG System Monitoring
Our product catalog RAG system processes customer queries by retrieving relevant product information and generating natural language responses. We needed to track:
- Response latency across different product categories
- Token consumption patterns and cost optimization opportunities
- Failure rates and error categorization
- Retrieval quality correlation with response quality
HolySheep AI delivers sub-50ms latency globally, which gave us a performance baseline to measure against. The free credits on signup allowed us to develop and test this entire pipeline without immediate cost concerns.
Architecture Overview
The complete pipeline consists of four stages: log generation at the application layer, centralized collection via Filebeat, processing and enrichment in Logstash, and visualization with Kibana dashboards. Elasticsearch serves as the distributed search and analytics engine throughout.
Step 1: Structured Logging Implementation
First, we need application-level logging with consistent JSON structure. The following Python module creates standardized log entries for every HolySheep API call:
# holysheep_logger.py
import logging
import json
import uuid
from datetime import datetime
from typing import Optional, Dict, Any
from pythonjsonlogger import jsonlogger
import httpx
class HolySheepAPILogger:
"""Structured logging for HolySheep AI API calls with ELK integration."""
def __init__(self, service_name: str, log_level: int = logging.INFO):
self.service_name = service_name
self.logger = logging.getLogger(f"holysheep.{service_name}")
self.logger.setLevel(log_level)
# Prevent duplicate handlers
if not self.logger.handlers:
handler = logging.StreamHandler()
formatter = jsonlogger.JsonFormatter(
'%(asctime)s %(name)s %(levelname)s %(message)s',
datefmt='%Y-%m-%dT%H:%M:%S.%fZ'
)
handler.setFormatter(formatter)
self.logger.addHandler(handler)
def log_request(
self,
request_id: str,
model: str,
messages: list,
temperature: float,
max_tokens: int
) -> Dict[str, Any]:
"""Log outgoing API request."""
log_entry = {
"@timestamp": datetime.utcnow().isoformat() + "Z",
"event_type": "api_request",
"request_id": request_id,
"service": self.service_name,
"provider": "holysheep",
"model": model,
"input_tokens_estimate": self._estimate_tokens(messages),
"temperature": temperature,
"max_tokens": max_tokens,
"message_count": len(messages),
"environment": "production"
}
self.logger.info("API request initiated", extra=log_entry)
return log_entry
def log_response(
self,
request_id: str,
status_code: int,
response_time_ms: float,
output_tokens: int,
model: str,
error: Optional[str] = None
) -> Dict[str, Any]:
"""Log API response or error."""
log_entry = {
"@timestamp": datetime.utcnow().isoformat() + "Z",
"event_type": "api_response",
"request_id": request_id,
"service": self.service_name,
"provider": "holysheep",
"model": model,
"status_code": status_code,
"response_time_ms": response_time_ms,
"output_tokens": output_tokens,
"error": error,
"environment": "production"
}
if status_code == 200:
self.logger.info("API request completed", extra=log_entry)
else:
self.logger.error(f"API request failed: {error}", extra=log_entry)
return log_entry
def _estimate_tokens(self, messages: list) -> int:
"""Rough token estimation: ~4 characters per token for English."""
total_chars = sum(len(msg.get("content", "")) for msg in messages)
return total_chars // 4
Usage example
if __name__ == "__main__":
logger = HolySheepAPILogger("product-catalog-rag")
# Log a request
logger.log_request(
request_id=str(uuid.uuid4()),
model="deepseek-v3.2",
messages=[{"role": "user", "content": "Show me red running shoes under $100"}],
temperature=0.7,
max_tokens=500
)Step 2: HolySheep API Integration with Logging
Now we integrate the logger with actual HolySheep API calls. The base URL is https://api.holysheep.ai/v1 and their pricing is exceptional—DeepSeek V3.2 at $0.42 per million output tokens represents 85% savings compared to GPT-4.1 at $8.00:
# holysheep_client.py
import httpx
import time
import uuid
from typing import Optional, Dict, Any, List
from holysheep_logger import HolySheepAPILogger
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
class HolySheepRAGClient:
"""Production RAG client with comprehensive logging for ELK analysis."""
def __init__(self, api_key: str, service_name: str):
self.api_key = api_key
self.base_url = HOLYSHEEP_BASE_URL
self.logger = HolySheepAPILogger(service_name)
self.client = httpx.Client(
base_url=self.base_url,
timeout=30.0,
headers={"Authorization": f"Bearer {self.api_key}"}
)
def query(
self,
user_query: str,
retrieved_context: List[str],
model: str = "deepseek-v3.2",
temperature: float = 0.7,
max_tokens: int = 500
) -> Dict[str, Any]:
"""Execute RAG query with full logging."""
request_id = str(uuid.uuid4())
# Build messages with retrieved context
messages = [
{"role": "system", "content": "You are a helpful product assistant."},
{"role": "context", "content": "\n\n".join(retrieved_context)},
{"role": "user", "content": user_query}
]
# Log the request
self.logger.log_request(
request_id=request_id,
model=model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens
)
start_time = time.time()
try:
response = self.client.post(
"/chat/completions",
json={
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
)
response_time_ms = (time.time() - start_time) * 1000
if response.status_code == 200:
data = response.json()
output_tokens = data.get("usage", {}).get("completion_tokens", 0)
self.logger.log_response(
request_id=request_id,
status_code=200,
response_time_ms=response_time_ms,
output_tokens=output_tokens,
model=model
)
return {
"success": True,
"content": data["choices"][0]["message"]["content"],
"usage": data.get("usage", {}),
"response_time_ms": response_time_ms,
"request_id": request_id
}
else:
error_msg = response.text
self.logger.log_response(
request_id=request_id,
status_code=response.status_code,
response_time_ms=response_time_ms,
output_tokens=0,
model=model,
error=error_msg
)
return {"success": False, "error": error_msg}
except Exception as e:
response_time_ms = (time.time() - start_time) * 1000
self.logger.log_response(
request_id=request_id,
status_code=0,
response_time_ms=response_time_ms,
output_tokens=0,
model=model,
error=str(e)
)
return {"success": False, "error": str(e)}
def close(self):
self.client.close()
Production usage
if __name__ == "__main__":
client = HolySheepRAGClient(
api_key=HOLYSHEEP_API_KEY,
service_name="product-catalog-rag"
)
# Simulate RAG query
result = client.query(
user_query="What are the best running shoes for marathon training?",
retrieved_context=[
"Nike Pegasus 40: $120, 4.5 stars, excellent cushioning for long distance",
"Adidas Ultraboost 23: $190, 4.7 stars, premium energy return technology",
"Brooks Ghost 15: $140, 4.6 stars, ideal for neutral gaits"
],
model="deepseek-v3.2"
)
print(f"Query successful: {result.get('success')}")
print(f"Response: {result.get('content', result.get('error'))}")
client.close()Step 3: ELK Stack Configuration
Elasticsearch Index Template
Create an index template optimized for AI API log analysis:
# elasticsearch-template.json
{
"index_patterns": ["holysheep-api-logs-*"],
"template": {
"settings": {
"number_of_shards": 2,
"number_of_replicas": 1,
"index.lifecycle.name": "holysheep-logs-policy",
"index.lifecycle.rollover_alias": "holysheep-api-logs"
},
"mappings": {
"properties": {
"@timestamp": {"type": "date"},
"event_type": {"type": "keyword"},
"request_id": {"type": "keyword"},
"service": {"type": "keyword"},
"provider": {"type": "keyword"},
"model": {"type": "keyword"},
"status_code": {"type": "integer"},
"response_time_ms": {"type": "float"},
"input_tokens_estimate": {"type": "integer"},
"output_tokens": {"type": "integer"},
"temperature": {"type": "float"},
"max_tokens": {"type": "integer"},
"message_count": {"type": "integer"},
"error": {"type": "text"},
"environment": {"type": "keyword"}
}
}
}
}
Apply the template
curl -X PUT "localhost:9200/_index_template/holysheep-api-logs" \
-H "Content-Type: application/json" \
-d @elasticsearch-template.jsonLogstash Pipeline Configuration
# /etc/logstash/conf.d/holysheep-pipeline.conf
input {
beats {
port => 5044
host => "0.0.0.0"
}
}
filter {
# Parse JSON logs
json {
source => "message"
target => "parsed"
}
# Extract nested fields
if [parsed][@timestamp] {
mutate {
add_field => { "[@metadata][timestamp]" => "%{[parsed][@timestamp]}" }
}
}
# Calculate cost based on token usage
if [parsed][output_tokens] {
ruby {
code => '
# Pricing: DeepSeek V3.2 = $0.42/M tokens, others proportional
prices = {
"deepseek-v3.2" => 0.42,
"gpt-4.1" => 8.00,
"claude-sonnet-4.5" => 15.00,
"gemini-2.5-flash" => 2.50
}
model = event.get("[parsed][model]") || "deepseek-v3.2"
tokens = event.get("[parsed][output_tokens]") || 0
price_per_million = prices.fetch(model, 0.42)
cost_usd = (tokens.to_f / 1_000_000) * price_per_million
event.set("[parsed][estimated_cost_usd]", cost_usd.round(4))
'
}
}
# Categorize response times
if [parsed][response_time_ms] {
ruby {
code => '
rt = event.get("[parsed][response_time_ms]").to_f
category = case rt
when 0..50 then "ultra_fast"
when 51..200 then "fast"
when 201..500 then "normal"
when 501..1000 then "slow"
else "very_slow"
end
event.set("[parsed][latency_category]", category)
'
}
}
# Enrich with date components
date {
match => ["[@metadata][timestamp]", "ISO8601"]
target => "@timestamp"
}
# Clean up temporary fields
mutate {
remove_field => ["[@metadata][timestamp]", "host", "agent", "ecs", "input"]
}
}
output {
elasticsearch {
hosts => ["elasticsearch:9200"]
index => "holysheep-api-logs-%{+YYYY.MM.dd}"
document_type => "_doc"
}
# Debug output (remove in production)
stdout { codec => rubydebug }
}Filebeat Configuration
# /etc/filebeat/filebeat.yml
filebeat.inputs:
- type: log
enabled: true
paths:
- /var/log/holysheep/*.log
json.keys_under_root: true
json.add_error_key: true
json.message_key: message
fields:
environment: production
service_type: ai-api
fields_under_root: true
processors:
- add_host_metadata:
when.not.contains.tags: forwarded
- add_cloud_metadata: ~
- add_docker_metadata: ~
output.logstash:
hosts: ["logstash:5044"]
logging.level: info
logging.to_files: true
logging.files:
path: /var/log/filebeat
name: filebeat
keepfiles: 7
permissions: 0640Step 4: Kibana Dashboards and Visualizations
After data flows into Elasticsearch, create these essential visualizations:
Visualization 1: Response Time Distribution
# Kibana Vega-Lite specification for latency histogram
{
"$schema": "https://vega.github.io/schema/vega-lite/v5.json",
"title": "API Response Time Distribution (Last 24h)",
"width": "container",
"height": 300,
"layer": [
{
"mark": {"type": "area", "opacity": 0.4, "color": "#3182bd"},
"encoding": {
"x": {
"field": "response_time_ms",
"type": "quantitative",
"bin": {"maxbins": 50},
"title": "Response Time (ms)"
},
"y": {
"aggregate": "count",
"type": "quantitative",
"title": "Request Count"
}
}
},
{
"mark": {"type": "rule", "color": "#e24a33", "strokeWidth": 2},
"encoding": {
"x": {
"aggregate": "average",
"field": "response_time_ms"
},
"size": {"value": 40}
}
}
],
"transform": [{
"filter": "datum.event_type == 'api_response' && datum.status_code == 200"
}]
}Visualization 2: Cost Analysis by Model
# Kibana Lens formula for cost tracking
Metric: Total Estimated Cost
Sum(field="estimated_cost_usd")
Breakdown by Model (Data Table)
Aggregation: Terms
Field: model.keyword
Metrics:
- Sum: estimated_cost_usd
- Sum: output_tokens
- Average: response_time_ms
- Count: request_id
Cost Trend (Line Chart)
X-Axis: @timestamp (Date Histogram, 1 hour)
Y-Axis: Sum(estimated_cost_usd)
Split Series: model.keywordStep 5: Alerting Configuration
Set up Watcher alerts for critical conditions:
# Kibana Stack Management > Watcher > Create Watch
PUT _watcher/watch/holysheep_high_error_rate
{
"trigger": {
"schedule": {
"interval": "5m"
}
},
"input": {
"search": {
"request": {
"indices": ["holysheep-api-logs-*"],
"body": {
"query": {
"bool": {
"must": [
{"range": {"@timestamp": {"gte": "now-5m"}}},
{"term": {"event_type": "api_response"}}
]
}
},
"aggs": {
"by_status": {
"terms": {"field": "status_code"},
"aggs": {
"total_requests": {"value_count": {"field": "request_id"}}
}
},
"total": {"value_count": {"field": "request_id"}}
}
}
}
}
},
"condition": {
"script": {
"source": "
def resp = ctx.payload.aggregations.by_status.buckets;
def total = ctx.payload.aggregations.total.value;
for (bucket in resp) {
if (bucket.key >= 400 && (bucket.doc_count / total) > 0.05) {
return true;
}
}
return false;
"
}
},
"actions": {
"log_error_alert": {
"logging": {
"text": "High error rate detected: {{ctx.payload.aggregations.by_status.buckets}}"
}
},
"webhook_alert": {
"webhook": {
"scheme": "https",
"host": "hooks.slack.com",
"port": 443,
"method": "post",
"path": "/services/XXX/YYY/ZZZ",
"headers": {
"Content-Type": "application/json"
},
"body": "{\"text\": \"HolySheep API Error Alert: High error rate detected in product-catalog-rag service\"}"
}
}
}
}
Latency alert - notify when p99 > 500ms
PUT _watcher/watch/holysheep_high_latency
{
"trigger": {"schedule": {"interval": "1m"}},
"input": {
"search": {
"request": {
"indices": ["holysheep-api-logs-*"],
"body": {
"query": {
"bool": {
"must": [
{"range": {"@timestamp": {"gte": "now-5m"}}},
{"term": {"event_type": "api_response"}},
{"term": {"status_code": 200}}
]
}
}
},
"aggs": {
"latency_percentiles": {
"percentiles": {
"field": "response_time_ms",
"percents": [50, 90, 95, 99]
}
}
}
}
}
},
"condition": {
"script": {
"source": "return ctx.payload.aggregations.latency_percentiles.values['99.0'] > 500"
}
},
"actions": {
"webhook_alert": {
"webhook": {
"scheme": "https",
"host": "hooks.slack.com",
"port": 443,