[アップデート] Amazon OpenSearch Service のログ分析最適化エンジン「Optimized」が利用可能になったので試してみた

#!/usr/bin/env python3
"""SigV4 署名付きで Amazon OpenSearch Service ドメインに HTTP リクエストを送る。

usage: os_request.py <METHOD> <https://endpoint/path> [--data-file FILE | --data JSON] [--region REGION]
"""
import argparse
import json
import sys

import boto3
from botocore.awsrequest import AWSRequest
from botocore.auth import SigV4Auth
import urllib.request
import urllib.error

def main():
    p = argparse.ArgumentParser()
    p.add_argument("method")
    p.add_argument("url")
    p.add_argument("--data-file")
    p.add_argument("--data")
    p.add_argument("--region", default="ap-northeast-1")
    p.add_argument("--content-type", default="application/json")
    args = p.parse_args()

    body = None
    if args.data_file:
        with open(args.data_file, "rb") as f:
            body = f.read()
    elif args.data:
        body = args.data.encode("utf-8")

    session = boto3.Session()
    creds = session.get_credentials().get_frozen_credentials()
    req = AWSRequest(method=args.method, url=args.url, data=body,
                     headers={"Content-Type": args.content_type})
    SigV4Auth(creds, "es", args.region).add_auth(req)

    http_req = urllib.request.Request(args.url, data=body, method=args.method,
                                      headers=dict(req.headers))
    try:
        with urllib.request.urlopen(http_req, timeout=120) as resp:
            print(f"HTTP {resp.status}", file=sys.stderr)
            out = resp.read().decode("utf-8")
    except urllib.error.HTTPError as e:
        print(f"HTTP {e.code}", file=sys.stderr)
        out = e.read().decode("utf-8")
        print(out)
        sys.exit(1)

    try:
        print(json.dumps(json.loads(out), indent=2, ensure_ascii=False))
    except json.JSONDecodeError:
        print(out)

if __name__ == "__main__":
    main()

{
  "mappings": {
    "properties": {
      "@timestamp": { "type": "date" },
      "level": { "type": "keyword" },
      "service": { "type": "keyword" },
      "host": { "type": "keyword" },
      "status_code": { "type": "integer" },
      "latency_ms": { "type": "integer" },
      "message": { "type": "text" }
    }
  }
}

#!/usr/bin/env python3
"""アプリケーションログ風のサンプルデータを OpenSearch Bulk API 形式で生成する。"""
import json
import random
import sys
from datetime import datetime, timedelta, timezone

random.seed(42)

SERVICES = ["checkout-api", "payment-api", "inventory-api", "auth-api", "search-api"]
HOSTS = [f"ip-10-0-{i}-{j}" for i in (1, 2, 3) for j in (11, 12, 13)]
LEVELS = ["INFO"] * 80 + ["WARN"] * 12 + ["ERROR"] * 7 + ["DEBUG"] * 1

INFO_MSG = [
    "request completed successfully",
    "user session refreshed",
    "cache hit for product catalog",
    "healthcheck ok",
    "order accepted and queued",
]
WARN_MSG = [
    "slow query detected on orders table",
    "retrying upstream call after transient failure",
    "connection pool utilization above 80 percent",
    "response latency above threshold",
]
ERROR_MSG = [
    "connection timeout to payment gateway",
    "java.lang.OutOfMemoryError: Java heap space",
    "failed to acquire database connection",
    "upstream service returned 503 Service Unavailable",
    "deadline exceeded while calling inventory service",
]

def main(count: int, index: str):
    now = datetime(2026, 7, 3, 12, 0, 0, tzinfo=timezone.utc)
    lines = []
    for i in range(count):
        level = random.choice(LEVELS)
        if level == "ERROR":
            msg = random.choice(ERROR_MSG)
            status = random.choice([500, 502, 503, 504])
            latency = random.randint(800, 30000)
        elif level == "WARN":
            msg = random.choice(WARN_MSG)
            status = random.choice([200, 200, 429])
            latency = random.randint(300, 5000)
        else:
            msg = random.choice(INFO_MSG)
            status = 200
            latency = random.randint(5, 300)
        ts = now - timedelta(seconds=random.randint(0, 6 * 3600))
        doc = {
            "@timestamp": ts.strftime("%Y-%m-%dT%H:%M:%S.000Z"),
            "level": level,
            "service": random.choice(SERVICES),
            "host": random.choice(HOSTS),
            "status_code": status,
            "latency_ms": latency,
            "message": msg,
        }
        lines.append(json.dumps({"index": {"_index": index}}))
        lines.append(json.dumps(doc, ensure_ascii=False))
    sys.stdout.write("\n".join(lines) + "\n")

if __name__ == "__main__":
    count = int(sys.argv[1]) if len(sys.argv) > 1 else 5000
    index = sys.argv[2] if len(sys.argv) > 2 else "app-logs"
    main(count, index)