Hướng Dẫn Thiết Kế Ingestion API cho IoT

Khi bạn kết nối hàng nghìn device vào internet, tất cả đều muốn gửi data — temperature reading, GPS coordinate, heartbeat signal, error log. Câu hỏi không phải là bạn có cần ingestion API hay không, mà là ingestion API của bạn có chịu nổi những gì sắp đến hay không.

Ingestion API là entry point cho toàn bộ data từ device. Khác với REST API thông thường nơi user click một cái button rồi đợi response, ingestion API phải deal với machine gửi data liên tục, thường xuyên burst, đôi khi mất kết nối, và luôn luôn ở scale lớn.

Bài viết này sẽ đi qua cách thiết kế một ingestion API có thể handle workload IoT thực tế mà không bị sập.

Thời gian đọc: 2-3 giờ
Yêu cầu: Hiểu biết về REST API, HTTP, và concept cơ bản về message queue

Bạn Sẽ Học Được

✅ Điều gì khiến IoT ingestion khác biệt so với API thông thường
✅ Cách thiết kế ingestion endpoint cho device data throughput cao
✅ Batching strategy để giảm network overhead
✅ Payload validation và schema enforcement cho device telemetry
✅ Backpressure và flow control để bảo vệ backend
✅ Storage strategy — time-series database, message queue, và data lake
✅ Authentication và device identity management
✅ Implementation thực tế với Python (FastAPI) và MQTT

1. Ingestion API Là Gì?

Ingestion API nhận, validate, và route data đến từ external source vào system của bạn. Trong IoT, những source đó là device — sensor, actuator, gateway, và edge node.

Khác Gì So Với REST API Thông Thường?

Đặc điểm	REST API thông thường	IoT Ingestion API
Client type	Con người (qua browser/app)	Machine (sensor, device)
Traffic pattern	Request-response, lác đác	Stream liên tục, bursty
Payload size	Đa dạng (KB đến MB)	Nhỏ nhưng liên tục (byte đến KB)
Throughput	Hàng trăm đến hàng nghìn RPS	Hàng chục nghìn đến hàng triệu RPS
Latency requirement	Response < 500ms	Acknowledge nhanh, xử lý sau
Data format	JSON	JSON, CBOR, Protobuf, binary
Connection	HTTP ngắn hạn	Long-lived (MQTT, WebSocket) hoặc HTTP
Error handling	Trả error cho user	Buffer, retry, không bao giờ mất data

Điểm khác biệt cốt lõi: REST API phục vụ user kỳ vọng response có ý nghĩa ngay lập tức. Ingestion API nhận data từ machine chỉ cần một cái "nhận rồi" nhanh gọn — xử lý thật sự diễn ra asynchronously phía sau.

2. Ingestion Pattern

Có ba pattern chính cho cách device gửi data đến ingestion API của bạn.

Pattern 1: HTTP Push (Request-Response)

Cách đơn giản nhất. Device gửi HTTP POST request kèm telemetry data.

# FastAPI ingestion endpoint
from fastapi import FastAPI, HTTPException, Header
from pydantic import BaseModel, Field
from datetime import datetime
from typing import Optional
import uuid
 
app = FastAPI()
 
class TelemetryPayload(BaseModel):
    device_id: str = Field(..., min_length=1, max_length=64)
    timestamp: datetime
    metrics: dict
    metadata: Optional[dict] = None
 
@app.post("/api/v1/ingest", status_code=202)
async def ingest_telemetry(
    payload: TelemetryPayload,
    x_device_token: str = Header(...),
):
    # Validate device identity
    if not await verify_device_token(payload.device_id, x_device_token):
        raise HTTPException(status_code=401, detail="Invalid device token")
 
    # Enqueue cho async processing — respond ngay lập tức
    message_id = str(uuid.uuid4())
    await enqueue_message(message_id, payload.model_dump())
 
    return {"status": "accepted", "message_id": message_id}

Những quyết định design quan trọng:

HTTP 202 Accepted — không phải 200 OK. Data được chấp nhận để xử lý, chưa xử lý xong.
Async processing — validate, enqueue, respond. Không làm heavy processing trong request path.
Device token trong header — tách biệt khỏi payload cho clean separation of concern.

Khi nào dùng: Throughput thấp đến trung bình (< 10K message/sec), device có internet ổn định, deployment đơn giản.

MQTT là protocol chuẩn cho IoT. Device publish message lên topic, ingestion service của bạn subscribe.

# MQTT ingestion service
import asyncio
import json
from aiomqtt import Client
 
async def mqtt_ingestion():
    async with Client("mqtt-broker.internal", port=1883) as client:
        await client.subscribe("devices/+/telemetry")
 
        async for message in client.messages:
            topic_parts = message.topic.value.split("/")
            device_id = topic_parts[1]
 
            try:
                payload = json.loads(message.payload.decode())
                await process_telemetry(device_id, payload)
            except json.JSONDecodeError:
                await handle_malformed_message(device_id, message.payload)
 
async def process_telemetry(device_id: str, payload: dict):
    # Validate schema
    validated = validate_telemetry_schema(payload)
 
    # Enrich với server-side timestamp
    validated["server_received_at"] = datetime.utcnow().isoformat()
    validated["device_id"] = device_id
 
    # Forward đến message queue cho downstream processing
    await publish_to_kafka("telemetry-raw", validated)

Lợi thế của MQTT cho IoT:

Lightweight — packet overhead tối thiểu, thiết kế cho constrained device
QoS level — QoS 0 (at most once), QoS 1 (at least once), QoS 2 (exactly once)
Last Will and Testament — broker thông báo khi device disconnect bất ngờ
Retained message — subscriber mới nhận được message cuối cùng trên topic ngay lập tức
Persistent session — broker queue message khi device offline

Khi nào dùng: Throughput cao, constrained device, mạng không ổn định, cần bidirectional communication.

Pattern 3: Batch Upload

Device thu thập data locally rồi upload theo batch — hữu ích cho device thường xuyên offline.

from pydantic import BaseModel, Field
from typing import List
from datetime import datetime
 
class TelemetryReading(BaseModel):
    timestamp: datetime
    metrics: dict
 
class BatchPayload(BaseModel):
    device_id: str
    batch_id: str
    readings: List[TelemetryReading] = Field(..., max_length=1000)
 
@app.post("/api/v1/ingest/batch", status_code=202)
async def ingest_batch(
    payload: BatchPayload,
    x_device_token: str = Header(...),
):
    if not await verify_device_token(payload.device_id, x_device_token):
        raise HTTPException(status_code=401, detail="Invalid device token")
 
    # Check duplicate batch
    if await is_duplicate_batch(payload.batch_id):
        return {"status": "duplicate", "batch_id": payload.batch_id}
 
    # Enqueue toàn bộ batch
    await enqueue_batch(payload.batch_id, payload.model_dump())
 
    return {
        "status": "accepted",
        "batch_id": payload.batch_id,
        "readings_count": len(payload.readings),
    }

Quyết định design quan trọng:

batch_id cho idempotency — device retry khi fail. Không có deduplication thì sẽ bị duplicate.
Max batch size (1000) — ngăn một device đơn lẻ overwhelm endpoint.
202 Accepted — giống single ingestion. Process batch asynchronously.

Khi nào dùng: Device có kết nối gián đoạn, edge computing scenario, giảm network overhead.

3. Payload Design và Validation

Device telemetry data cần structure nhất quán. Không có schema enforcement, downstream pipeline của bạn sẽ thành một mớ hỗn độn.

Standard Telemetry Schema

{
  "device_id": "sensor-factory-floor-001",
  "timestamp": "2026-03-27T10:15:30.123Z",
  "metrics": {
    "temperature": 23.5,
    "humidity": 67.2,
    "pressure": 1013.25
  },
  "metadata": {
    "firmware_version": "2.1.0",
    "battery_level": 85,
    "signal_strength": -67
  }
}

Validation Layer

Validation trong ingestion API diễn ra theo layer — mỗi layer bắt những problem khác nhau:

Layer 1 — Protocol validation: Content type có đúng không? Body có parse được không?

Layer 2 — Schema validation: Payload có đủ required field không? Type có đúng không?

from pydantic import BaseModel, Field, validator
from datetime import datetime, timedelta
 
class TelemetryPayload(BaseModel):
    device_id: str = Field(..., pattern=r"^[a-zA-Z0-9\-_]{1,64}$")
    timestamp: datetime
    metrics: dict = Field(..., min_length=1)
    metadata: dict = {}
 
    @validator("timestamp")
    def timestamp_not_in_future(cls, v):
        if v > datetime.utcnow() + timedelta(minutes=5):
            raise ValueError("Timestamp cannot be more than 5 minutes in the future")
        return v
 
    @validator("timestamp")
    def timestamp_not_too_old(cls, v):
        if v < datetime.utcnow() - timedelta(days=7):
            raise ValueError("Timestamp cannot be more than 7 days old")
        return v

Layer 3 — Semantic validation: Giá trị có hợp lý không? Temperature 9999°C thì đúng syntax nhưng sai semantic.

# Define expected range cho mỗi metric type
METRIC_RANGES = {
    "temperature": (-50.0, 150.0),
    "humidity": (0.0, 100.0),
    "pressure": (800.0, 1200.0),
    "battery_level": (0, 100),
    "signal_strength": (-120, 0),
}
 
def validate_metric_ranges(metrics: dict) -> list[str]:
    warnings = []
    for key, value in metrics.items():
        if key in METRIC_RANGES:
            min_val, max_val = METRIC_RANGES[key]
            if not (min_val <= value <= max_val):
                warnings.append(
                    f"{key}={value} outside expected range [{min_val}, {max_val}]"
                )
    return warnings

Layer 4 — Business rule: Device này đã register chưa? Có đang gửi data từ location dự kiến không? Firmware có bị recall không?

Quan trọng: Layer 1-2 nên reject ngay (HTTP 400). Layer 3-4 nên accept data nhưng flag lại — route đến dead letter queue để investigate. Bạn không bao giờ muốn mất data từ device ngoài field vì một cái validation bug trong API.

4. Backpressure và Flow Control

Problem khó nhất trong ingestion: chuyện gì xảy ra khi data đến nhanh hơn khả năng xử lý?

Vấn Đề

10,000 message/sec vào, 5,000 message/sec ra. Không có backpressure, queue tăng vô hạn cho đến khi có thứ gì đó crash.

Strategy 1: Rate Limiting per Device

from fastapi import Request, HTTPException
import time
 
# In-memory rate limiter (dùng Redis cho production)
device_request_counts: dict[str, list[float]] = {}
 
RATE_LIMIT = 10  # max request/giây/device
WINDOW = 1.0     # 1 second window
 
async def check_rate_limit(device_id: str):
    now = time.time()
 
    if device_id not in device_request_counts:
        device_request_counts[device_id] = []
 
    # Xóa entry hết hạn
    device_request_counts[device_id] = [
        ts for ts in device_request_counts[device_id]
        if now - ts < WINDOW
    ]
 
    if len(device_request_counts[device_id]) >= RATE_LIMIT:
        raise HTTPException(
            status_code=429,
            detail="Rate limit exceeded",
            headers={"Retry-After": "1"},
        )
 
    device_request_counts[device_id].append(now)

Strategy 2: Load Shedding

Khi system bị overwhelm, cố ý drop traffic priority thấp để bảo vệ data quan trọng.

import asyncio
 
class LoadShedder:
    def __init__(self, max_queue_depth: int = 10000):
        self.max_queue_depth = max_queue_depth
        self.current_depth = 0
 
    async def should_accept(self, priority: str) -> bool:
        utilization = self.current_depth / self.max_queue_depth
 
        if utilization < 0.7:
            return True  # Accept tất cả
 
        if utilization < 0.9:
            # Chỉ accept medium và high priority
            return priority in ("high", "medium")
 
        # Critical mode — chỉ accept high priority
        return priority == "high"

Priority level cho IoT data:

High: Alarm, safety event, device failure
Medium: Regular telemetry, status update
Low: Diagnostic log, debug data, heartbeat

Strategy 3: Buffering tại Edge

Đẩy buffer gần hơn source. Edge gateway thu thập và batch data trước khi gửi đến ingestion API.

Cách này giảm số connection đến API của bạn cỡ vài bậc. Thay vì 10,000 sensor kết nối trực tiếp, 100 gateway mỗi cái aggregate 100 sensor và gửi batch request mỗi 30 giây.

5. Authentication và Device Identity

Mỗi device gửi data phải chứng minh nó là ai. Không có điều này, bất kỳ ai cũng có thể inject fake data vào system.

Device Authentication Strategy

API Key (Đơn giản)

# Device gửi API key trong header
@app.post("/api/v1/ingest")
async def ingest(
    payload: TelemetryPayload,
    x_api_key: str = Header(...),
):
    device = await lookup_device_by_api_key(x_api_key)
    if not device:
        raise HTTPException(status_code=401, detail="Invalid API key")
    if device.device_id != payload.device_id:
        raise HTTPException(status_code=403, detail="Device ID mismatch")
 
    # Tiến hành ingestion
    ...

Mutual TLS (mTLS) — Bảo mật cao

Mỗi device có client certificate. TLS handshake verify cả server lẫn client identity.

# Nginx configuration cho mTLS
server {
    listen 8883 ssl;
 
    ssl_certificate /etc/nginx/certs/server.crt;
    ssl_certificate_key /etc/nginx/certs/server.key;
 
    # Yêu cầu client certificate
    ssl_client_certificate /etc/nginx/certs/ca.crt;
    ssl_verify_client on;
 
    location /api/v1/ingest {
        # Truyền client cert info đến backend
        proxy_set_header X-Client-Cert-DN $ssl_client_s_dn;
        proxy_set_header X-Client-Cert-Serial $ssl_client_serial;
        proxy_pass http://ingestion-api:8000;
    }
}

JWT với Device Provisioning

Device lấy short-lived token thông qua provisioning flow, rồi dùng token đó cho ingestion.

Chọn Cái Nào?

Method	Security	Device Complexity	Management Overhead
API Key	Thấp	Rất thấp	Trung bình (key rotation)
mTLS	Cao	Trung bình (cert management)	Cao (PKI infrastructure)
JWT + Provisioning	Trung bình-Cao	Thấp-Trung bình	Trung bình

Khuyến nghị: Bắt đầu với API key cho development và prototyping. Chuyển sang mTLS hoặc JWT cho production deployment với > 1,000 device.

6. Storage Strategy

Khi data đã đi qua ingestion API và message queue, nó cần được lưu ở đâu đó. Lựa chọn phụ thuộc vào query pattern của bạn.

Time-Series Database (InfluxDB, TimescaleDB)

Tốt nhất cho: real-time monitoring, dashboard, alerting.

-- TimescaleDB: Tạo hypertable cho telemetry
CREATE TABLE telemetry (
    time        TIMESTAMPTZ NOT NULL,
    device_id   TEXT NOT NULL,
    metric_name TEXT NOT NULL,
    value       DOUBLE PRECISION,
    metadata    JSONB
);
 
SELECT create_hypertable('telemetry', 'time');
 
-- Tự động retention: xóa data cũ hơn 90 ngày
SELECT add_retention_policy('telemetry', INTERVAL '90 days');
 
-- Continuous aggregate cho dashboard
CREATE MATERIALIZED VIEW telemetry_hourly
WITH (timescaledb.continuous) AS
SELECT
    time_bucket('1 hour', time) AS bucket,
    device_id,
    metric_name,
    AVG(value) AS avg_value,
    MIN(value) AS min_value,
    MAX(value) AS max_value,
    COUNT(*) AS sample_count
FROM telemetry
GROUP BY bucket, device_id, metric_name;

Message Queue → Stream Processing

Tốt nhất cho: event-driven architecture, real-time analytics, complex event processing.

# Kafka consumer cho stream processing
from confluent_kafka import Consumer
 
consumer = Consumer({
    "bootstrap.servers": "kafka:9092",
    "group.id": "telemetry-processor",
    "auto.offset.reset": "earliest",
})
consumer.subscribe(["telemetry-raw"])
 
while True:
    msg = consumer.poll(1.0)
    if msg is None:
        continue
 
    telemetry = json.loads(msg.value())
 
    # Real-time anomaly detection
    if is_anomalous(telemetry):
        publish_alert(telemetry)
 
    # Ghi vào time-series DB
    write_to_timescaledb(telemetry)
 
    # Archive vào data lake
    write_to_s3_parquet(telemetry)

Data Lake (S3 + Parquet)

Tốt nhất cho: historical analysis, machine learning, long-term storage.

import pyarrow as pa
import pyarrow.parquet as pq
from datetime import datetime
 
# Partition theo date và device cho query hiệu quả
def write_to_data_lake(records: list[dict], base_path: str):
    table = pa.Table.from_pylist(records)
 
    # Partitioned write
    pq.write_to_dataset(
        table,
        root_path=f"{base_path}/telemetry",
        partition_cols=["date", "device_id"],
    )

Multi-Tier Storage Architecture

Hầu hết production IoT system dùng cả ba:

Hot tier (Time-Series DB): 7-30 ngày gần nhất. Query nhanh cho dashboard và monitoring.
Cold tier (Data Lake): Toàn bộ historical data. Chậm nhưng rẻ. Dùng cho ML training và compliance.
Downsampling: Sau hot retention period, aggregate raw data (reading 1 giây → average 1 phút) trước khi archive.

7. Implementation End-to-End

Gộp tất cả lại với một ingestion service production-ready.

Project Structure

iot-ingestion/
├── app/
│   ├── main.py              # FastAPI application
│   ├── models.py             # Pydantic model
│   ├── auth.py               # Device authentication
│   ├── rate_limiter.py       # Rate limiting
│   ├── queue.py              # Message queue producer
│   └── health.py             # Health check endpoint
├── docker-compose.yml
└── requirements.txt

Core Ingestion Service

# app/main.py
from fastapi import FastAPI, HTTPException, Header, Depends
from contextlib import asynccontextmanager
from app.models import TelemetryPayload, BatchPayload
from app.auth import verify_device
from app.rate_limiter import RateLimiter
from app.queue import QueueProducer
 
producer: QueueProducer | None = None
rate_limiter: RateLimiter | None = None
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    global producer, rate_limiter
    producer = QueueProducer(brokers="kafka:9092")
    await producer.start()
    rate_limiter = RateLimiter(max_requests=10, window_seconds=1)
    yield
    await producer.stop()
 
app = FastAPI(title="IoT Ingestion API", lifespan=lifespan)
 
@app.post("/api/v1/ingest", status_code=202)
async def ingest(
    payload: TelemetryPayload,
    x_device_token: str = Header(...),
):
    # Authenticate
    device = await verify_device(payload.device_id, x_device_token)
    if not device:
        raise HTTPException(status_code=401, detail="Invalid device")
 
    # Rate limit
    if not await rate_limiter.allow(payload.device_id):
        raise HTTPException(status_code=429, detail="Rate limit exceeded")
 
    # Validate metric range (warn, không reject)
    warnings = validate_metric_ranges(payload.metrics)
 
    # Enqueue
    await producer.send(
        topic="telemetry-raw",
        key=payload.device_id,
        value=payload.model_dump_json(),
    )
 
    return {
        "status": "accepted",
        "warnings": warnings if warnings else None,
    }
 
@app.post("/api/v1/ingest/batch", status_code=202)
async def ingest_batch(
    payload: BatchPayload,
    x_device_token: str = Header(...),
):
    device = await verify_device(payload.device_id, x_device_token)
    if not device:
        raise HTTPException(status_code=401, detail="Invalid device")
 
    if not await rate_limiter.allow(payload.device_id):
        raise HTTPException(status_code=429, detail="Rate limit exceeded")
 
    # Gửi mỗi reading như message riêng cho parallel processing
    for reading in payload.readings:
        await producer.send(
            topic="telemetry-raw",
            key=payload.device_id,
            value=reading.model_dump_json(),
        )
 
    return {
        "status": "accepted",
        "readings_count": len(payload.readings),
    }

Health Check

Production ingestion API cần health check vượt xa "process có đang chạy không":

# app/health.py
from fastapi import APIRouter
 
router = APIRouter()
 
@router.get("/health/ready")
async def readiness():
    """Instance này có thể nhận traffic không?"""
    checks = {
        "kafka": await producer.is_connected(),
        "rate_limiter": rate_limiter is not None,
    }
    all_healthy = all(checks.values())
 
    return {
        "status": "ready" if all_healthy else "not_ready",
        "checks": checks,
    }
 
@router.get("/health/live")
async def liveness():
    """Process này còn sống không?"""
    return {"status": "alive"}

/health/live — Orchestrator (Kubernetes) dùng để biết process có bị stuck không. Nếu fail, restart pod.
/health/ready — Load balancer dùng để biết instance này có thể serve traffic không. Nếu Kafka down, ngừng gửi request đến instance này.

8. Monitoring và Observability

Bạn không thể quản lý thứ bạn không đo được. Ingestion API cần monitoring kỹ lưỡng vì failure là silent — device không gọi support để phàn nàn.

Key Metric Cần Track

from prometheus_client import Counter, Histogram, Gauge
 
# Ingestion metric
messages_received = Counter(
    "ingestion_messages_total",
    "Total messages received",
    ["device_type", "status"],
)
 
ingestion_latency = Histogram(
    "ingestion_latency_seconds",
    "Time to accept and enqueue a message",
    buckets=[0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25],
)
 
queue_depth = Gauge(
    "ingestion_queue_depth",
    "Current message queue depth",
)
 
active_devices = Gauge(
    "ingestion_active_devices",
    "Number of devices that sent data in the last 5 minutes",
)

Dashboard Thiết Yếu

Metric	Alert Threshold	Tại Sao Quan Trọng
Message/sec	Giảm > 30% trong 5 phút	Device có thể mất kết nối
p99 latency	> 100ms	Ingestion quá chậm, risk timeout
Queue depth	> 100K message	Processing không theo kịp ingestion
Error rate (4xx)	> 5%	Nhiều device fail authentication
Error rate (5xx)	> 0.1%	Service của bạn đang fail
Active device	Giảm > 10% trong 15 phút	Network outage hoặc firmware bug

Device Silence Detection

Một trong những alert IoT-specific quan trọng nhất: phát hiện khi device ngừng gửi data.

import asyncio
from datetime import datetime, timedelta
 
# Track last-seen time cho mỗi device
device_last_seen: dict[str, datetime] = {}
 
async def update_last_seen(device_id: str):
    device_last_seen[device_id] = datetime.utcnow()
 
async def detect_silent_devices(
    expected_interval_seconds: int = 60,
    silence_multiplier: int = 3,
):
    """Flag device chưa report trong 3x expected interval."""
    threshold = timedelta(seconds=expected_interval_seconds * silence_multiplier)
    now = datetime.utcnow()
 
    silent = [
        device_id
        for device_id, last_seen in device_last_seen.items()
        if now - last_seen > threshold
    ]
 
    if silent:
        await publish_alert({
            "type": "device_silence",
            "devices": silent,
            "threshold_seconds": threshold.total_seconds(),
        })

9. Scaling Consideration

Horizontal Scaling

Ingestion layer là stateless — scale horizontally bằng cách thêm pod sau load balancer. Message queue decouple ingestion khỏi processing, nên mỗi phần scale độc lập.

Con Số Scale Quan Trọng

Scale	Device	Message/sec	Architecture
Nhỏ	< 1,000	< 1,000	Single instance + PostgreSQL
Trung bình	1K - 100K	1K - 50K	3-5 instance + Kafka + TimescaleDB
Lớn	100K - 1M	50K - 500K	Auto-scaling cluster + Kafka + multi-tier storage
Khổng lồ	> 1M	> 500K	Regional ingestion + global aggregation

Chọn Protocol Theo Scale

< 10K device: HTTP REST là đủ. Đơn giản, dễ hiểu, dễ debug.
10K - 100K device: MQTT hoặc HTTP với aggressive batching.
> 100K device: MQTT với QoS 1, edge gateway cho aggregation, regional ingestion endpoint.

10. Những Pitfall Thường Gặp

Pitfall 1: Synchronous Processing

# ❌ Đừng làm thế này — processing block response
@app.post("/api/v1/ingest")
async def ingest(payload: TelemetryPayload):
    await write_to_database(payload)        # Chậm!
    await run_anomaly_detection(payload)     # Chậm hơn!
    await update_dashboard(payload)          # Chậm nhất!
    return {"status": "ok"}
 
# ✅ Làm thế này — accept nhanh, xử lý sau
@app.post("/api/v1/ingest", status_code=202)
async def ingest(payload: TelemetryPayload):
    await enqueue(payload)                   # Nhanh!
    return {"status": "accepted"}

Pitfall 2: Không Có Idempotency

Device retry. Network trục trặc. Nếu ingestion API không handle duplicate message, data của bạn sẽ bẩn.

# ✅ Dùng deduplication key
@app.post("/api/v1/ingest", status_code=202)
async def ingest(
    payload: TelemetryPayload,
    x_idempotency_key: str = Header(None),
):
    # Generate key từ device_id + timestamp nếu không có
    dedup_key = x_idempotency_key or f"{payload.device_id}:{payload.timestamp.isoformat()}"
 
    if await is_duplicate(dedup_key):
        return {"status": "duplicate"}
 
    await enqueue(payload)
    await mark_as_seen(dedup_key, ttl_seconds=3600)
 
    return {"status": "accepted"}

Pitfall 3: Tin Tưởng Device Timestamp

Device clock bị drift. Đôi khi drift nặng. Luôn record server-side timestamp và dùng nó làm authoritative time cho ordering.

enriched_payload = {
    **payload.model_dump(),
    "server_received_at": datetime.utcnow().isoformat(),
    "time_drift_seconds": (
        datetime.utcnow() - payload.timestamp
    ).total_seconds(),
}

Pitfall 4: Không Có Dead Letter Queue

Khi message fail processing, nó đi đâu? Nếu câu trả lời là "biến mất", bạn có data loss problem.

async def process_message(message):
    try:
        validated = validate(message)
        await store(validated)
    except ValidationError as e:
        # Không để mất — gửi đến DLQ để investigate
        await send_to_dlq(message, reason=str(e))
    except Exception as e:
        # Unexpected error — DLQ với đầy đủ context
        await send_to_dlq(message, reason=str(e), retryable=True)

Tổng Kết

Thiết kế ingestion API cho IoT khác biệt cơ bản so với xây dựng web API thông thường. Những nguyên tắc chính:

Accept nhanh, xử lý sau — luôn trả 202, enqueue cho async processing
Design cho failure — device retry, network fail, clock drift. Handle hết
Validation theo layer — reject bad syntax ngay, flag bad semantic để review
Implement backpressure — rate limiting, load shedding, và edge buffering
Authenticate mọi device — API key cho dev, mTLS hoặc JWT cho production
Dùng multi-tier storage — hot data trong time-series DB, cold data trong data lake
Monitor sự im lặng — trong IoT, sự vắng mặt của data thường là signal quan trọng nhất

Architecture pattern hoạt động ở mọi scale:

Device → Ingestion API → Message Queue → Stream Processor → Storage Tier

Bắt đầu đơn giản với HTTP và single database. Thêm MQTT, Kafka, và multi-tier storage khi fleet device của bạn lớn lên. Ingestion API pattern giữ nguyên — chỉ infrastructure phía sau nó scale.

Đọc Thêm

REST API Complete Guide — nền tảng API design
API Gateway Complete Guide — quản lý API traffic ở scale lớn
HTTP Protocol Guide — hiểu protocol layer