it
BetterUptime กับ Multi-tenant Design — วิธีใช้
BetterUptime Monitoring
BetterUptime ตรวจสอบ Website และ API ทุก 30 วินาที แจ้งเตือนเมื่อ Service Down มี Status Pages, Incident Management และ Heartbeat Monitoring สำหรับ Cron Jobs
เนื้อหาเกี่ยวข้อง — ทำความเข้าใจ Neon Serverless Postgres Security Hardening
Multi-tenant Architecture ให้หลาย Tenants ใช้ Infrastructure เดียวกัน ต้อง Isolate Data, Compute, Network ใช้ Monitoring แยกตาม Tenant
เนื้อหาเกี่ยวข้อง — บทความที่เกี่ยวข้อง: เก็บเงินภาษาอังกฤษ — วิธีตั้งค่าและใช้งานจริงพร้อมตัวอย่าง
BetterUptime Configuration
# betteruptime_setup.py — BetterUptime API Configuration
# pip install requests
import requests
import json
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from datetime import datetime
BETTERUPTIME_API = "https://betteruptime.com/api/v2"
@dataclass
class Monitor:
url: str
monitor_type: str = "status" # status, keyword, ping, heartbeat
check_frequency: int = 30 # seconds
regions: List[str] = field(default_factory=lambda: ["us", "eu", "asia"])
expected_status: int = 200
alert_channels: List[str] = field(default_factory=list)
class BetterUptimeManager:
"""BetterUptime API Manager"""
def __init__(self, api_token):
self.token = api_token
self.headers = {
"Authorization": f"Bearer {api_token}",
"Content-Type": "application/json",
}
def create_monitor(self, monitor: Monitor):
"""สร้าง Monitor"""
payload = {
"monitor_type": monitor.monitor_type,
"url": monitor.url,
"check_frequency": monitor.check_frequency,
"regions": ",".join(monitor.regions),
"expected_status_codes": [monitor.expected_status],
}
resp = requests.post(
f"{BETTERUPTIME_API}/monitors",
headers=self.headers,
json=payload,
)
return resp.json()
def create_heartbeat(self, name, period=300, grace=60):
"""สร้าง Heartbeat Monitor สำหรับ Cron Jobs"""
payload = {
"name": name,
"period": period, # Expected interval (seconds)
"grace": grace, # Grace period before alert
}
resp = requests.post(
f"{BETTERUPTIME_API}/heartbeats",
headers=self.headers,
json=payload,
)
return resp.json()
def create_status_page(self, name, subdomain, resources):
"""สร้าง Status Page"""
payload = {
"company_name": name,
"subdomain": subdomain,
"timezone": "Asia/Bangkok",
}
resp = requests.post(
f"{BETTERUPTIME_API}/status-pages",
headers=self.headers,
json=payload,
)
return resp.json()
def create_incident(self, name, description, affected_resources):
"""สร้าง Incident"""
payload = {
"name": name,
"description": description,
"status": "investigating",
"affected_resources": affected_resources,
}
resp = requests.post(
f"{BETTERUPTIME_API}/incidents",
headers=self.headers,
json=payload,
)
return resp.json()
def setup_tenant_monitoring(self, tenant_id, services):
"""Setup Monitoring สำหรับ Tenant"""
print(f"\nSetup monitoring for tenant: {tenant_id}")
for service in services:
monitor = Monitor(
url=f"https://{tenant_id}.app.example.com{service['path']}",
monitor_type=service.get("type", "status"),
check_frequency=service.get("interval", 60),
)
result = self.create_monitor(monitor)
print(f" Monitor: {monitor.url} — Created")
# ตัวอย่าง
# manager = BetterUptimeManager("your-api-token")
# manager.setup_tenant_monitoring("acme", [
# {"path": "/health", "type": "status", "interval": 30},
# {"path": "/api/v1/status", "type": "status", "interval": 60},
# ])
tenants = ["acme", "globex", "initech", "umbrella"]
services = ["/health", "/api/v1/status", "/api/v1/metrics"]
print("BetterUptime Tenant Monitoring:")
for tenant in tenants:
print(f"\n Tenant: {tenant}")
for svc in services:
print(f" Monitor: https://{tenant}.app.example.com{svc}")Multi-tenant Database Design
# multi_tenant_db.py — Multi-tenant Database Design
# pip install sqlalchemy asyncpg
from dataclasses import dataclass
from typing import List, Dict, Optional
from enum import Enum
class IsolationLevel(Enum):
SHARED_DB = "shared_database" # แชร์ DB, แยก tenant_id
SCHEMA_PER_TENANT = "schema" # แยก Schema per tenant
DB_PER_TENANT = "database" # แยก Database per tenant
@dataclass
class TenantConfig:
tenant_id: str
name: str
plan: str # free, pro, enterprise
isolation: IsolationLevel
db_host: str = "localhost"
db_name: str = "app"
schema_name: str = "public"
max_users: int = 100
rate_limit: int = 1000 # requests/min
class MultiTenantManager:
"""Multi-tenant Database Manager"""
def __init__(self):
self.tenants: Dict[str, TenantConfig] = {}
def register_tenant(self, config: TenantConfig):
self.tenants[config.tenant_id] = config
print(f"Registered tenant: {config.tenant_id} ({config.isolation.value})")
def get_connection_string(self, tenant_id):
"""Get Database Connection String สำหรับ Tenant"""
config = self.tenants.get(tenant_id)
if not config:
raise ValueError(f"Tenant not found: {tenant_id}")
if config.isolation == IsolationLevel.DB_PER_TENANT:
return f"postgresql://{config.db_host}/{config.tenant_id}_db"
elif config.isolation == IsolationLevel.SCHEMA_PER_TENANT:
return f"postgresql://{config.db_host}/{config.db_name}?schema={config.tenant_id}"
else:
return f"postgresql://{config.db_host}/{config.db_name}"
def get_query_filter(self, tenant_id):
"""Get SQL Filter สำหรับ Shared DB"""
config = self.tenants.get(tenant_id)
if config and config.isolation == IsolationLevel.SHARED_DB:
return f"WHERE tenant_id = '{tenant_id}'"
return ""
def dashboard(self):
"""Tenant Dashboard"""
print(f"\n{'='*60}")
print(f"Multi-tenant Dashboard")
print(f"{'='*60}")
for tid, config in self.tenants.items():
print(f"\n [{config.plan:>10}] {config.name} ({tid})")
print(f" Isolation: {config.isolation.value}")
print(f" DB: {self.get_connection_string(tid)}")
print(f" Limits: {config.max_users} users, {config.rate_limit} req/min")
# SQL Schema สำหรับ Shared Database
SHARED_DB_SCHEMA = """
-- Shared Database Schema (tenant_id approach)
CREATE TABLE tenants (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
name VARCHAR(255) NOT NULL,
slug VARCHAR(100) UNIQUE NOT NULL,
plan VARCHAR(50) DEFAULT 'free',
settings JSONB DEFAULT '{}',
created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE TABLE users (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
tenant_id UUID NOT NULL REFERENCES tenants(id),
email VARCHAR(255) NOT NULL,
name VARCHAR(255) NOT NULL,
role VARCHAR(50) DEFAULT 'member',
UNIQUE(tenant_id, email)
);
CREATE TABLE projects (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
tenant_id UUID NOT NULL REFERENCES tenants(id),
name VARCHAR(255) NOT NULL,
status VARCHAR(50) DEFAULT 'active',
created_by UUID REFERENCES users(id)
);
-- Row-Level Security
ALTER TABLE users ENABLE ROW LEVEL SECURITY;
ALTER TABLE projects ENABLE ROW LEVEL SECURITY;
CREATE POLICY tenant_isolation_users ON users
USING (tenant_id = current_setting('app.current_tenant')::UUID);
CREATE POLICY tenant_isolation_projects ON projects
USING (tenant_id = current_setting('app.current_tenant')::UUID);
-- Indexes
CREATE INDEX idx_users_tenant ON users(tenant_id);
CREATE INDEX idx_projects_tenant ON projects(tenant_id);
"""
# ตัวอย่าง
manager = MultiTenantManager()
manager.register_tenant(TenantConfig(
"acme", "Acme Corp", "enterprise", IsolationLevel.DB_PER_TENANT,
max_users=1000, rate_limit=10000))
manager.register_tenant(TenantConfig(
"globex", "Globex Inc", "pro", IsolationLevel.SCHEMA_PER_TENANT,
max_users=100, rate_limit=5000))
manager.register_tenant(TenantConfig(
"initech", "Initech LLC", "free", IsolationLevel.SHARED_DB,
max_users=10, rate_limit=1000))
manager.dashboard()Tenant-aware API Middleware
# tenant_middleware.py — Multi-tenant API Middleware
# pip install fastapi uvicorn
from dataclasses import dataclass
from typing import Dict, Optional
from datetime import datetime
import time
@dataclass
class TenantContext:
tenant_id: str
plan: str
rate_limit: int
request_count: int = 0
last_reset: float = 0
class TenantMiddleware:
"""Middleware สำหรับ Multi-tenant API"""
def __init__(self):
self.tenants: Dict[str, TenantContext] = {}
self.request_log: list = []
def register(self, tenant_id, plan="free", rate_limit=1000):
self.tenants[tenant_id] = TenantContext(
tenant_id=tenant_id, plan=plan,
rate_limit=rate_limit, last_reset=time.time())
def resolve_tenant(self, host=None, header=None, path=None):
"""Resolve Tenant จาก Request"""
tenant_id = None
# 1. Subdomain: acme.app.example.com
if host and "." in host:
tenant_id = host.split(".")[0]
# 2. Header: X-Tenant-ID
if not tenant_id and header:
tenant_id = header
# 3. Path: /api/v1/tenants/acme/...
if not tenant_id and path and "/tenants/" in path:
parts = path.split("/tenants/")
if len(parts) > 1:
tenant_id = parts[1].split("/")[0]
return tenant_id
def check_rate_limit(self, tenant_id):
"""ตรวจสอบ Rate Limit"""
ctx = self.tenants.get(tenant_id)
if not ctx:
return False, "Tenant not found"
now = time.time()
# Reset counter every minute
if now - ctx.last_reset > 60:
ctx.request_count = 0
ctx.last_reset = now
ctx.request_count += 1
if ctx.request_count > ctx.rate_limit:
return False, f"Rate limit exceeded ({ctx.rate_limit}/min)"
return True, f"OK ({ctx.request_count}/{ctx.rate_limit})"
def process_request(self, host, path, method="GET"):
"""Process Request ผ่าน Middleware"""
tenant_id = self.resolve_tenant(host=host)
if not tenant_id or tenant_id not in self.tenants:
return {"status": 404, "error": "Tenant not found"}
ok, msg = self.check_rate_limit(tenant_id)
if not ok:
return {"status": 429, "error": msg}
self.request_log.append({
"tenant": tenant_id,
"path": path,
"method": method,
"timestamp": datetime.now().isoformat(),
})
return {"status": 200, "tenant": tenant_id, "message": msg}
def metrics(self):
"""แสดง Metrics per Tenant"""
print(f"\n{'='*55}")
print(f"Tenant API Metrics")
print(f"{'='*55}")
for tid, ctx in self.tenants.items():
print(f"\n {tid} [{ctx.plan}]:")
print(f" Requests: {ctx.request_count}/{ctx.rate_limit}")
tenant_logs = [r for r in self.request_log if r["tenant"] == tid]
print(f" Total: {len(tenant_logs)}")
# ตัวอย่าง
mw = TenantMiddleware()
mw.register("acme", "enterprise", 10000)
mw.register("globex", "pro", 5000)
mw.register("initech", "free", 100)
# จำลอง Requests
requests_sim = [
("acme.app.example.com", "/api/users"),
("globex.app.example.com", "/api/projects"),
("acme.app.example.com", "/api/orders"),
("initech.app.example.com", "/api/users"),
]
for host, path in requests_sim:
result = mw.process_request(host, path)
print(f" {host}{path} -> {result['status']}")
mw.metrics()Best Practices
- Monitoring per Tenant: ตั้ง Monitors แยกตาม Tenant ดู Health ของแต่ละ Tenant
- Status Pages: สร้าง Status Page สำหรับลูกค้า ลด Support Tickets
- Row-Level Security: ใช้ RLS ป้องกัน Data Leak ระหว่าง Tenants
- Rate Limiting: ตั้ง Rate Limit ตาม Plan ป้องกัน Noisy Neighbor
- Tenant Resolution: ใช้ Subdomain, Header หรือ Path resolve Tenant
- Isolation Strategy: เลือก Isolation Level ตาม Compliance และ Cost
BetterUptime คืออะไร
Uptime Monitoring Platform ตรวจสอบ Website API ทุก 30 วินาที แจ้งเตือน SMS Email Slack PagerDuty มี Status Pages Incident Management Heartbeat สำหรับ Cron Jobs
แนะนำเพิ่มเติม — อ่านเพิ่มเติมที่ SiamCafeBook
เนื้อหาเกี่ยวข้อง — A/B Testing ML Best Practices ที่ต้องรู้
