Part 3: Time Series Database 통합과 배포 - 현대적 TDB 생태계 완성

# microservices/timeseries-service/main.py
from fastapi import FastAPI, HTTPException, Depends
from influxdb_client import InfluxDBClient
from pydantic import BaseModel
from typing import List, Optional
import asyncio
import json
from datetime import datetime, timedelta
import logging

app = FastAPI(title="Time Series Service", version="1.0.0")

# 설정
INFLUXDB_URL = "http://influxdb:8086"
INFLUXDB_TOKEN = "admin-token"
INFLUXDB_ORG = "myorg"
INFLUXDB_BUCKET = "mybucket"

# InfluxDB 클라이언트
influx_client = InfluxDBClient(
    url=INFLUXDB_URL,
    token=INFLUXDB_TOKEN,
    org=INFLUXDB_ORG
)

class TimeSeriesData(BaseModel):
    measurement: str
    tags: dict
    fields: dict
    timestamp: Optional[datetime] = None

class QueryRequest(BaseModel):
    start: datetime
    stop: datetime
    measurement: str
    filters: Optional[dict] = None
    aggregation: Optional[str] = None

class TimeSeriesService:
    def __init__(self, client: InfluxDBClient):
        self.client = client
        self.write_api = client.write_api()
        self.query_api = client.query_api()
    
    async def write_data(self, data: TimeSeriesData) -> bool:
        """시계열 데이터 쓰기"""
        try:
            point = {
                "measurement": data.measurement,
                "tags": data.tags,
                "fields": data.fields,
                "time": data.timestamp or datetime.utcnow()
            }
            
            self.write_api.write(
                bucket=INFLUXDB_BUCKET,
                record=point
            )
            
            return True
        except Exception as e:
            logging.error(f"데이터 쓰기 실패: {e}")
            return False
    
    async def query_data(self, request: QueryRequest) -> List[dict]:
        """시계열 데이터 쿼리"""
        try:
            # Flux 쿼리 구성
            flux_query = f'''
            from(bucket: "{INFLUXDB_BUCKET}")
            |> range(start: {request.start.isoformat()}, stop: {request.stop.isoformat()})
            |> filter(fn: (r) => r._measurement == "{request.measurement}")
            '''
            
            # 필터 추가
            if request.filters:
                for key, value in request.filters.items():
                    flux_query += f'|> filter(fn: (r) => r.{key} == "{value}")\n'
            
            # 집계 추가
            if request.aggregation:
                flux_query += f'|> {request.aggregation}()\n'
            
            # 쿼리 실행
            result = self.query_api.query(flux_query)
            
            # 결과 변환
            data_points = []
            for table in result:
                for record in table.records:
                    data_points.append({
                        "time": record.get_time(),
                        "measurement": record.get_measurement(),
                        "field": record.get_field(),
                        "value": record.get_value(),
                        "tags": record.values
                    })
            
            return data_points
            
        except Exception as e:
            logging.error(f"데이터 쿼리 실패: {e}")
            raise HTTPException(status_code=500, detail=str(e))

# 서비스 인스턴스
ts_service = TimeSeriesService(influx_client)

@app.post("/data")
async def write_timeseries_data(data: TimeSeriesData):
    """시계열 데이터 쓰기 엔드포인트"""
    success = await ts_service.write_data(data)
    if success:
        return {"status": "success", "message": "데이터가 성공적으로 저장되었습니다"}
    else:
        raise HTTPException(status_code=500, detail="데이터 저장 실패")

@app.post("/query")
async def query_timeseries_data(request: QueryRequest):
    """시계열 데이터 쿼리 엔드포인트"""
    data = await ts_service.query_data(request)
    return {
        "status": "success",
        "data": data,
        "count": len(data)
    }

@app.get("/health")
async def health_check():
    """헬스 체크"""
    return {"status": "healthy", "timestamp": datetime.utcnow()}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

# microservices/data-ingestion/main.py
import asyncio
import json
import logging
from datetime import datetime
from kafka import KafkaConsumer, KafkaProducer
from influxdb_client import InfluxDBClient
import random

class DataIngestionPipeline:
    def __init__(self):
        self.kafka_consumer = KafkaConsumer(
            'sensor-data',
            bootstrap_servers=['kafka:9092'],
            value_deserializer=lambda x: json.loads(x.decode('utf-8')),
            group_id='data-ingestion-group'
        )
        
        self.kafka_producer = KafkaProducer(
            bootstrap_servers=['kafka:9092'],
            value_serializer=lambda x: json.dumps(x).encode('utf-8')
        )
        
        self.influx_client = InfluxDBClient(
            url="http://influxdb:8086",
            token="admin-token",
            org="myorg"
        )
        
        self.write_api = self.influx_client.write_api()
    
    async def generate_sensor_data(self):
        """센서 데이터 생성 시뮬레이터"""
        sensor_types = ['temperature', 'humidity', 'pressure', 'vibration']
        locations = ['seoul', 'busan', 'daegu', 'incheon']
        
        while True:
            data = {
                'timestamp': datetime.utcnow().isoformat(),
                'sensor_type': random.choice(sensor_types),
                'location': random.choice(locations),
                'value': round(random.uniform(20, 30), 2),
                'quality': random.randint(80, 100)
            }
            
            # Kafka로 전송
            self.kafka_producer.send('sensor-data', data)
            
            await asyncio.sleep(1)  # 1초마다 데이터 생성
    
    async def process_kafka_data(self):
        """Kafka에서 데이터를 읽어서 InfluxDB에 저장"""
        for message in self.kafka_consumer:
            try:
                data = message.value
                
                # 데이터 검증
                if self.validate_data(data):
                    # InfluxDB에 저장
                    await self.store_to_influxdb(data)
                    
                    # 분석용으로 처리된 데이터 전송
                    processed_data = self.process_data(data)
                    self.kafka_producer.send('processed-data', processed_data)
                    
                else:
                    logging.warning(f"유효하지 않은 데이터: {data}")
                    
            except Exception as e:
                logging.error(f"데이터 처리 오류: {e}")
    
    def validate_data(self, data):
        """데이터 유효성 검증"""
        required_fields = ['timestamp', 'sensor_type', 'location', 'value', 'quality']
        
        if not all(field in data for field in required_fields):
            return False
        
        # 값 범위 검증
        if data['value'] < -50 or data['value'] > 100:
            return False
        
        if data['quality'] < 0 or data['quality'] > 100:
            return False
        
        return True
    
    async def store_to_influxdb(self, data):
        """InfluxDB에 데이터 저장"""
        try:
            point = {
                "measurement": "sensor_data",
                "tags": {
                    "sensor_type": data['sensor_type'],
                    "location": data['location']
                },
                "fields": {
                    "value": data['value'],
                    "quality": data['quality']
                },
                "time": data['timestamp']
            }
            
            self.write_api.write(
                bucket="mybucket",
                record=point
            )
            
            logging.info(f"데이터 저장 완료: {data['sensor_type']} at {data['location']}")
            
        except Exception as e:
            logging.error(f"InfluxDB 저장 실패: {e}")
    
    def process_data(self, data):
        """데이터 처리 및 변환"""
        processed = data.copy()
        
        # 이상치 탐지
        if data['value'] > 35 or data['value'] < 15:
            processed['anomaly'] = True
            processed['anomaly_type'] = 'outlier'
        else:
            processed['anomaly'] = False
        
        # 품질 점수 기반 분류
        if data['quality'] >= 95:
            processed['quality_level'] = 'excellent'
        elif data['quality'] >= 85:
            processed['quality_level'] = 'good'
        else:
            processed['quality_level'] = 'poor'
        
        processed['processed_at'] = datetime.utcnow().isoformat()
        
        return processed
    
    async def run(self):
        """파이프라인 실행"""
        logging.info("데이터 수집 파이프라인 시작...")
        
        # 병렬 실행
        await asyncio.gather(
            self.generate_sensor_data(),
            self.process_kafka_data()
        )

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    pipeline = DataIngestionPipeline()
    asyncio.run(pipeline.run())

# microservices/monitoring/main.py
from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time
import asyncio
import aiohttp
from datetime import datetime, timedelta

# Prometheus 메트릭 정의
write_requests_total = Counter('influxdb_write_requests_total', 'Total write requests')
write_request_duration = Histogram('influxdb_write_request_duration_seconds', 'Write request duration')
query_requests_total = Counter('influxdb_query_requests_total', 'Total query requests')
query_request_duration = Histogram('influxdb_query_request_duration_seconds', 'Query request duration')
data_points_total = Counter('data_points_processed_total', 'Total data points processed')
system_health = Gauge('system_health_score', 'System health score (0-100)')

class TDBMonitoring:
    def __init__(self):
        self.influx_url = "http://influxdb:8086"
        self.headers = {
            'Authorization': 'Token admin-token',
            'Content-Type': 'application/json'
        }
    
    async def collect_metrics(self):
        """메트릭 수집"""
        while True:
            try:
                # InfluxDB 상태 확인
                async with aiohttp.ClientSession() as session:
                    # 헬스 체크
                    async with session.get(f"{self.influx_url}/health", headers=self.headers) as response:
                        if response.status == 200:
                            system_health.set(100)
                        else:
                            system_health.set(50)
                    
                    # 데이터 포인트 수 확인
                    query = {
                        'query': 'from(bucket: "mybucket") |> range(start: -1h) |> count()'
                    }
                    
                    start_time = time.time()
                    async with session.post(f"{self.influx_url}/api/v2/query", headers=self.headers, json=query) as response:
                        duration = time.time() - start_time
                        
                        if response.status == 200:
                            query_requests_total.inc()
                            query_request_duration.observe(duration)
                            
                            # 결과에서 카운트 추출
                            result = await response.json()
                            if result and 'results' in result:
                                for table in result['results'][0]['series']:
                                    for row in table['values']:
                                        count = row[1] if len(row) > 1 else 0
                                        data_points_total.inc(count)
                
                # 30초마다 메트릭 수집
                await asyncio.sleep(30)
                
            except Exception as e:
                print(f"메트릭 수집 오류: {e}")
                system_health.set(0)
                await asyncio.sleep(30)

# 메트릭 서버 시작
start_http_server(8000)

# 모니터링 시작
monitor = TDBMonitoring()
asyncio.run(monitor.collect_metrics())