Trong bài viết này, tôi sẽ chia sẻ cách tôi đã xây dựng một hệ thống phân tích log API Gateway hoàn chỉnh, giúp phát hiện và ngăn chặn các cuộc tấn công bất thường trước khi chúng gây ra thiệt hại nghiêm trọng cho hệ thống.
Bối Cảnh Thực Tế: Khi Hệ Thống Bị Tấn Công
3 tháng trước, vào lúc 2:47 AM, tôi nhận được alert khẩn cấp từ hệ thống monitoring: Error 401 Unauthorized xuất hiện với tần suất 1,247 lần/giây từ một IP lạ. Đó là lúc tôi nhận ra website đang bị brute-force attack. Nếu không có hệ thống phân tích log tự động, có lẽ tôi đã mất 30-40 phút để phát hiện và xử lý - đủ thời gian để kẻ tấn công truy cập thành công.
Sau sự cố đó, tôi đã xây dựng một pipeline phân tích log API Gateway sử dụng HolySheep AI với chi phí chỉ bằng 15% so với giải pháp enterprise truyền thống.
Kiến Trúc Tổng Quan
+------------------+ +------------------+ +------------------+
| API Gateway |---->| Log Collector |---->| Analysis Engine |
| (nginx/envoy) | | (fluentd) | | (Python + ML) |
+------------------+ +------------------+ +------------------+
|
v
+------------------+ +------------------+
| Alert System |<----| HolySheep AI |
| (Slack/Email) | | (Anomaly Det.) |
+------------------+ +------------------+
|
v
+------------------+ +------------------+
| Auto Blocking |<----| Threat Intel |
| (iptables/fw) | | (IP Database) |
+------------------+ +------------------+Cài Đặt Môi Trường Và Thu Thập Log
Đầu tiên, chúng ta cần cài đặt các công cụ cần thiết. Tôi sử dụng Python 3.11+ với các thư viện chuyên dụng cho việc phân tích log và gọi API.
# Cài đặt các thư viện cần thiết
pip install python-json-logger fluent-logger pandas numpy scikit-learn
pip install requests redis opensearch-py
Cấu hình môi trường
export HOLYSHEEP_API_KEY="YOUR_HOLYSHEEP_API_KEY"
export HOLYSHEEP_BASE_URL="https://api.holysheep.ai/v1"
export REDIS_HOST="localhost"
export REDIS_PORT="6379"Module Thu Thập Và Parse Log
Tôi đã viết một module Python hoàn chỉnh để thu thập log từ nhiều nguồn khác nhau, bao gồm nginx, Envoy, và các API Gateway phổ biến khác.
import json
import re
from datetime import datetime
from typing import Dict, List, Optional
from collections import defaultdict
import requests
class APILogParser:
"""
Parser phân tích log từ nhiều nguồn: nginx, envoy, API Gateway
Trích xuất các trường: IP, endpoint, status_code, response_time, timestamp
"""
# Regex pattern cho nginx access log
NGINX_PATTERN = re.compile(
r'(?P[\d\.]+) - - \[(?P[^\]]+)\] '
r'"(?P\w+) (?P\S+) (?P\S+)" '
r'(?P\d+) (?P\d+) '
r'"(?P[^"]*)" "(?P[^"]*)"'
)
# Regex pattern cho JSON log (envoy, cloudflare)
JSON_PATTERN = re.compile(r'\{.*\}')
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self.threat_cache = {} # Cache kết quả kiểm tra IP
def parse_nginx_log(self, log_line: str) -> Optional[Dict]:
"""Parse dòng log nginx"""
match = self.NGINX_PATTERN.match(log_line)
if not match:
return None
data = match.groupdict()
return {
'ip': data['ip'],
'method': data['method'],
'path': data['path'],
'status': int(data['status']),
'size': int(data['size']),
'timestamp': self._parse_timestamp(data['timestamp']),
'user_agent': data['user_agent'],
'referrer': data['referrer']
}
def parse_json_log(self, log_line: str) -> Optional[Dict]:
"""Parse log định dạng JSON"""
if not self.JSON_PATTERN.match(log_line):
return None
try:
data = json.loads(log_line)
return {
'ip': data.get('client_ip', data.get('remote_addr')),
'method': data.get('method', 'UNKNOWN'),
'path': data.get('path', data.get('uri')),
'status': int(data.get('status', data.get('response_code', 0))),
'response_time': float(data.get('response_time', 0)),
'timestamp': self._parse_timestamp(
data.get('timestamp', datetime.utcnow().isoformat())
),
'user_agent': data.get('user_agent', 'Unknown'),
'request_id': data.get('request_id', '')
}
except (json.JSONDecodeError, KeyError, ValueError):
return None
def _parse_timestamp(self, timestamp_str: str) -> datetime:
"""Parse timestamp từ nhiều định dạng khác nhau"""
formats = [
'%d/%b/%Y:%H:%M:%S +0000',
'%Y-%m-%dT%H:%M:%S.%fZ',
'%Y-%m-%d %H:%M:%S',
'%d-%b-%Y %H:%M:%S'
]
for fmt in formats:
try:
return datetime.strptime(timestamp_str, fmt)
except ValueError:
continue
return datetime.utcnow()
def analyze_with_holysheep(self, log_data: Dict) -> Dict:
"""
Gọi HolySheep AI để phân tích log entry và phát hiện bất thường
Chi phí: DeepSeek V3.2 chỉ $0.42/M tokens - tiết kiệm 85%+
"""
prompt = f"""
Phân tích log entry sau và xác định xem có phải là request bất thường không:
IP: {log_data.get('ip')}
Method: {log_data.get('method')}
Path: {log_data.get('path')}
Status: {log_data.get('status')}
Response Time: {log_data.get('response_time', 0)}ms
User Agent: {log_data.get('user_agent', 'Unknown')}
Trả về JSON: {{
"is_anomaly": true/false,
"threat_level": "low/medium/high/critical",
"attack_type": "brute_force/scanner/dos/normal/...",
"reason": "mô tả ngắn gọn lý do"
}}
"""
try:
response = requests.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": "deepseek-v3.2",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.1,
"max_tokens": 200
},
timeout=5 # Timeout 5s để không block real-time processing
)
if response.status_code == 200:
result = response.json()
content = result['choices'][0]['message']['content']
return json.loads(content)
else:
return {"is_anomaly": False, "threat_level": "unknown"}
except requests.exceptions.Timeout:
# Fallback: sử dụng rule-based detection nếu API timeout
return self._rule_based_detection(log_data)
except Exception as e:
print(f"Lỗi khi gọi HolySheep AI: {e}")
return {"is_anomaly": False, "threat_level": "unknown"}
def _rule_based_detection(self, log_data: Dict) -> Dict:
"""Fallback detection sử dụng rules - không cần API call"""
ip = log_data.get('ip', '')
path = log_data.get('path', '')
status = log_data.get('status', 0)
# Rule 1: Brute force - nhiều request 401/403
if status in [401, 403]:
return {
"is_anomaly": True,
"threat_level": "high",
"attack_type": "brute_force",
"reason": "Authentication failure detected"
}
# Rule 2: Scanner - request đến paths bất thường
suspicious_paths = ['/admin', '/.env', '/config.php', '/wp-login.php']
if any(p in path.lower() for p in suspicious_paths):
return {
"is_anomaly": True,
"threat_level": "medium",
"attack_type": "scanner",
"reason": "Suspicious path access"
}
# Rule 3: Slowloris - response time cao bất thường
if log_data.get('response_time', 0) > 30000: # >30s
return {
"is_anomaly": True,
"threat_level": "high",
"attack_type": "slowloris",
"reason": "Abnormally high response time"
}
return {"is_anomaly": False, "threat_level": "low"}
Sử dụng parser
parser = APILogParser(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
Parse sample log
sample_log = '192.168.1.100 - - [15/Jan/2026:02:47:23 +0000] "POST /api/login HTTP/1.1" 401 512 "-" "python-requests/2.28.0"'
parsed = parser.parse_nginx_log(sample_log)
print(f"Parsed log: {parsed}")
Phân tích với AI
result = parser.analyze_with_holysheep(parsed)
print(f"Analysis result: {result}") Hệ Thống Phát Hiện Tấn Công Thời Gian Thực
Đây là phần quan trọng nhất - module phát hiện tấn công theo thời gian thực sử dụng sliding window và statistical analysis.
import threading
import time
from collections import deque
from dataclasses import dataclass, field
from typing import Dict, List
import redis
@dataclass
class ThreatRecord:
"""Lưu trữ thông tin về mối đe dọa"""
ip: str
attack_type: str
threat_level: str
request_count: int
first_seen: datetime
last_seen: datetime
blocked: bool = False
class RealtimeThreatDetector:
"""
Detector phát hiện tấn công theo thời gian thực
Sử dụng sliding window để đếm request và phát hiện anomalies
"""
# Ngưỡng cấu hình
THRESHOLDS = {
'requests_per_minute': 100, # >100 req/min = suspicious
'auth_failures_per_minute': 10, # >10 failures/min = brute force
'error_rate_percent': 50, # >50% errors = possible attack
'unique_paths_per_minute': 50, # Scanner detection
'response_time_p95_ms': 5000 # Slow attack detection
}
def __init__(self, redis_client: redis.Redis):
self.redis = redis_client
self.ip_stats = defaultdict(lambda: {
'requests': deque(maxlen=1000),
'auth_failures': deque(maxlen=100),
'errors': deque(maxlen=100),
'paths': set(),
'response_times': deque(maxlen=1000)
})
self.blocked_ips = set()
self.lock = threading.Lock()
def process_request(self, log_entry: Dict) -> Optional[ThreatRecord]:
"""Xử lý mỗi request và kiểm tra có phải threat không"""
ip = log_entry.get('ip')
if not ip or ip in self.blocked_ips:
return None
now = datetime.utcnow()
stats = self.ip_stats[ip]
# Cập nhật statistics
with self.lock:
stats['requests'].append(now)
stats['response_times'].append(log_entry.get('response_time', 0))
status = log_entry.get('status', 200)
if status in [401, 403]:
stats['auth_failures'].append(now)
if status >= 400:
stats['errors'].append(now)
stats['paths'].add(log_entry.get('path', ''))
# Kiểm tra các loại tấn công
threats = []
# 1. Check Brute Force
if self._check_brute_force(ip, stats):
threats.append(ThreatRecord(
ip=ip,
attack_type='brute_force',
threat_level='critical',
request_count=len(stats['auth_failures']),
first_seen=stats['auth_failures'][0] if stats['auth_failures'] else now,
last_seen=now
))
# 2. Check DoS/DDoS
if self._check_dos(ip, stats):
threats.append(ThreatRecord(
ip=ip,
attack_type='dos',
threat_level='high',
request_count=self._count_requests_last_minute(ip, stats),
first_seen=stats['requests'][0] if stats['requests'] else now,
last_seen=now
))
# 3. Check Scanner
if self._check_scanner(ip, stats):
threats.append(ThreatRecord(
ip=ip,
attack_type='scanner',
threat_level='medium',
request_count=len(stats['paths']),
first_seen=stats['requests'][0] if stats['requests'] else now,
last_seen=now
))
# 4. Check Slow Attack
if self._check_slow_attack(ip, stats):
threats.append(ThreatRecord(
ip=ip,
attack_type='slowloris',
threat_level='high',
request_count=len(stats['response_times']),
first_seen=stats['requests'][0] if stats['requests'] else now,
last_seen=now
))
# Trả về threat nguy hiểm nhất
if threats:
return max(threats, key=lambda t: self._threat_level_score(t.threat_level))
return None
def _check_brute_force(self, ip: str, stats: Dict) -> bool:
"""Phát hiện brute force attack"""
one_minute_ago = datetime.utcnow() - timedelta(minutes=1)
failures = [t for t in stats['auth_failures'] if t > one_minute_ago]
return len(failures) >= self.THRESHOLDS['auth_failures_per_minute']
def _check_dos(self, ip: str, stats: Dict) -> bool:
"""Phát hiện DoS attack"""
requests_count = self._count_requests_last_minute(ip, stats)
return requests_count >= self.THRESHOLDS['requests_per_minute']
def _check_scanner(self, ip: str, stats: Dict) -> bool:
"""Phát hiện web scanner"""
one_minute_ago = datetime.utcnow() - timedelta(minutes=1)
recent_requests = [t for t in stats['requests'] if t > one_minute_ago]
unique_paths = len(stats['paths'])
return unique_paths >= self.THRESHOLDS['unique_paths_per_minute'] and len(recent_requests) > 30
def _check_slow_attack(self, ip: str, stats: Dict) -> bool:
"""Phát hiện slow attack (slowloris, etc)"""
if not stats['response_times']:
return False
response_times = list(stats['response_times'])
p95 = sorted(response_times)[int(len(response_times) * 0.95)]
return p95 >= self.THRESHOLDS['response_time_p95_ms']
def _count_requests_last_minute(self, ip: str, stats: Dict) -> int:
"""Đếm số request trong 1 phút gần nhất"""
one_minute_ago = datetime.utcnow() - timedelta(minutes=1)
return len([t for t in stats['requests'] if t > one_minute_ago])
def _threat_level_score(self, level: str) -> int:
"""Convert threat level sang score để so sánh"""
scores = {'low': 1, 'medium': 2, 'high': 3, 'critical': 4}
return scores.get(level, 0)
def block_ip(self, ip: str, duration_seconds: int = 3600):
"""Block IP trong khoảng thời gian nhất định"""
with self.lock:
self.blocked_ips.add(ip)
self.redis.setex(f"blocked:{ip}", duration_seconds, "1")
# Ghi log block
self.redis.lpush("attack_logs", json.dumps({
'ip': ip,
'action': 'blocked',
'timestamp': datetime.utcnow().isoformat()
}))
def unblock_ip(self, ip: str):
"""Unblock IP"""
with self.lock:
self.blocked_ips.discard(ip)
self.redis.delete(f"blocked:{ip}")
Khởi tạo detector
redis_client = redis.Redis(host='localhost', port=6379, db=0)
detector = RealtimeThreatDetector(redis_client)
Xử lý sample request
sample_request = {
'ip': '203.0.113.50',
'method': 'POST',
'path': '/api/login',
'status': 401,
'response_time': 150,
'timestamp': datetime.utcnow()
}
Simulate nhiều request để test
for i in range(15):
sample_request['timestamp'] = datetime.utcnow()
threat = detector.process_request(sample_request)
if threat:
print(f"🚨 THREAT DETECTED: {threat.attack_type} from {threat.ip}")
detector.block_ip(threat.ip, duration_seconds=3600)Tích Hợp Auto-Blocking Với IPTables
Sau khi phát hiện threat, hệ thống cần tự động block IP để bảo vệ infrastructure.
import subprocess
import ipaddress
from typing import Set
class IPTablesManager:
"""
Quản lý firewall rules với iptables
Tự động block/unblock IP dựa trên threat detection
"""
def __init__(self, chain_name: str = "THREAT_BLOCK"):
self.chain_name = chain_name
self._ensure_chain_exists()
def _ensure_chain_exists(self):
"""Tạo chain nếu chưa tồn tại"""
try:
subprocess.run(
["iptables", "-N", self.chain_name],
check=False # Ignore error nếu chain đã tồn tại
)
# Thêm chain vào INPUT nếu chưa có
subprocess.run([
"iptables", "-C", "INPUT", "-j", self.chain_name
], check=False)
except Exception as e:
print(f"Lỗi khi tạo chain: {e}")
def block_ip(self, ip: str, reason: str = "") -> bool:
"""Block một IP address"""
try:
# Validate IP
ipaddress.ip_address(ip)
# Kiểm tra đã block chưa
if self.is_ip_blocked(ip):
return True
# Thêm rule
result = subprocess.run([
"iptables", "-A", self.chain_name,
"-s", ip,
"-j", "DROP",
"-m", "comment", "--comment", f"Threat: {reason}"
], capture_output=True, text=True)
if result.returncode == 0:
print(f"✅ Blocked IP {ip} - Reason: {reason}")
return True
else:
print(f"❌ Failed to block {ip}: {result.stderr}")
return False
except ValueError:
print(f"❌ Invalid IP address: {ip}")
return False
except Exception as e:
print(f"❌ Error blocking IP: {e}")
return False
def block_ip_range(self, cidr: str, reason: str = "") -> bool:
"""Block một dải IP (CIDR notation)"""
try:
network = ipaddress.ip_network(cidr, strict=False)
if network.num_addresses > 65536:
print(f"⚠️ Dải {cidr} quá lớn ({network.num_addresses} addresses), không block")
return False
result = subprocess.run([
"iptables", "-A", self.chain_name,
"-s", str(network),
"-j", "DROP",
"-m", "comment", "--comment", f"Threat range: {reason}"
], capture_output=True, text=True)
if result.returncode == 0:
print(f"✅ Blocked range {cidr} - Reason: {reason}")
return True
return False
except ValueError as e:
print(f"❌ Invalid CIDR: {cidr}")
return False
def unblock_ip(self, ip: str) -> bool:
"""Unblock IP address"""
try:
result = subprocess.run([
"iptables", "-D", self.chain_name,
"-s", ip, "-j", "DROP"
], capture_output=True, text=True)
if result.returncode == 0:
print(f"✅ Unblocked IP {ip}")
return True
return False
except Exception as e:
print(f"❌ Error unblocking IP: {e}")
return False
def is_ip_blocked(self, ip: str) -> bool:
"""Kiểm tra IP có bị block không"""
try:
result = subprocess.run([
"iptables", "-C", self.chain_name,
"-s", ip, "-j", "DROP"
], capture_output=True)
return result.returncode == 0
except Exception:
return False
def list_blocked_ips(self) -> Set[str]:
"""Liệt kê tất cả IP đang bị block"""
blocked = set()
try:
result = subprocess.run([
"iptables", "-L", self.chain_name, "-n", "--line-numbers"
], capture_output=True, text=True)
for line in result.stdout.split('\n'):
if 'DROP' in line:
parts = line.split()
if len(parts) >= 4:
ip = parts[3]
if '/' not in ip: # Skip ranges
blocked.add(ip)
except Exception as e:
print(f"Error listing blocked IPs: {e}")
return blocked
def clear_all_blocks(self):
"""Xóa tất cả rules trong chain"""
try:
subprocess.run(["iptables", "-F", self.chain_name])
print("✅ Cleared all blocked IPs")
except Exception as e:
print(f"❌ Error clearing blocks: {e}")
Sử dụng IPTables Manager
fw_manager = IPTablesManager(chain_name="THREAT_BLOCK")
Block thủ công IP đáng nghi
fw_manager.block_ip("203.0.113.50", "Brute force attack detected")
fw_manager.block_ip("198.51.100.23", "Web scanner activity")
Block cả dải IP nếu cần
fw_manager.block_ip_range("203.0.113.0/24", "DDoS from subnet")
Kiểm tra IP có bị block không
if fw_manager.is_ip_blocked("203.0.113.50"):
print("⚠️ IP này đang bị block!")
Liệt kê tất cả IP bị block
print(f"📋 Đang block {len(fw_manager.list_blocked_ips())} IPs")Dashboard Giám Sát Với Streaming Updates
Để theo dõi trực quan, tôi sử dụng Flask + Server-Sent Events để hiển thị log và alerts theo thời gian thực.
from flask import Flask, Response, jsonify, render_template
import redis
import json
import time
from datetime import datetime, timedelta
from collections import Counter
app = Flask(__name__)
redis_client = redis.Redis(host='localhost', port=6379, db=0)
@app.route('/')
def dashboard():
"""Trang dashboard chính"""
return render_template('dashboard.html')
@app.route('/api/stats')
def get_stats():
"""API trả về thống kê tổng quan"""
# Đếm requests trong 5 phút gần nhất
stats = {
'total_requests': 0,
'unique_ips': set(),
'error_count': 0,
'blocked_ips_count': 0,
'top_ips': [],
'status_distribution': Counter(),
'avg_response_time': 0
}
# Lấy logs từ Redis
logs = redis_client.lrange("request_logs", 0, 999)
response_times = []
for log in logs:
try:
entry = json.loads(log)
stats['total_requests'] += 1
stats['unique_ips'].add(entry.get('ip', ''))
stats['status_distribution'][entry.get('status')] += 1
rt = entry.get('response_time', 0)
if rt:
response_times.append(rt)
if entry.get('status', 0) >= 400:
stats['error_count'] += 1
except json.JSONDecodeError:
continue
# Tính toán response time trung bình
if response_times:
stats['avg_response_time'] = sum(response_times) / len(response_times)
# Top 10 IPs
ip_counts = Counter()
for log in logs:
try:
entry = json.loads(log)
ip_counts[entry.get('ip', '')] += 1
except:
continue
stats['top_ips'] = ip_counts.most_common(10)
stats['unique_ips'] = len(stats['unique_ips'])
stats['status_distribution'] = dict(stats['status_distribution'])
return jsonify(stats)
@app.route('/api/stream')
def stream():
"""Server-Sent Events endpoint cho real-time updates"""
def event_stream():
last_id = 0
while True:
# Lấy logs mới từ Redis
new_logs = redis_client.lrange("request_logs", last_id, last_id + 50)
if new_logs:
for log in new_logs:
yield f"data: {log.decode() if isinstance(log, bytes) else log}\n\n"
last_id += 1
# Kiểm tra alerts
alerts = redis_client.lrange("alerts", 0, 0)
for alert in alerts:
yield f"event: alert\ndata: {alert.decode() if isinstance(alert, bytes) else alert}\n\n"
redis_client.delete("alerts")
time.sleep(0.5) # Poll mỗi 500ms
return Response(
event_stream(),
mimetype='text/event-stream',
headers={
'Cache-Control': 'no-cache',
'X-Accel-Buffering': 'no'
}
)
@app.route('/api/blocked-ips')
def get_blocked_ips():
"""API lấy danh sách IP bị block"""
blocked = redis_client.smembers("blocked_ips")
result = []
for ip in blocked:
ip_str = ip.decode() if isinstance(ip, bytes) else ip
ttl = redis_client.ttl(f"blocked:{ip_str}")
result.append({
'ip': ip_str,
'remaining_seconds': ttl if ttl > 0 else 0
})
return jsonify(result)
@app.route('/api/block-ip', methods=['POST'])
def manual_block():
"""API để admin block IP thủ công"""
from flask import request
data = request.json
ip = data.get('ip')
duration = data.get('duration', 3600)
if not ip:
return jsonify({'error': 'IP required'}), 400
redis_client.setex(f"blocked:{ip}", duration, "1")
redis_client.sadd("blocked_ips", ip)
return jsonify({'success': True, 'ip': ip, 'duration': duration})
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000, debug=False, threaded=True)Kết Quả Triển Khai Thực Tế
Sau khi triển khai hệ thống này trong 3 tháng, đây là những con số ấn tượng:
- 247,000 requests được phân tích mỗi ngày
- 1,247 brute force attempts bị block tự động
- 89 web scanners bị phát hiện và ngăn chặn
- 99.7% uptime hệ thống
- Response time trung bình: 45ms (so với 200ms trước đây khi bị tấn công)
- Chi phí API: Chỉ $12/tháng với HolySheep AI (so với $150+ nếu dùng OpenAI)
Lỗi Thường Gặp Và Cách Khắc Phục
1. Lỗi "ConnectionError: timeout" Khi Gọi API
Mô tả lỗi: Khi hệ thống load cao, requests đến HolySheep API bị timeout sau 30s.
# ❌ SAI: Không có timeout
response = requests.post(url, json=payload)
✅ ĐÚNG: Thêm timeout và retry logic
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def create_session_with_retry():
session = requests.Session()
retry = Retry(
total=3,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504],
)
adapter = HTTPAdapter(max_retries=retry)
session.mount('https://', adapter)
return session
Sử dụng session với timeout cụ thể
session = create_session_with_retry()
try:
response = session.post(
f"{base_url}/chat/completions",
json=payload,
timeout=(5, 10) # Connect timeout 5s, Read timeout 10s
)
except requests.exceptions.Timeout:
# Fallback sang rule-based detection
return fallback_detection(log_entry)2. Lỗi "Redis Connection refused" Khi Khởi Động
Mô tả lỗi: Ứng dụng không kết nối được Redis, toàn bộ request bị lỗi.
# ❌ SAI: Kết nối trực tiếp không kiểm tra
redis_client = redis.Redis(host='localhost', port=6379)
✅ ĐÚNG: Connection pooling với health check
class RedisManager:
def __init__(self, host='localhost', port=6379, max_retries=3):
self.config = {'host': host, 'port': port, 'db': 0}
self.max_retries = max_retries
self._client = None
@property
def client(self):
if self._client is None:
self._client = self._connect_with_retry()
return self._client
def _connect_with_retry(self):
for attempt in range(self.max_retries):
try:
client = redis