Part 2: Time Series Database 고급 기능과 최적화 - 프로덕션급 TDB 시스템 구축

# cluster-performance-test.py
import asyncio
import aiohttp
import time
import json
from concurrent.futures import ThreadPoolExecutor
import random

class InfluxDBClusterTest:
    def __init__(self, nodes):
        self.nodes = nodes
        self.results = []
    
    async def write_test(self, node_url, num_points=1000):
        """쓰기 성능 테스트"""
        url = f"{node_url}/api/v2/write"
        headers = {
            'Authorization': 'Token admin-token',
            'Content-Type': 'text/plain'
        }
        
        # 테스트 데이터 생성
        data = []
        for i in range(num_points):
            timestamp = int(time.time() * 1000000000) + i
            value = random.uniform(20, 30)
            data.append(f"temperature,location=seoul value={value} {timestamp}")
        
        payload = '\n'.join(data)
        
        start_time = time.time()
        async with aiohttp.ClientSession() as session:
            async with session.post(url, headers=headers, data=payload) as response:
                end_time = time.time()
                
                if response.status == 204:
                    duration = end_time - start_time
                    throughput = num_points / duration
                    return {
                        'node': node_url,
                        'operation': 'write',
                        'points': num_points,
                        'duration': duration,
                        'throughput': throughput,
                        'status': 'success'
                    }
                else:
                    return {
                        'node': node_url,
                        'operation': 'write',
                        'status': 'failed',
                        'error': await response.text()
                    }
    
    async def query_test(self, node_url, num_queries=100):
        """쿼리 성능 테스트"""
        url = f"{node_url}/api/v2/query"
        headers = {
            'Authorization': 'Token admin-token',
            'Content-Type': 'application/json'
        }
        
        query = {
            'query': 'from(bucket: "mybucket") |> range(start: -1h) |> mean()'
        }
        
        start_time = time.time()
        async with aiohttp.ClientSession() as session:
            async with session.post(url, headers=headers, json=query) as response:
                end_time = time.time()
                
                if response.status == 200:
                    duration = end_time - start_time
                    qps = num_queries / duration
                    return {
                        'node': node_url,
                        'operation': 'query',
                        'queries': num_queries,
                        'duration': duration,
                        'qps': qps,
                        'status': 'success'
                    }
                else:
                    return {
                        'node': node_url,
                        'operation': 'query',
                        'status': 'failed',
                        'error': await response.text()
                    }
    
    async def run_comprehensive_test(self):
        """종합 성능 테스트"""
        print("🔥 클러스터 성능 테스트 시작...")
        
        tasks = []
        
        # 모든 노드에서 쓰기 테스트
        for node in self.nodes:
            tasks.append(self.write_test(node, 5000))
        
        # 병렬 실행
        write_results = await asyncio.gather(*tasks)
        
        # 쿼리 테스트
        tasks = []
        for node in self.nodes:
            tasks.append(self.query_test(node, 100))
        
        query_results = await asyncio.gather(*tasks)
        
        # 결과 분석
        self.analyze_results(write_results, query_results)
    
    def analyze_results(self, write_results, query_results):
        """결과 분석 및 리포트"""
        print("\n📊 성능 테스트 결과 분석")
        print("=" * 50)
        
        # 쓰기 성능 분석
        successful_writes = [r for r in write_results if r['status'] == 'success']
        if successful_writes:
            avg_write_throughput = sum(r['throughput'] for r in successful_writes) / len(successful_writes)
            print(f"📈 평균 쓰기 처리량: {avg_write_throughput:.2f} points/sec")
            
            for result in successful_writes:
                print(f"   - {result['node']}: {result['throughput']:.2f} points/sec")
        
        # 쿼리 성능 분석
        successful_queries = [r for r in query_results if r['status'] == 'success']
        if successful_queries:
            avg_query_qps = sum(r['qps'] for r in successful_queries) / len(successful_queries)
            print(f"📈 평균 쿼리 처리량: {avg_query_qps:.2f} queries/sec")
            
            for result in successful_queries:
                print(f"   - {result['node']}: {result['qps']:.2f} queries/sec")
        
        # 권장사항
        print("\n💡 성능 최적화 권장사항:")
        if avg_write_throughput < 10000:
            print("   - 쓰기 성능 개선: 배치 크기 증가, 압축 설정 최적화")
        if avg_query_qps < 100:
            print("   - 쿼리 성능 개선: 인덱스 추가, 캐시 설정")

# 테스트 실행
if __name__ == "__main__":
    # 클러스터 노드 URL 설정
    nodes = [
        "http://influxdb-0.influxdb:8086",
        "http://influxdb-1.influxdb:8086", 
        "http://influxdb-2.influxdb:8086"
    ]
    
    tester = InfluxDBClusterTest(nodes)
    asyncio.run(tester.run_comprehensive_test())