API 게이트웨이 로그 분석은 시스템 안정성과 비용 최적화의 핵심입니다. HolySheep AI를 ELK Stack(Elasticsearch, Logstash, Kibana)와 통합하면 모든 AI 모델 호출 로그를 중앙 집중 관리하고, 성능 병목 구간을 시각화하며, 비정상적 호출 패턴을 실시간 탐지할 수 있습니다. 이 튜토리얼에서는 HolySheep의 /v1/logs 엔드포인트를 활용하여 완전한 로그 파이프라인을 구축하는 방법을 다룹니다.

핵심 결론

서비스 비교: HolySheep vs 공식 API vs 경쟁 서비스

비교 항목 HolySheep AI 공식 OpenAI API 공식 Anthropic API 기타 중개 서비스
API Gateway URL api.holysheep.ai/v1 api.openai.com/v1 api.anthropic.com/v1 제각각
결제 방식 로컬 결제 지원
(신용카드 불필요)		 해외 신용카드 필수 해외 신용카드 필수 해외 신용카드 필수
GPT-4.1 비용 $8.00/MTok $8.00/MTok N/A $8.50~10/MTok
Claude Sonnet 4.5 비용 $15.00/MTok N/A $15.00/MTok $15.50~18/MTok
Gemini 2.5 Flash 비용 $2.50/MTok N/A N/A $3.00~4/MTok
DeepSeek V3.2 비용 $0.42/MTok N/A N/A $0.50~0.60/MTok
로그 관리 ✅ 내장 로그 API
/v1/logs		 ❌ 제한적 ❌ 제한적 ⚠️ 부가 기능
평균 응답 지연 120~180ms 150~200ms 160~220ms 200~350ms
무료 크레딧 ✅ 가입 시 제공 $5 크레딧 $5 크레딧 제한적
동시 연결 제한 제한 없음 tier별 제한 tier별 제한 제한적

이런 팀에 적합 / 비적합

✅ HolySheep가 적합한 팀

❌ HolySheep가 적합하지 않은 팀

사전 요구사항

ELK Stack 아키텍처 개요

┌─────────────────────────────────────────────────────────────────┐
│                    HolySheep AI Logs 파이프라인                     │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  ┌──────────────┐      ┌──────────────┐      ┌──────────────┐   │
│  │ HolySheep    │      │   Python     │      │   Logstash   │   │
│  │  /v1/logs    │ ──▶  │  Collector   │ ──▶  │   Pipeline   │   │
│  │  API         │      │  (Polling)   │      │   :5044      │   │
│  └──────────────┘      └──────────────┘      └──────┬───────┘   │
│                                                      │           │
│                                                      ▼           │
│  ┌──────────────┐      ┌──────────────┐      ┌──────────────┐   │
│  │   Kibana     │ ◀──  │ Elasticsearch│ ◀──  │  Beats       │   │
│  │  Dashboard   │      │   :9200      │      │  (Optional)  │   │
│  └──────────────┘      └──────────────┘      └──────────────┘   │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

1단계: Docker Compose로 ELK Stack 구성

version: '3.8'

services:
  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:8.11.0
    container_name: elasticsearch
    environment:
      - discovery.type=single-node
      - xpack.security.enabled=false
      - "ES_JAVA_OPTS=-Xms2g -Xmx2g"
    ports:
      - "9200:9200"
    volumes:
      - elasticsearch_data:/usr/share/elasticsearch/data
    networks:
      - elk

  logstash:
    image: docker.elastic.co/logstash/logstash:8.11.0
    container_name: logstash
    volumes:
      - ./logstash/pipeline:/usr/share/logstash/pipeline:ro
      - ./logstash/config/logstash.yml:/usr/share/logstash/config/logstash.yml:ro
    ports:
      - "5044:5044"
      - "9600:9600"
    environment:
      - "LS_JAVA_OPTS=-Xms512m -Xmx512m"
    depends_on:
      - elasticsearch
    networks:
      - elk

  kibana:
    image: docker.elastic.co/kibana/kibana:8.11.0
    container_name: kibana
    ports:
      - "5601:5601"
    environment:
      - ELASTICSEARCH_HOSTS=http://elasticsearch:9200
    depends_on:
      - elasticsearch
    networks:
      - elk

volumes:
  elasticsearch_data:

networks:
  elk:
    driver: bridge

2단계: HolySheep 로그 수집기 구현

# holy_sheep_log_collector.py
"""
HolySheep AI 로그 수집기
ELK Stack와 통합하여 API 호출 로그를 실시간 수집
"""

import requests
import time
import json
from datetime import datetime, timezone
from collections import deque
import logging


로깅 설정

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class HolySheepLogCollector:
    """
    HolySheep API 로그 수집 및 ELK Stack 전송 클래스
    """
    
    BASE_URL = "https://api.holysheep.ai/v1"
    
    def __init__(self, api_key: str, logstash_host: str = "localhost", 
                 logstash_port: int = 5044):
        self.api_key = api_key
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
        self.logstash_host = logstash_host
        self.logstash_port = logstash_port
        self.last_log_id = None
        self.buffer = deque(maxlen=100)  # 배치 전송을 위한 버퍼
        
    def fetch_logs(self, limit: int = 100) -> list:
        """
        HolySheep 로그 엔드포인트에서 최신 로그 조회
        """
        endpoint = f"{self.BASE_URL}/logs"
        params = {
            "limit": limit,
            "stream": "false"
        }
        
        if self.last_log_id:
            params["after"] = self.last_log_id
            
        try:
            response = requests.get(
                endpoint,
                headers=self.headers,
                params=params,
                timeout=30
            )
            response.raise_for_status()
            data = response.json()
            
            logs = data.get("data", [])
            if logs:
                self.last_log_id = logs[-1].get("id")
                
            logger.info(f"Successfully fetched {len(logs)} logs from HolySheep")
            return logs
            
        except requests.exceptions.RequestException as e:
            logger.error(f"Failed to fetch logs: {e}")
            return []
    
    def parse_log_entry(self, raw_log: dict) -> dict:
        """
        HolySheep 로그 엔트리를 ELK 포맷으로 변환
        """
        return {
            "@timestamp": raw_log.get("created_at", datetime.now(timezone.utc).isoformat()),
            "log_id": raw_log.get("id"),
            "model": raw_log.get("model"),
            "provider": raw_log.get("provider"),
            "prompt_tokens": raw_log.get("usage", {}).get("prompt_tokens", 0),
            "completion_tokens": raw_log.get("usage", {}).get("completion_tokens", 0),
            "total_tokens": raw_log.get("usage", {}).get("total_tokens", 0),
            "latency_ms": raw_log.get("latency_ms"),
            "status": raw_log.get("status"),
            "error_message": raw_log.get("error", {}).get("message") if raw_log.get("error") else None,
            "cost_usd": self.calculate_cost(raw_log),
            "metadata": raw_log.get("metadata", {})
        }
    
    def calculate_cost(self, log_entry: dict) -> float:
        """
        토큰 사용량 기반 비용 계산 (HolySheep 공식 가격표 기준)
        """
        pricing = {
            "gpt-4.1": 8.00,      # $8.00/MTok
            "gpt-4.1-nano": 0.30,  # $0.30/MTok
            "claude-sonnet-4-5": 15.00,  # $15.00/MTok
            "claude-opus-3-5": 75.00,    # $75.00/MTok
            "gemini-2.5-flash": 2.50,     # $2.50/MTok
            "deepseek-v3.2": 0.42,        # $0.42/MTok
        }
        
        model = log_entry.get("model", "").lower()
        total_tokens = log_entry.get("usage", {}).get("total_tokens", 0)
        rate = pricing.get(model, 8.00)  # 기본값: $8/MTok
        
        return (total_tokens / 1_000_000) * rate
    
    def send_to_logstash(self, parsed_logs: list) -> bool:
        """
        파싱된 로그를 Logstash로 전송
        """
        if not parsed_logs:
            return True
            
        logstash_url = f"http://{self.logstash_host}:5044"
        
        try:
            # Logstash의 Beats 프로토콜 대신 HTTP로 전송
            response = requests.post(
                logstash_url,
                json=parsed_logs,
                headers={"Content-Type": "application/json"},
                timeout=10
            )
            logger.info(f"Sent {len(parsed_logs)} logs to Logstash, status: {response.status_code}")
            return True
            
        except requests.exceptions.RequestException as e:
            logger.error(f"Failed to send logs to Logstash: {e}")
            # 실패 시 로컬 파일에 백업
            self._backup_to_file(parsed_logs)
            return False
    
    def _backup_to_file(self, logs: list):
        """Logstash 연결 실패 시 로컬 파일에 백업"""
        backup_file = f"logs_backup_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
        with open(backup_file, 'a') as f:
            for log in logs:
                f.write(json.dumps(log) + '\n')
        logger.warning(f"Backed up {len(logs)} logs to {backup_file}")
    
    def run(self, interval_seconds: int = 60):
        """
        continuous 로그 수집 실행
        """
        logger.info("Starting HolySheep log collector...")
        logger.info(f"Polling interval: {interval_seconds} seconds")
        
        while True:
            try:
                raw_logs = self.fetch_logs(limit=100)
                
                if raw_logs:
                    parsed_logs = [self.parse_log_entry(log) for log in raw_logs]
                    self.send_to_logstash(parsed_logs)
                    
                    # 통계 로깅
                    total_cost = sum(log["cost_usd"] for log in parsed_logs)
                    logger.info(
                        f"Batch processed: {len(parsed_logs)} logs, "
                        f"estimated cost: ${total_cost:.4f}"
                    )
                    
            except Exception as e:
                logger.error(f"Unexpected error in collector loop: {e}")
                
            time.sleep(interval_seconds)


def main():
    """메인 실행 함수"""
    API_KEY = "YOUR_HOLYSHEEP_API_KEY"
    
    collector = HolySheepLogCollector(
        api_key=API_KEY,
        logstash_host="localhost",
        logstash_port=5044
    )
    
    collector.run(interval_seconds=60)


if __name__ == "__main__":
    main()

3단계: Logstash 파이프라인 설정

# logstash/pipeline/holy_sheep.conf

input {
  # HTTP를 통한 직접 입력 (Python 스크립트에서 사용)
  http {
    port => 5044
    codec => json
    type => "holysheep_api"
  }
  
  # 파일 입력 (백업 파일 처리용)
  file {
    path => "/var/log/holy_sheep_backup/*.json"
    start_position => "beginning"
    sincedb_path => "/dev/null"
    codec => json
    type => "holysheep_backup"
  }
}

filter {
  # HolySheep API 로그 필터링
  if [type] == "holysheep_api" {
    
    # 타임스탬프 파싱
    date {
      match => ["@timestamp", "ISO8601"]
      target => "@timestamp"
    }
    
    # 비용 데이터 변환
    mutate {
      convert => {
        "prompt_tokens" => "integer"
        "completion_tokens" => "integer"
        "total_tokens" => "integer"
        "latency_ms" => "float"
        "cost_usd" => "float"
      }
    }
    
    # 에러 로그 태깅
    if [status] == "error" {
      mutate {
        add_tag => ["error", "api_failure"]
      }
    }
    
    # 높은 지연 시간 (>500ms) 태깅
    if [latency_ms] and [latency_ms] > 500 {
      mutate {
        add_tag => ["high_latency"]
      }
    }
    
    # 모델별 필드 추가
    if [model] =~ /gpt-4/ {
      mutate {
        add_field => { "provider_category" => "openai" }
      }
    } else if [model] =~ /claude/ {
      mutate {
        add_field => { "provider_category" => "anthropic" }
      }
    } else if [model] =~ /gemini/ {
      mutate {
        add_field => { "provider_category" => "google" }
      }
    } else if [model] =~ /deepseek/ {
      mutate {
        add_field => { "provider_category" => "deepseek" }
      }
    }
    
    # 비용 경보 필드
    if [cost_usd] and [cost_usd] > 1.0 {
      mutate {
        add_tag => ["high_cost"]
      }
    }
    
    # 불필요 필드 제거
    mutate {
      remove_field => ["host", "port"]
    }
  }
}

output {
  elasticsearch {
    hosts => ["elasticsearch:9200"]
    index => "holysheep-logs-%{+YYYY.MM.dd}"
    document_type => "_doc"
  }
  
  # 디버그용 stdout (개발 시에만 활성화)
  # stdout {
  #   codec => rubydebug
  # }
  
  # Slack 알림 (에러 발생 시)
  if "error" in [tags] {
    http {
      url => "${SLACK_WEBHOOK_URL:}"
      http_method => post
      content_type => "application/json"
      format => "message"
      message => '{"text":"🚨 HolySheep API Error Detected","blocks":[{"type":"section","text":{"type":"mrkdwn","text":"*API Error Alert*\n• Model: %{model}\n• Error: %{error_message}\n• Time: %{@timestamp}"}}]}'
    }
  }
}

4단계: Kibana 대시보드 설정

{
  "version": "8.11.0",
  "objects": [
    {
      "id": "holysheep-dashboard",
      "type": "dashboard",
      "attributes": {
        "title": "HolySheep AI - API 로그 분석 대시보드",
        "description": "AI API 호출 패턴, 비용, 성능 모니터링",
        "panelsJSON": [
          {
            "version": "8.11.0",
            "type": "lens",
            "gridData": {"x": 0, "y": 0, "w": 12, "h": 8},
            "panelIndex": "1",
            "title": "일별 API 호출량",
            "embeddableConfig": {
              "references": [
                {
                  "id": "holysheep-logs-*",
                  "name": "indexpattern-datasource-layer-0",
                  "type": "index-pattern"
                }
              ],
              "state": {
                "datasourceStates": {
                  "indexpattern": {
                    "layers": [
                      {
                        "id": "layer-0",
                        "columns": [
                          {"id": "model", "label": "Model", "dataType": "string"},
                          {"id": "total_tokens", "label": "Total Tokens", "dataType": "number"},
                          {"id": "cost_usd", "label": "Cost (USD)", "dataType": "number"},
                          {"id": "latency_ms", "label": "Latency (ms)", "dataType": "number"}
                        ]
                      }
                    ]
                  }
                }
              }
            }
          },
          {
            "version": "8.11.0",
            "type": "lens",
            "gridData": {"x": 12, "y": 0, "w": 12, "h": 8},
            "panelIndex": "2",
            "title": "모델별 비용 분포"
          },
          {
            "version": "8.11.0",
            "type": "lens",
            "gridData": {"x": 0, "y": 8, "w": 16, "h": 8},
            "panelIndex": "3",
            "title": "응답 시간 히스토그램"
          },
          {
            "version": "8.11.0",
            "type": "lens",
            "gridData": {"x": 16, "y": 8, "w": 8, "h": 8},
            "panelIndex": "4",
            "title": "에러율"
          }
        ],
        "timeRestore": true,
        "timeTo": "now",
        "timeFrom": "now-24h",
        "refreshInterval": {
          "pause": false,
          "value": 60000
        }
      }
    },
    {
      "id": "holysheep-alerts",
      "type": "alert",
      "attributes": {
        "name": "높은 비용 경보",
        "alertTypeId": "metrics.alert.threshold",
        "schedule": {"interval": "1h"},
        "params": {
          "criteria": [
            {
              "metric": "cost_usd",
              "condition": ">",
              "threshold": [10.0],
              "timeSize": 1,
              "timeUnit": "h"
            }
          ],
          "filterQueryText": "cost_usd > 10"
        },
        "actions": [
          {
            "group": "Slack",
            "id": "slack-action",
            "params": {
              "message": "🚨 HolySheep API 비용 경보: 지난 1시간 동안 $10 이상 사용됨"
            }
          }
        ]
      }
    }
  ]
}

실전 모니터링 시나리오

# monitoring_queries.py
"""
Kibana Lens 쿼리 예제: HolySheep 로그 분석
"""


=== 시나리오 1: 월간 비용 분석 ===

monthly_cost_query = {
  "query": {
    "bool": {
      "must": [
        {
          "range": {
            "@timestamp": {
              "gte": "now-30d/d",
              "lte": "now/d"
            }
          }
        }
      ]
    }
  },
  "aggs": {
    "total_cost": {
      "sum": {"field": "cost_usd"}
    },
    "by_model": {
      "terms": {"field": "model.keyword", "size": 20},
      "aggs": {
        "cost": {"sum": {"field": "cost_usd"}},
        "tokens": {"sum": {"field": "total_tokens"}},
        "requests": {"value_count": {"field": "log_id"}},
        "avg_latency": {"avg": {"field": "latency_ms"}}
      }
    },
    "by_day": {
      "date_histogram": {
        "field": "@timestamp",
        "calendar_interval": "day"
      },
      "aggs": {
        "daily_cost": {"sum": {"field": "cost_usd"}}
      }
    }
  }
}


=== 시나리오 2: 에러 패턴 분석 ===

error_analysis_query = {
  "query": {
    "bool": {
      "must": [
        {"term": {"status": "error"}},
        {"range": {"@timestamp": {"gte": "now-7d/d"}}}
      ]
    }
  },
  "aggs": {
    "error_types": {
      "terms": {"field": "error_message.keyword", "size": 10}
    },
    "error_by_model": {
      "terms": {"field": "model.keyword"},
      "aggs": {
        "count": {"value_count": {"field": "log_id"}}
      }
    },
    "error_timeline": {
      "date_histogram": {
        "field": "@timestamp",
        "calendar_interval": "1h"
      }
    }
  },
  "size": 0
}


=== 시나리오 3: 토큰 사용 최적화 기회 탐지 ===

optimization_query = {
  "query": {
    "bool": {
      "must": [
        {"range": {"@timestamp": {"gte": "now-30d/d"}}},
        {"range": {"total_tokens": {"gte": 100000}}}  # 100K 토큰 이상 사용
      ]
    }
  },
  "aggs": {
    "high_usage_requests": {
      "filter": {"range": {"total_tokens": {"gte": 50000}}},
      "aggs": {
        "avg_completion_ratio": {
          "scripted_metric": {
            "init_script": "params._all.tokens = []",
            "map_script": """
              params._all.tokens.add([
                'prompt': doc['prompt_tokens'].value,
                'completion': doc['completion_tokens'].value,
                'total': doc['total_tokens'].value
              ])
            """,
            "combine_script": "return params._all.tokens",
            "reduce_script": """
              long promptSum = 0;
              long completionSum = 0;
              for (t in states[0]) {
                promptSum += t.prompt;
                completionSum += t.completion;
              }
              return promptSum > 0 ? (double)completionSum / promptSum : 0;
            """
          }
        }
      }
    }
  }
}


=== 시나리오 4: 지연 시간 이상 탐지 ===

latency_anomaly_query = {
  "query": {
    "bool": {
      "must": [
        {"range": {"@timestamp": {"gte": "now-24h/h"}}},
        {"exists": {"field": "latency_ms"}}
      ]
    }
  },
  "aggs": {
    "latency_stats": {
      "extended_stats": {"field": "latency_ms"},
      "aggs": {
        "percentiles": {
          "percentiles": {
            "field": "latency_ms",
            "percents": [50, 90, 95, 99]
          }
        },
        "by_model": {
          "terms": {"field": "model.keyword"},
          "aggs": {
            "avg_latency": {"avg": {"field": "latency_ms"}},
            "max_latency": {"max": {"field": "latency_ms"}}
          }
        }
      }
    },
    "outliers": {
      "filter": {
        "range": {
          "latency_ms": {"gt": 500}  # 500ms 이상
        }
      },
      "aggs": {
        "slow_requests": {
          "top_hits": {
            "size": 10,
            "sort": [{"latency_ms": "desc"}],
            "_source": ["model", "latency_ms", "@timestamp"]
          }
        }
      }
    }
  }
}


def execute_query(es_client, index_name: str, query: dict):
    """Elasticsearch 쿼리 실행 헬퍼"""
    return es_client.search(index=index_name, body=query)



Kibana Visualize API로 차트 생성

def create_kibana_visualization(es_client):
    """Kibana Lens 시각화 생성"""
    
    # 누적 막대 차트: 모델별 비용
    cost_by_model = {
        "title": "모델별 API 비용 (누적)",
        "type": "histogram",
        "aggs": [
            {
                "id": "1",
                "type": "sum",
                "schema": "metric",
                "params": {"field": "cost_usd"}
            },
            {
                "id": "2",
                "type": "terms",
                "schema": "segment",
                "params": {
                    "field": "model.keyword",
                    "orderBy": "1",
                    "order": "desc",
                    "size": 10
                }
            }
        ]
    }
    
    # 라인 차트: 시간별 응답 시간
    latency_trend = {
        "title": "응답 시간 추이 (95 percentile)",
        "type": "line",
        "aggs": [
            {
                "id": "1",
                "type": "percentiles",
                "schema": "metric",
                "params": {
                    "field": "latency_ms",
                    "percents": [95]
                }
            },
            {
                "id": "2",
                "type": "date_histogram",
                "schema": "segment",
                "params": {
                    "field": "@timestamp",
                    "interval": "auto"
                }
            }
        ]
    }
    
    return [cost_by_model, latency_trend]

자주 발생하는 오류와 해결책

오류 1: 401 Unauthorized - API 키 인증 실패

# ❌ 잘못된 예시 (공식 API URL 사용)
BASE_URL = "https://api.openai.com/v1"  # 절대 사용 금지


✅ 올바른 예시 (HolySheep Gateway 사용)

BASE_URL = "https://api.holysheep.ai/v1"


원인: HolySheep API 키는 api.holysheep.ai에서만 유효


해결: .env 파일에 올바른 base_url과 API 키 설정

import os
from dotenv import load_dotenv

load_dotenv()

class HolySheepConfig:
    API_KEY = os.getenv("HOLYSHEEP_API_KEY")  # HolySheep 대시보드에서 발급
    BASE_URL = "https://api.holysheep.ai/v1"  # 고정값
    
    if not API_KEY or API_KEY == "YOUR_HOLYSHEEP_API_KEY":
        raise ValueError(
            "HolySheep API 키가 설정되지 않았습니다. "
            "https://www.holysheep.ai/register 에서 발급받아 .env 파일에 "
            "HOLYSHEEP_API_KEY=your_key_here 형식으로 저장하세요."
        )

오류 2: Logstash 5044 포트 연결 거부

# ❌ 흔한 실수: Docker 네트워크 미설정

docker-compose.yml의 networks 설정을 누락한 경우



✅ 해결 방법 1: 동일한 Docker 네트워크에 포함


docker-compose.yml에서 모든 서비스를 elk 네트워크에 연결

services:
  logstash:
    networks:
      - elk
  my_collector:
    networks:
      - elk  # 이 줄이 없으면 연결 불가


✅ 해결 방법 2: Docker Compose 네트워킹 명시적 설정

networks:
  elk:
    driver: bridge
    ipam:
      config:
        - subnet: 172.28.0.0/16


✅ 해결 방법 3: 컬렉터에서 localhost 대신 컨테이너 이름 사용

collector = HolySheepLogCollector(
    api_key=API_KEY,
    logstash_host="logstash",  # localhost 대신 컨테이너 이름
    logstash_port=5044
)


연결 테스트

import socket
def test_logstash_connection(host: str, port: int) -> bool:
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(5)
        result = sock.connect_ex((host, port))
        sock.close()
        return result == 0
    except Exception:
        return False


테스트 실행

if not test_logstash_connection("logstash", 5044):
    print("⚠️ Logstash 연결 실패. Docker 네트워크 설정을 확인하세요.")

오류 3: Elasticsearch 색인 실패 - 매핑 오류

# ❌ 잘못된 데이터 타입으로 색인 시도시 발생

예: 문자열 필드에 정수 연산 시도



✅ 해결: Elasticsearch 매핑 사전 생성

from elasticsearch import Elasticsearch

def create_index_with_mapping(es_host: str = "localhost:9200"):
    es = Elasticsearch([es_host])
    index_name = "holysheep-logs"
    
    mapping = {
        "mappings": {
            "properties": {
                "@timestamp": {"type": "date"},
                "log_id": {"type": "keyword"},
                "model": {"type": "keyword"},
                "provider": {"type": "keyword"},
                "provider_category": {"type": "keyword"},
                "prompt_tokens": {"type": "long"},
                "completion_tokens": {"type": "long"},
                "total_tokens": {"type": "long"},
                "latency_ms": {"type": "float"},
                "cost_usd": {"type": "float"},
                "status": {"type": "keyword"},
                "error_message": {"type": "text"},
                "tags": {"type": "keyword"}
            }
        },
        "settings": {
            "number_of_shards": 1,
            "number_of_replicas": 0,
            "index.lifecycle.name": "holysheep-logs-policy"
        }
    }
    
    if not es.indices.exists(index=index_name):
        es.indices.create(index=index_name, body=mapping)
        print(f"✅ Created index: {index_name}")
    else:
        print(f"ℹ️ Index {index_name} already exists")


Kibana에서 인덱스 패턴 생성 명령

kibana_setup = {
    "attributes": {

관련 리소스
관련 문서