사이트 통계 대시보드 개발기
사이트가 어느 정도 안정화됨에 따라 사용자 동향 및 사이트 주요 지표를 실시간으로 파악할 수 있는 대시보드를 개발하게 되었습니다.
본 대시보드는 운영 중인 다양한 서비스의 상태와 API 요청 현황을 한눈에 확인할 수 있도록 구성하였습니다.
기능 설계
사용자 인증 여부에 관계 없이 수집하는 것으로, 흐름은 다음과 같습니다.
API 호출 로그는 FastAPI의 미들웨어를 통해 Kafka에 스트리밍 처리되어 데이터 적재됩니다.
History Table 설계
사용자 API 요청 History 테이블 정보입니다. 사용자 흔적이라는 의미에서 user_footprint로 만들었습니다.
스키마는 다음과 같습니다.
CREATE TABLE IF NOT EXISTS USER_FOOTPRINT
(
ID SERIAL PRIMARY KEY,
LINK TEXT,
REQUEST TEXT,
FOOTPRINT_TIME timestamp with time zone,
EXECUTE_TIME timestamp with time zone default now()
);
COMMENT ON COLUMN USER_FOOTPRINT.ID IS '자동 생성 아이디';
COMMENT ON COLUMN USER_FOOTPRINT.REQUEST IS '요청 타입';
COMMENT ON COLUMN USER_FOOTPRINT.LINK IS '방문 주소';
COMMENT ON COLUMN USER_FOOTPRINT.FOOTPRINT_TIME IS '전송 시간';
COMMENT ON COLUMN USER_FOOTPRINT.EXECUTE_TIME IS '적재 시간';IP도 수집할 까 했는데 민감한 정보인 거 같아 제거했습니다.
FastAPI 코드 작성
- api/
- util/
-- middleware.py
- main.pyMiddleware 부분에 Kafka로 Produce 하는 부분을 추가하면 됩니다.
필터링을 통해 배포한 서버의 IP와 동일한 경우는 제거하고 있습니다.
import json
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
from starlette.middleware.base import BaseHTTPMiddleware
from util.kafka_producer import produce_message
import logging
import time
from dotenv import load_dotenv
import os
from fastapi.responses import Response
from collections import defaultdict
logger = logging.getLogger("api.access")
load_dotenv()
class KafkaProducerMiddleware(BaseHTTPMiddleware):
request_counts = defaultdict(list)
async def dispatch(self, request, call_next):
start_time = time.time()
path = request.url.path
real_ip = (
request.headers.get("cf-connecting-ip")
or request.headers.get("x-real-ip")
or request.headers.get("x-forwarded-for", "").split(",")[0].strip()
or request.client.host
if request.client
else "unknown"
)
# Rate limiting
now = datetime.now()
self.request_counts[real_ip] = [
t for t in self.request_counts[real_ip] if now - t < timedelta(minutes=1)
]
if len(self.request_counts[real_ip]) > 100:
logger.warning(f"Rate limit exceeded: {real_ip}")
return Response(status_code=429)
self.request_counts[real_ip].append(now)
response = await call_next(request)
process_time = time.time() - start_time
if real_ip != os.getenv("IP"):
logger.info(
f'{real_ip} - "{request.method} {path}" '
f"{response.status_code} - {process_time:.3f}s"
)
data = {
"method": request.method,
"link": path,
"footprint_time": datetime.now(ZoneInfo("Asia/Seoul")).isoformat(),
}
produce_message(json.dumps(data))
return response해당 파일을 작성 후 main.py에 추가해주면 됩니다.
from util.middleware import KafkaProducerMiddleware
app.add_middleware(KafkaProducerMiddleware)Kafka Consumer
Python을 사용하여 별도의 Consumer를 만들었습니다.
Comsume후 데이터를 DB에 넣는 과정을 진행하고 있습니다.
여기는 코드가 많아서 주요한 부분만 올립니다.
# Consume 동작의 기본이 되는 함수 입니다. Config를 매개변수로 받아 해당 topic에 연결 후 대기 합니다.
# Message가 들어오면 매개변수로 받은 callback 함수를 사용하여 가공 후 DB에 적재를 진행합니다.
import json
from confluent_kafka import Consumer, KafkaError
from consumer.logger import logger
def run_consumer(kafka_config, process_message_callback):
consumer = Consumer(
{
"bootstrap.servers": kafka_config["bootstrap.servers"],
"group.id": kafka_config["group.id"],
"auto.offset.reset": kafka_config["auto.offset.reset"],
}
)
topic = kafka_config["topic"]
consumer.subscribe([topic])
logger.info(f"Subscribed to topic: {topic}")
try:
while True:
msg = consumer.poll(1.0)
if msg is None:
continue
if msg.error():
if msg.error().code() == KafkaError._PARTITION_EOF:
logger.warning(
f"End of partition: {msg.topic()} [{msg.partition()}]"
)
else:
logger.error(f"Kafka error: {msg.error().str()}")
continue
try:
data = json.loads(msg.value().decode("utf-8"))
logger.info(f"Received message JSON: {data}")
process_message_callback(data)
except json.JSONDecodeError as e:
logger.error(f"JSON 디코딩 실패: {e}")
except Exception as e:
logger.error(f"메시지 처리 실패: {e}")
except KeyboardInterrupt:
logger.info("Consumer 종료됨")
finally:
consumer.close()
logger.info("Consumer 연결 종료")
# 적재 함수입니다. 적재시간은 auto라서 나머지 정보만 넣어줍니다.
def save_log_to_postgresql(conn, data):
with conn.cursor() as cur:
insert_query = """
INSERT INTO user_footprint (request, link, footprint_time)
VALUES (%s, %s, %s)
"""
cur.execute(
insert_query, (data["method"], data["link"], data["footprint_time"])
)
conn.commit()
logger.info("데이터 PostgreSQL 저장 완료")적재 결과
정상적으로 진행이 된다면 로그가 잘남고, DB에 적재가 잘 되는 것을 확인 할 수 있습니다.
kafka 로그
DB 확인
주요 지표 쿼리
대시보드는 다음과 같은 6가지 핵심 지표로 구성되어 있습니다
1. 총 요청 수
- 선택한 기간 동안의 전체 API 요청 수 집계
WITH current_period AS (
SELECT COUNT(*) AS request_count
FROM USER_FOOTPRINT
WHERE FOOTPRINT_TIME AT TIME ZONE 'Asia/Seoul' >= :start_date
AND FOOTPRINT_TIME AT TIME ZONE 'Asia/Seoul' < :end_date
)
SELECT
c.request_count AS current_requests
FROM current_period c2. 상위 10개 API 엔드포인트
- 요청 수 기준으로 가장 많이 호출된 API 10개 목록
- 각 API별 요청 수 및 상세 정보 제공
SELECT
REQUEST,
LINK,
COUNT(*) AS request_count
FROM USER_FOOTPRINT
WHERE (EXECUTE_TIME AT TIME ZONE 'Asia/Seoul') BETWEEN :start_date AND :end_date
AND LINK NOT LIKE '%health%'
GROUP BY REQUEST, LINK
ORDER BY request_count DESC
LIMIT 103. 시간대별 요청 분포
- 요청이 집중되는 시간대 트렌드 시각화
- 피크 시간 파악 및 성능 튜닝 참고
SELECT
TO_CHAR(
date_trunc('hour', EXECUTE_TIME AT TIME ZONE 'Asia/Seoul') +
INTERVAL '1 minute' * (FLOOR(EXTRACT(MINUTE FROM EXECUTE_TIME AT TIME ZONE 'Asia/Seoul') / 15) * 15),
'HH24:MI'
) AS time,
COUNT(*) AS requests
FROM USER_FOOTPRINT
WHERE (EXECUTE_TIME AT TIME ZONE 'Asia/Seoul') BETWEEN :start_date AND :end_date
GROUP BY time
ORDER BY time4. 서비스 평균 응답 시간
- FastAPI, Next.js 기준 평균 응답 시간
SELECT
service_name,
ROUND(AVG(response_ms::DOUBLE PRECISION) * 1000) AS avg_response_ms
FROM
response_time
WHERE
(checked_time AT TIME ZONE 'Asia/Seoul') BETWEEN :start_date AND :end_date
AND response_ms IS NOT NULL
GROUP BY
service_name5. 사용자 정보
- 전체 가입 사용자 수
- 선택 기간 내 활성 사용자 수 (출석일 기준)
SELECT COUNT(*) AS user_total_count
FROM USER_INFO
SELECT COUNT(*) AS active_user
FROM USER_INFO
WHERE (attendance_time AT TIME ZONE 'Asia/Seoul') BETWEEN :start_date AND :end_date6. 서비스 상태 모니터링
- 주요 서비스별 실시간 상태 확인 가능
| 서비스명 | 설명 |
|---|---|
| Airflow | 배치 및 워크플로 관리 |
| FastAPI | 백엔드 API 서버 |
| Kafka Topic | 실시간 로그 수집 |
| Minio | 오브젝트 스토리지 |
| Next.js | 프론트엔드 서버 |
| Nginx Proxy Manager | 리버스 프록시 및 SSL 관리 |
| PostgreSQL | 운영 데이터베이스 |
| Redis | 외부 프로그램, 사용자 알림 캐시 메모리 |
SELECT
service_name,
COUNT(*) AS total,
SUM(CASE WHEN status = 'OK' THEN 1 ELSE 0 END) AS ok_count,
SUM(CASE WHEN status = 'FAIL' THEN 1 ELSE 0 END) AS fail_count,
ROUND(SUM(CASE WHEN status = 'OK' THEN 1 ELSE 0 END)::numeric / COUNT(*) * 100, 2) AS ok_percentage,
ROUND(SUM(CASE WHEN status = 'FAIL' THEN 1 ELSE 0 END)::numeric / COUNT(*) * 100, 2) AS fail_percentage
FROM health_check
WHERE (checked_time AT TIME ZONE 'Asia/Seoul') BETWEEN :start_date AND :end_date
GROUP BY service_name
ORDER BY service_nameNextJS 구현
Front에서는 전달 받은 값을 별도의 가공 없이 그대로 사용하면 되어서, 간단하게 구현할 수 있습니다.
저는 recharts를 사용하여 그래프를 그렸습니다.
시간대별 요청분포
import {
AreaChart,
Area,
XAxis,
YAxis,
CartesianGrid,
Tooltip,
ResponsiveContainer,
} from "recharts";
import { Clock, TrendingUp } from "lucide-react";
import { ALL_COLOR } from "@/lib/consts/colorConsts";
import { HourlyDistributionChartTypes } from "../dashboard.types";
import { dashboardI18N } from "@/lib/consts/i18nConsts";
import { getLocaleKey } from "@/lib/func/localeFunction";
import { useLocale } from "next-intl";
export default function HourlyDistributionChart({
time_distribution,
}: HourlyDistributionChartTypes) {
const locale = useLocale();
const localeKey = getLocaleKey(locale);
const maxRequestEntry = time_distribution.reduce((max, current) => {
return current.requests > max.requests ? current : max;
});
return (
<div className="bg-gray-800 rounded-lg shadow-2xl border border-gray-700">
<div className="p-6 pb-2">
<h3 className="text-lg font-semibold text-white flex items-center space-x-2">
<Clock className="h-5 w-5 text-green-400" />
<span>{dashboardI18N.requestsByHour[localeKey]}</span>
</h3>
<p className="text-sm text-gray-300 mt-1">
{dashboardI18N.apiRequestPatternByHour[localeKey]} (
{maxRequestEntry.time} {dashboardI18N.focus[localeKey]})
</p>
</div>
<div className="p-6 pt-0 h-96">
<ResponsiveContainer width="100%" height="100%">
<AreaChart
data={time_distribution}
margin={{ top: 10, right: 30, left: 0, bottom: 0 }}
>
<defs>
<linearGradient id="colorRequests" x1="0" y1="0" x2="0" y2="1">
<stop offset="5%" stopColor="#60a5fa" stopOpacity={0.8} />
<stop offset="95%" stopColor="#60a5fa" stopOpacity={0.1} />
</linearGradient>
</defs>
<CartesianGrid strokeDasharray="3 3" stroke="#374151" />
<XAxis
dataKey="time"
tick={{ fontSize: 12, fill: ALL_COLOR.White }}
angle={-45}
textAnchor="end"
height={60}
/>
<YAxis tick={{ fill: ALL_COLOR.White }} />
<Tooltip
formatter={(value) => {
const safeValue = value ?? 0;
return [
`${safeValue.toLocaleString()} ${
dashboardI18N.request[localeKey]
}`,
`${dashboardI18N.requestCount[localeKey]}`,
];
}}
labelFormatter={(label) =>
`${dashboardI18N.time[localeKey]}: ${label}`
}
contentStyle={{
backgroundColor: "#1f2937",
border: "1px solid #374151",
borderRadius: "8px",
color: "#ffffff",
}}
/>
<Area
type="monotone"
dataKey="requests"
stroke="#60a5fa"
fillOpacity={1}
fill="url(#colorRequests)"
/>
</AreaChart>
</ResponsiveContainer>
{/* 피크 시간 정보 */}
<div className="mt-4 p-3 bg-blue-900/30 rounded-lg border border-blue-700/50">
<div className="flex items-center space-x-2 text-blue-300">
<TrendingUp className="h-4 w-4" />
<span className="font-semibold">
{dashboardI18N.peakTime[localeKey]}
</span>
</div>
<p className="text-sm text-white mt-1">
{maxRequestEntry.time}{" "}
{dashboardI18N.focusedRequestsAtTime[localeKey]} (
{maxRequestEntry.requests.toLocaleString()} !!!)
</p>
</div>
</div>
</div>
);
}
대시보드
완성된 화면입니다.
개발 후기
- Kafka 기반의 로그 수집 및 분석 구조를 설계하면서 실시간 데이터 흐름에 대한 이해가 어느정도 생겼습니다.
- 기존에 Airflow를 통해 health check와 같은 서비스들을 만들어두니 금방 적용할 수 있어서 편했습니다. (잘 만들어 둔 거 같습니다.)
대시보드는 단순한 지표 표시 도구를 넘어서, 서비스 운영 전반에 대한 인사이트를 제공하는 중요한 관제 시스템으로 활용될 예정입니다.
만드는게 좋아