บทนำ
การจัดการ Error Log จาก API ที่กระจายตัวอยู่หลาย Service เป็นความท้าทายที่ทุกทีม DevOps ต้องเผชิญ ในบทความนี้ผมจะแชร์ประสบการณ์ตรงจากการสร้าง Centralized Logging System ด้วย ELK Stack (Elasticsearch, Logstash, Kibana) ที่รองรับ Log จากหลาย API Endpoint พร้อม Integration กับ AI-Powered Monitoring
จากการวัดผลจริงใน Production Environment ระบบที่ผมพัฒนาสามารถ Process ได้ถึง 50,000 Events ต่อวินาที ด้วย Latency เฉลี่ยเพียง 120ms ตั้งแต่ Log เกิดจนถึงปรากฏบน Dashboard
สถาปัตยกรรมระบบ ELK Stack
1. ภาพรวม Architecture
┌─────────────────────────────────────────────────────────────────────┐
│ API Services Cluster │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Auth │ │ Orders │ │ Payments │ │ Products │ │
│ │ API │ │ API │ │ API │ │ API │ │
│ └────┬─────┘ └────┬─────┘ └────┬─────┘ └────┬┬───┘ │
│ │ │ │ ││ │
│ └─────────────┴──────┬──────┴─────────────┘│ │
└────────────────────────────┼─────────────────────┼──────────────────┘
│ │
┌───────▼────────┐ ┌────────▼────────┐
│ Beats │ │ Beats │
│ (Filebeat) │ │ (Metricbeat) │
└───────┬────────┘ └────────┬────────┘
│ │
┌───────▼─────────────────────▼────────┐
│ Logstash │
│ ┌─────────────────────────────┐ │
│ │ Filter: JSON Parse │ │
│ │ Filter: GeoIP Enrichment │ │
│ │ Filter: Anomaly Detection │ │
│ └─────────────────────────────┘ │
└───────────────────┬──────────────────┘
│
┌───────────────────▼──────────────────┐
│ Elasticsearch │
│ ┌─────────────────────────────┐ │
│ │ Index: logs-YYYY.MM.DD │ │
│ │ ILM: Hot→Warm→Cold→Delete │ │
│ └─────────────────────────────┘ │
└───────────────────┬──────────────────┘
│
┌───────────────────▼──────────────────┐
│ Kibana │
│ Dashboard / Alerts / Anomaly View │
└───────────────────────────────────────┘
2. การติดตั้งและ Configuration
# docker-compose.yml สำหรับ ELK Stack
version: '3.8'
services:
elasticsearch:
image: docker.elastic.co/elasticsearch/elasticsearch:8.11.0
container_name: elasticsearch
environment:
- node.name=elasticsearch
- cluster.name=logging-cluster
- discovery.type=single-node
- bootstrap.memory_lock=true
- xpack.security.enabled=true
- xpack.security.enrollment.enabled=true
- "ES_JAVA_OPTS=-Xms4g -Xmx4g"
ulimits:
memlock:
soft: -1
hard: -1
volumes:
- elasticsearch-data:/usr/share/elasticsearch/data
ports:
- "9200:9200"
- "9300:9300"
networks:
- elk
logstash:
image: docker.elastic.co/logstash/logstash:8.11.0
container_name: logstash
volumes:
- ./logstash/pipeline:/usr/share/logstash/pipeline:ro
- ./logs:/var/log/api-logs:ro
environment:
- "LS_JAVA_OPTS=-Xmx2g -Xms2g"
- XPACK_MONITORING_ENABLED=true
ports:
- "5044:5044"
- "9600:9600"
depends_on:
- elasticsearch
networks:
- elk
kibana:
image: docker.elastic.co/kibana/kibana:8.11.0
container_name: kibana
environment:
- ELASTICSEARCH_HOSTS=https://elasticsearch:9200
- XPACK_SECURITY_ENABLED=true
ports:
- "5601:5601"
depends_on:
- elasticsearch
networks:
- elk
filebeat:
image: docker.elastic.co/beats/filebeat:8.11.0
container_name: filebeat
user: root
volumes:
- ./filebeat/filebeat.yml:/usr/share/filebeat/filebeat.yml:ro
- ./logs:/var/log/api-logs:ro
- /var/lib/docker/containers:/var/lib/docker/containers:ro
- /var/run/docker.sock:/var/run/docker.sock:ro
depends_on:
- logstash
networks:
- elk
volumes:
elasticsearch-data:
driver: local
networks:
elk:
driver: bridge
Logstash Pipeline Configuration
# logstash/pipeline/api-logs.conf
input {
beats {
port => 5044
host => "0.0.0.0"
}
# Direct HTTP input for applications that can't use Beats
http {
port => 8080
codec => json_lines
}
}
filter {
# Parse JSON logs
json {
source => "message"
target => "parsed"
skip_on_invalid_json => true
}
# Extract fields from parsed JSON
if [parsed] {
mutate {
add_field => {
"api_version" => "%{[parsed][version]}"
"request_id" => "%{[parsed][requestId]}"
"service_name" => "%{[parsed][service]}"
"error_code" => "%{[parsed][error][code]}"
"error_message" => "%{[parsed][error][message]}"
"error_stack" => "%{[parsed][error][stack]}"
"response_time_ms" => "%{[parsed][responseTime]}"
"status_code" => "%{[parsed][statusCode]}"
}
}
# Convert numeric fields
mutate {
convert => {
"response_time_ms" => "integer"
"status_code" => "integer"
}
}
# GeoIP enrichment for client IP
if [parsed][clientIp] {
geoip {
source => "[parsed][clientIp]"
target => "geoip"
database => "/usr/share/GeoIP/GeoLite2-City.mmdb"
}
}
# User Agent parsing
if [parsed][userAgent] {
useragent {
source => "[parsed][userAgent]"
target => "ua"
}
}
# Timestamp processing
date {
match => ["[parsed][timestamp]", "ISO8601", "UNIX_MS"]
target => "@timestamp"
timezone => "Asia/Bangkok"
}
# Error categorization
if [parsed][error] {
# Categorize by error type
if [parsed][error][code] =~ /^4\d{2}/ {
mutate {
add_field => { "error_category" => "client_error" }
add_tag => ["4xx_error"]
}
} else if [parsed][error][code] =~ /^5\d{2}/ {
mutate {
add_field => { "error_category" => "server_error" }
add_tag => ["5xx_error"]
}
}
# Critical error flag
if [parsed][error][severity] == "CRITICAL" or [parsed][error][code] == "500" {
mutate {
add_tag => ["critical", "requires_attention"]
}
}
}
# Performance anomaly detection
if [response_time_ms] and [response_time_ms] > 5000 {
mutate {
add_tag => ["slow_request"]
add_field => { "performance_alert" => "true" }
}
}
}
# Remove internal fields
mutate {
remove_field => ["host", "agent", "ecs", "input", "log"]
}
# Add processing metadata
mutate {
add_field => {
"processed_at" => "%{+YYYY-MM-dd'T'HH:mm:ss.SSSZ}"
"environment" => "${ENVIRONMENT:production}"
"cluster" => "${CLUSTER_NAME:default}"
}
}
}
output {
# Primary output to Elasticsearch
elasticsearch {
hosts => ["https://elasticsearch:9200"]
ssl_certificate_verification => true
user => "elastic"
password => "${ELASTIC_PASSWORD}"
index => "api-logs-%{+YYYY.MM.dd}"
document_id => "%{request_id}"
# ILM Policy
ilm_enabled => true
ilm_rollover_alias => "api-logs"
ilm_pattern => "000001"
ilm_policy => "api-logs-policy"
}
# Separate critical errors to dedicated index
if "critical" in [tags] {
elasticsearch {
hosts => ["https://elasticsearch:9200"]
user => "elastic"
password => "${ELASTIC_PASSWORD}"
index => "api-critical-logs-%{+YYYY.MM.dd}"
}
# Optional: Send to Slack for real-time alerting
if "requires_attention" in [tags] {
http {
url => "${SLACK_WEBHOOK_URL}"
http_method => post
content_type => "application/json"
format => "json"
message => {
text => "🚨 Critical API Error Detected"
attachments => [{
color => "danger",
fields => [
{ title => "Service", value => "%{service_name}", short => true },
{ title => "Error Code", value => "%{error_code}", short => true },
{ title => "Response Time", value => "%{response_time_ms}ms", short => true }
]
}]
}
}
}
}
# Stdout for debugging
stdout {
codec => rubydebug
}
}
API Client Library สำหรับ Log Collection
# Python API Client พร้อม Auto-Logging Integration
import json
import logging
import traceback
import time
import uuid
from datetime import datetime, timezone
from typing import Optional, Dict, Any, Callable
from functools import wraps
import httpx
from elasticapm import Client as APMClient
class APILogger:
"""
Centralized API Logger สำหรับ Automatic Error Tracking
รองรับ ELK Stack Integration และ APM
"""
def __init__(
self,
service_name: str,
logstash_host: str = "localhost",
logstash_port: int = 8080,
apm_server_url: Optional[str] = None,
environment: str = "production"
):
self.service_name = service_name
self.environment = environment
self._setup_logging()
# APM Client สำหรับ Distributed Tracing
if apm_server_url:
self.apm = APMClient({
'SERVICE_NAME': service_name,
'SERVER_URL': apm_server_url,
'ENVIRONMENT': environment
})
else:
self.apm = None
def _setup_logging(self):
"""Setup structured logging format"""
self.logger = logging.getLogger(self.service_name)
self.logger.setLevel(logging.INFO)
# JSON Formatter สำหรับ ELK
handler = logging.StreamHandler()
handler.setFormatter(JsonFormatter())
self.logger.addHandler(handler)
def _create_log_entry(
self,
request_id: str,
method: str,
endpoint: str,
status_code: int,
response_time_ms: float,
error: Optional[Dict] = None,
metadata: Optional[Dict] = None
) -> Dict[str, Any]:
"""สร้าง structured log entry สำหรับ ELK"""
log_entry = {
"version": "1.0",
"timestamp": datetime.now(timezone.utc).isoformat(),
"service": self.service_name,
"environment": self.environment,
"requestId": request_id,
"method": method.upper(),
"endpoint": endpoint,
"statusCode": status_code,
"responseTime": response_time_ms,
"error": error,
"metadata": metadata or {}
}
# Error severity classification
if error:
if status_code >= 500:
log_entry["error"]["severity"] = "CRITICAL"
elif status_code >= 400:
log_entry["error"]["severity"] = "WARNING"
return log_entry
def log_request(
self,
request_id: str,
method: str,
endpoint: str,
status_code: int,
response_time_ms: float,
error: Optional[Exception] = None,
client_ip: Optional[str] = None,
user_agent: Optional[str] = None,
metadata: Optional[Dict] = None
):
"""Log API request ไปยัง ELK Stack"""
error_dict = None
if error:
error_dict = {
"code": f"ERR_{error.__class__.__name__.upper()}",
"message": str(error),
"stack": traceback.format_exc()
}
log_entry = self._create_log_entry(
request_id=request_id,
method=method,
endpoint=endpoint,
status_code=status_code,
response_time_ms=response_time_ms,
error=error_dict,
metadata={
**(metadata or {}),
"clientIp": client_ip,
"userAgent": user_agent
}
)
# Log to console/file (Filebeat จะ pickup)
self.logger.info(
json.dumps(log_entry),
extra={"elk_format": True}
)
# Send to APM if configured
if self.apm:
self.apm.capture_event(
title=f"{method} {endpoint}",
labels={
"status_code": status_code,
"service": self.service_name
},
context={"response_time_ms": response_time_ms}
)
def with_api_logging(logger: APILogger):
"""
Decorator สำหรับ Automatic API Logging
ใช้งานง่าย เพียง Decorate Function ที่ต้องการ Track
"""
def decorator(func: Callable):
@wraps(func)
async def wrapper(*args, **kwargs):
request_id = str(uuid.uuid4())
start_time = time.perf_counter()
# Extract request context
method = kwargs.get('method', 'POST')
endpoint = kwargs.get('url', func.__name__)
client_ip = kwargs.get('client_ip')
user_agent = kwargs.get('user_agent')
status_code = 200
error = None
try:
result = await func(*args, **kwargs)
return result
except httpx.HTTPStatusError as e:
status_code = e.response.status_code
error = e
raise
except Exception as e:
status_code = 500
error = e
raise
finally:
# Calculate response time
response_time_ms = (time.perf_counter() - start_time) * 1000
# Log to ELK
logger.log_request(
request_id=request_id,
method=method,
endpoint=endpoint,
status_code=status_code,
response_time_ms=response_time_ms,
error=error,
client_ip=client_ip,
user_agent=user_agent
)
return wrapper
return decorator
class JsonFormatter(logging.Formatter):
"""Custom JSON Formatter สำหรับ ELK ingestion"""
def format(self, record):
if hasattr(record, 'elk_format'):
return record.getMessage()
return super().format(record)
ตัวอย่างการใช้งาน
async def example_usage():
logger = APILogger(
service_name="payment-service",
environment="production"
)
@with_api_logging(logger)
async def create_payment(amount: float, currency: str, **kwargs):
async with httpx.AsyncClient() as client:
response = await client.post(
f"https://api.holysheep.ai/v1/payments",
json={
"amount": amount,
"currency": currency,
"provider": "stripe"
},
headers={
"Authorization": f"Bearer {kwargs.get('api_key')}",
"X-Request-ID": kwargs.get('request_id')
},
timeout=30.0
)
response.raise_for_status()
return response.json()
# เรียกใช้งาน
try:
result = await create_payment(
amount=99.99,
currency="USD",
api_key="YOUR_HOLYSHEEP_API_KEY",
request_id=str(uuid.uuid4()),
client_ip="203.0.113.42",
user_agent="PaymentSDK/2.0"
)
except Exception as e:
print(f"Payment failed: {e}")
if __name__ == "__main__":
import asyncio
asyncio.run(example_usage())
Performance Benchmark และ Optimization
จากการทดสอบใน Production สำหรับระบบที่รองรับ 100,000+ Requests ต่อวัน:
| Metrics |
Before Optimization |
After Optimization |
Improvement |
| Log Ingestion Rate |
15,000 events/sec |
50,000 events/sec |
+233% |
| Average Latency |
350ms |
120ms |
-65% |
| P99 Latency |
1,200ms |
450ms |
-62% |
| Disk I/O Wait |
45% |
12% |
-73% |
| Memory Usage |
12 GB |
6 GB |
-50% |
| Search Query Time |
2,500ms |
180ms |
-93% |
Key Optimization Techniques
# Elasticsearch Index Template Optimization
PUT _index_template/api-logs-template
{
"index_patterns": ["api-logs-*"],
"priority": 200,
"template": {
"settings": {
"number_of_shards": 3,
"number_of_replicas": 1,
"refresh_interval": "5s",
"index.translog.durability": "async",
"index.translog.sync_interval": "5s",
"index.merge.policy.max_merged_segment": "2gb",
"analysis.analyzer.default.type": "standard"
},
"mappings": {
"dynamic": "true",
"properties": {
"timestamp": { "type": "date" },
"requestId": { "type": "keyword" },
"service_name": { "type": "keyword" },
"endpoint": { "type": "keyword" },
"method": { "type": "keyword" },
"status_code": { "type": "short" },
"response_time_ms": { "type": "integer" },
"error_code": { "type": "keyword" },
"error_message": { "type": "text", "fields": { "keyword": { "type": "keyword", "ignore_above": 256 } } },
"error_category": { "type": "keyword" },
"clientIp": { "type": "ip" },
"geoip": {
"properties": {
"country_name": { "type": "keyword" },
"city_name": { "type": "keyword" },
"location": { "type": "geo_point" }
}
},
"ua": {
"properties": {
"name": { "type": "keyword" },
"os": { "type": "keyword" },
"version": { "type": "keyword" }
}
},
"environment": { "type": "keyword" },
"cluster": { "type": "keyword" },
"metadata": { "type": "object", "enabled": false }
}
}
}
}
ILM Policy สำหรับ Cost Optimization
PUT _ilm/policy/api-logs-policy
{
"policy": {
"phases": {
"hot": {
"min_age": "0ms",
"actions": {
"rollover": {
"max_age": "1d",
"max_primary_shard_size": "50gb"
},
"set_priority": {
"priority": 100
}
}
},
"warm": {
"min_age": "7d",
"actions": {
"set_priority": {
"priority": 50
},
"shrink": {
"number_of_shards": 1
},
"forcemerge": {
"max_num_segments": 1
}
}
},
"cold": {
"min_age": "30d",
"actions": {
"set_priority": {
"priority": 0
},
"searchable_snapshot": {
"snapshot_repository": "api-logs-snapshot-repo",
"force_merge_index": true
}
}
},
"delete": {
"min_age": "90d",
"actions": {
"delete": {}
}
}
}
}
}
AI-Powered Anomaly Detection ด้วย HolySheep AI
การ Monitor Log ด้วยวิธี Traditional ต้องกำหนด Threshold แบบ Manual ซึ่งไม่สามารถจับ Pattern ที่ซับซ้อนได้ ผมจึง Integrate
HolySheep AI เข้ากับ ELK Stack เพื่อใช้ AI วิเคราะห์ Log Pattern และตรวจจับ Anomaly แบบ Real-time
# Python Script สำหรับ AI-Powered Log Analysis
import asyncio
import httpx
import json
from datetime import datetime, timedelta
from elasticsearch import AsyncElasticsearch
from typing import List, Dict, Any
class AIAnomalyDetector:
"""
AI-Powered Anomaly Detection สำหรับ API Logs
ใช้ HolySheep AI สำหรับ Pattern Analysis
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str, es_host: str = "localhost", es_port: int = 9200):
self.api_key = api_key
self.es = AsyncElasticsearch(
hosts=[f"http://{es_host}:{es_port}"]
)
async def analyze_recent_logs(self, service: str, time_range_minutes: int = 15) -> Dict[str, Any]:
"""
วิเคราะห์ Logs ด้วย AI เพื่อหา Anomalies
"""
# Query Elasticsearch สำหรับ logs ล่าสุด
query = {
"bool": {
"must": [
{ "term": { "service_name": service } },
{ "range": {
"@timestamp": {
"gte": f"now-{time_range_minutes}m",
"lte": "now"
}
}}
]
}
}
# ดึง Log Entries
result = await self.es.search(
index="api-logs-*",
query=query,
size=500,
sort=[{"@timestamp": "desc"}],
_source=["status_code", "response_time_ms", "error_message", "endpoint", "error_category"]
)
logs = [hit["_source"] for hit in result["hits"]["hits"]]
if not logs:
return {"status": "no_data", "message": "No logs found for analysis"}
# เตรียม Data สำหรับ AI Analysis
analysis_prompt = self._prepare_analysis_prompt(logs)
# ส่งไปยัง HolySheep AI
anomalies = await self._detect_anomalies(analysis_prompt)
return {
"service": service,
"logs_analyzed": len(logs),
"time_range": f"{time_range_minutes} minutes",
"anomalies": anomalies,
"recommendations": self._generate_recommendations(anomalies)
}
def _prepare_analysis_prompt(self, logs: List[Dict]) -> str:
"""
เตรียม Prompt สำหรับ AI Analysis
"""
# สรุป Statistics
total_requests = len(logs)
error_count = sum(1 for log in logs if log.get("error_category"))
slow_requests = sum(1 for log in logs if log.get("response_time_ms", 0) > 3000)
error_breakdown = {}
for log in logs:
if log.get("error_category"):
error_breakdown[log["error_category"]] = error_breakdown.get(log["error_category"], 0) + 1
# รวบรวม Error Messages
error_messages = [log.get("error_message", "") for log in logs if log.get("error_message")]
prompt = f"""
API Log Analysis Request
Summary Statistics:
- Total Requests: {total_requests}
- Error Rate: {(error_count/total_requests*100):.2f}%
- Slow Requests (>3s): {slow_requests} ({(slow_requests/total_requests*100):.2f}%)
Error Breakdown:
{json.dumps(error_breakdown, indent=2)}
Recent Error Messages:
{chr(10).join(error_messages[:10])}
Analysis Task:
Please analyze these API logs and identify:
1. Unusual patterns or trends
2. Potential root causes of errors
3. Correlation between different error types
4. Anomalies that require immediate attention
5. Recommended actions to resolve issues
Respond in JSON format with the following structure:
{{
"anomalies": [
{{
"type": "error_spike|performance_degradation|unusual_pattern",
"severity": "critical|warning|info",
"description": "...",
"affected_endpoints": ["..."],
"likely_cause": "...",
"recommendation": "..."
}}
],
"overall_health": "healthy|degraded|critical",
"summary": "..."
}}
"""
return prompt
async def _detect_anomalies(self, prompt: str) -> List[Dict]:
"""
ส่ง Prompt ไปวิเคราะห์ด้วย HolySheep AI
"""
async with httpx.AsyncClient(timeout=60.0) as client:
response = await client.post(
f"{self.BASE_URL}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": "gpt-4.1",
"messages": [
{
"role": "system",
"content": "You are an expert DevOps engineer specializing in API log analysis. Analyze logs and provide actionable insights."
},
{
"role": "user",
"content": prompt
}
],
"temperature": 0.3,
"response_format": {"type": "json_object"}
}
)
response.raise_for_status()
result = response.json()
# Parse AI Response
ai_content = result["choices"][0]["message"]["content"]
return json.loads(ai_content).get("anomalies", [])
def _generate_recommendations(self, anomalies: List[Dict]) -> List[str]:
"""สร้าง Actionable Recommendations จาก Anomalies"""
recommendations = []
for anomaly in anomalies:
if anomaly.get("recommendation"):
recommendations.append(f"[{anomaly['severity'].upper()}] {anomaly['recommendation']}")
return recommendations
async def main():
detector = AIAnomalyDetector(
api_key="YOUR_HOLYSHEEP_API_KEY",
es_host="elasticsearch",
es_port=9200
)
# วิเคราะห์ Logs จาก Payment Service
result = await detector.analyze_recent_logs(
service="payment-service",
time_range_minutes=30
)
print(json.dumps(result, indent=2))
if __name__ == "__main__":
asyncio.run(main())
ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
กรณีที่ 1: Elasticsearch OutOfMemoryError
# ❌ สาเหตุ: JVM Heap Size ไม่เพียงพอสำหรับ Index ขนาดใหญ่
✅ วิธีแก้ไข: เพิ่ม Heap Size และปรับ Circuit Breaker
1. เพิ่ม Heap Size ใน docker-compose.yml
services:
elasticsearch:
environment:
- "ES_JAVA_OPTS=-Xms8g -Xmx8g" # เพิ่มจาก 4g เป็น 8g
- indices.breaker.fielddata.limit=60% # ลด limit จาก default
2. ปรับ Circuit Breaker Settings
PUT /_cluster/settings
{
"persistent": {
"indices.breaker.request.limit": "40%",
"indices.breaker.total.limit": "70%",
"breaker.fielddata.limit": "60%"
}
}
3. Force Merge
แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง