# 真实磁盘用量
GET _cat/allocation?v&h=disk.percent,disk.used,disk.avail,disk.total,node

# 段内存、 fielddata、查询缓存
GET _stats?level=indices&metric=store,segments,fielddata,query_cache,request_cache

# 节点级别占用
GET _nodes/stats/fs?filter_path=nodes.*.fs.total

# 索引级别占用
GET _cat/indices?v&s=store.size:desc&h=index,health,status,pri,rep,docs.count,store.size


先看索引总大小、文档数、删除文档数：
GET _cat/indices/test_api_review-202603?v&bytes=mb
看每个主分片/副本分片占用：
GET _cat/shards/test_api_review-202603?v&bytes=mb
看更详细的 store/indexing/search/segments 统计：
GET test_api_review-202603/_stats?human=true
如果只关心磁盘存储：
GET test_api_review-202603/_stats/store?human=true
看 Lucene segment 占用，能看到删除文档、段数量这类信息：
GET test_api_review-202603/_segments
如果你想直接算“平均每条文档占用”，建议查这个，返回 JSON 好取值：
GET test_api_review-202603/_stats/docs,store?human=true

#  update_by_query
POST /my_index/_update_by_query
{
  "query": {
    "match": {
      "user_id": "U10086"
    }
  },
  "script": {
    "source": "ctx._source.level = params.new_level; ctx._source.vip_expire = params.expire_date",
    "lang": "painless",
    "params": {
      "new_level": 5,
      "expire_date": "2027-08-03"
    }
  }
}

POST prod_xxxxxx/_update/txxxxxxxxx
{
  "script": {
    "lang": "painless",
    "source": """
      if (ctx._source.callback_tags == null || ctx._source.callback_tags.size() == 0) {
        ctx._source.callback_tags = [params.tag];
      } else {
        ctx._source.callback_tags[0].first = params.tag.first;
        ctx._source.callback_tags[0].second = params.tag.second;
        ctx._source.callback_tags[0].third = params.tag.third;
      }
    """,
    "params": {
      "tag": {
        "first": "生态一级",
        "second": "游戏黑产",
        "third": "破解或外挂或私服"
      }
    }
  }
}

{
  "aggs": {
    "date_buckets": {
      "date_histogram": {
        "field": "created_at",
        "time_zone": "+08:00",
        "fixed_interval": "1d",
        "format": "yyyy-MM-dd",
        "min_doc_count": 0,
        "extended_bounds": {
          "min": 1700000000000,
          "max": 1700000000000
        }
      },
      "aggs": {
        "risk_count": {
          "filter": {
            "term": {
              "show_result": "risk"
            }
          }
        },
        "no_risk_count": {
          "filter": {
            "term": {
              "show_result": "no_risk"
            }
          }
        },
        "review_count": {
          "filter": {
            "term": {
              "show_result": "risk"
            }
          }
        }
      }
    }
  }
}

#!/usr/bin/env python3
"""
基于 search_after 的 Elasticsearch 全量数据同步脚本 (优化版)

优化点：
1. 使用 search_after 替代 scroll，无状态，极低内存占用。
2. 增加每批次间休眠，防止源 ES CPU/IO 过载。
3. 支持别名同步。
4. 增强重试机制和错误处理。
"""

import argparse
import sys
import time
from typing import List, Optional

from elasticsearch import Elasticsearch
from elasticsearch.exceptions import ConnectionError as ESConnectionError, TransportError
from elasticsearch.helpers import parallel_bulk
from tqdm import tqdm


def sync_aliases(src_client: Elasticsearch, dst_client: Elasticsearch) -> int:
    """同步源集群的别名到目标集群"""
    try:
        all_aliases = src_client.indices.get_alias()
    except Exception as e:
        print(f"[跳过] 获取源端别名失败: {e}")
        return 0

    alias_map = {}
    for index_name, body in all_aliases.items():
        for alias_name, alias_cfg in body.get("aliases", {}).items():
            alias_map.setdefault(alias_name, []).append((index_name, alias_cfg))

    if not alias_map:
        print("[信息] 源端没有发现别名")
        return 0

    # 获取目标端已存在的索引，避免关联不存在的索引报错
    try:
        existing_indices = set(dst_client.indices.get(index="*").keys())
    except Exception:
        existing_indices = set()

    actions = []
    # 先移除所有旧别名关联（谨慎操作，确保业务允许）
    # 注意：生产环境建议更精细地管理别名，这里为了简单直接全量刷新
    for alias_name in alias_map.keys():
        actions.append({"remove": {"index": "*", "alias": alias_name}})

    for alias_name, entries in alias_map.items():
        for index_name, alias_cfg in entries:
            if index_name not in existing_indices:
                print(f"[跳过] 别名 {alias_name} -> {index_name}，目标端该索引不存在")
                continue
            
            add_action = {"index": index_name, "alias": alias_name}
            # 复制别名的配置属性
            for key in ("is_write_index", "filter", "routing", "index_routing", "search_routing"):
                if key in alias_cfg:
                    add_action[key] = alias_cfg[key]
            actions.append({"add": add_action})

    if not actions:
        print("[信息] 没有需要同步的别名动作")
        return 0

    try:
        dst_client.indices.update_aliases(body={"actions": actions})
        print(f"[信息] 已同步 {len(alias_map)} 个别名定义")
        return len(alias_map)
    except Exception as e:
        print(f"[错误] 同步别名失败: {e}")
        return 0


def get_client(hosts: str, user: str, password: str) -> Elasticsearch:
    parsed_hosts = [h.strip() for h in hosts.split(",") if h.strip()]
    if not parsed_hosts:
        raise ValueError("host 列表不能为空")
    return Elasticsearch(
        hosts=parsed_hosts,
        basic_auth=(user, password),
        request_timeout=60,
        max_retries=3,
        retry_on_timeout=True,
    )


def list_indices(client: Elasticsearch, pattern: str) -> List[str]:
    try:
        info = client.indices.get(index=pattern, expand_wildcards="open")
    except Exception as e:
        print(f"[跳过] 模式 {pattern} 获取索引失败: {e}")
        return []
    return sorted(info.keys())


def ensure_target_index(client: Elasticsearch, index_name: str) -> None:
    try:
        if not client.indices.exists(index=index_name):
            client.indices.create(index=index_name, body={})
    except Exception as e:
        print(f"[警告] 确保目标索引存在时出错: {e}")


def build_upsert_actions(hits, dst_index: str):
    for hit in hits:
        action = {"_index": dst_index, "_id": hit.get("_id")}
        if "_routing" in hit:
            action["_routing"] = hit["_routing"]
        if "_parent" in hit:
            action["_parent"] = hit["_parent"]
        yield {**action, "_source": hit.get("_source", {})}


def request_with_retry(func, label: str, max_retries: int = 3, retry_wait_sec: int = 5):
    last_error = None
    for attempt in range(1, max_retries + 1):
        try:
            return func()
        except (TransportError, ESConnectionError, ConnectionError, TimeoutError, OSError) as exc:
            last_error = exc
            if attempt >= max_retries:
                raise
            print(f"[重试] {label} 第 {attempt}/{max_retries} 次失败: {exc}，{retry_wait_sec}s 后重试")
            time.sleep(retry_wait_sec)
        except Exception as exc:
            last_error = exc
            if attempt >= max_retries:
                raise
            print(f"[重试] {label} 第 {attempt}/{max_retries} 次失败: {exc}，{retry_wait_sec}s 后重试")
            time.sleep(retry_wait_sec)
    if last_error is not None:
        raise last_error
    raise RuntimeError(f"{label} 未返回结果")


def bulk_write_with_retry(
    dst_client: Elasticsearch,
    hits,
    dst_index: str,
    batch_size: int,
    concurrency: int,
    max_retries: int = 3,
    retry_wait_sec: int = 5,
):
    def run_bulk():
        results = []
        for success, _info in parallel_bulk(
            dst_client,
            build_upsert_actions(hits, dst_index),
            chunk_size=batch_size,
            thread_count=max(1, concurrency),
            queue_size=max(1, concurrency * 2),
            refresh=False,
            raise_on_error=False,
        ):
            results.append(success)
        return results

    return request_with_retry(
        run_bulk,
        f"{dst_index} bulk write",
        max_retries=max_retries,
        retry_wait_sec=retry_wait_sec,
    )


def copy_index_search_after(
    src_client: Elasticsearch,
    dst_client: Elasticsearch,
    src_index: str,
    dst_index: str | None = None,
    batch_size: int = 2000,
    refresh: bool = True,
    concurrency: int = 2,
    max_retries: int = 3,
    retry_wait_sec: int = 5,
    sleep_interval: float = 0.1,
) -> int:
    """使用 search_after 进行全量同步"""
    dst_index = dst_index or src_index
    ensure_target_index(dst_client, dst_index)

    total = 0
    page_num = 0
    search_after_value = None
    
    # 使用 _shard_doc 作为第二排序字段通常比 _id 性能更好，因为它是 Lucene 内部顺序
    # 如果 ES 版本较老不支持 _shard_doc，可以回退到 _id
    sort_field_2 = "_shard_doc" 
    
    query_body = {
        "query": {"match_all": {}},
        "sort": [
            {"_shard_doc": "asc"}, 
            {"_id": "asc"}
        ],
        "size": batch_size
    }

    # 检查是否支持 _shard_doc，如果不支持则报错或回退（简单起见，这里假设现代 ES）
    # 注意：_shard_doc 是 ES 7.x+ 推荐的高效排序方式

    try:
        while True:
            page_num += 1
            
            if search_after_value is None:
                resp = request_with_retry(
                    lambda: src_client.search(index=src_index, body=query_body),
                    f"{src_index} search init",
                    max_retries=max_retries,
                    retry_wait_sec=retry_wait_sec,
                )
            else:
                query_body["search_after"] = search_after_value
                resp = request_with_retry(
                    lambda: src_client.search(index=src_index, body=query_body),
                    f"{src_index} search next",
                    max_retries=max_retries,
                    retry_wait_sec=retry_wait_sec,
                )

            hits = resp.get("hits", {}).get("hits", [])
            if not hits:
                break

            # 提取最后一行的排序值
            last_hit = hits[-1]
            sort_values = last_hit.get("sort")
            if sort_values:
                search_after_value = sort_values
            else:
                break
            
            # 写入目标集群
            if hits:
                results = bulk_write_with_retry(
                    dst_client,
                    hits,
                    dst_index,
                    batch_size=batch_size,
                    concurrency=concurrency,
                    max_retries=max_retries,
                    retry_wait_sec=retry_wait_sec,
                )
                error_count = sum(1 for success in results if not success)
                if error_count:
                    print(f"[警告] {src_index} -> {dst_index} 写入完成，但有 {error_count} 条失败")

            total += len(hits)
            
            if page_num % 50 == 0:
                print(f"[进度] {src_index} 第 {page_num} 批完成，累计已复制 {total} 条")
            
            # 关键优化：休眠片刻，给源 ES 释放资源的机会
            time.sleep(sleep_interval)

    except Exception as e:
        print(f"[错误] 同步索引 {src_index} 过程中发生异常: {e}")
        raise

    if refresh:
        try:
            dst_client.indices.refresh(index=dst_index)
        except Exception:
            pass

    return total


def main() -> int:
    parser = argparse.ArgumentParser(description="通过 search_after 全量同步 Elasticsearch 数据")
    parser.add_argument("--src-host", default="http://src-es:9200")
    parser.add_argument("--src-user", default="elastic")
    parser.add_argument("--src-pass", default="src_password")
    parser.add_argument("--dst-host", default="http://127.0.0.1:9200")
    parser.add_argument("--dst-user", default="elastic")
    parser.add_argument("--dst-pass", default="dst2_password")
    parser.add_argument("--patterns", default="logs-*,metrics-*,audit-*")
    parser.add_argument("--batch-size", type=int, default=2000, help="每批读取文档数，默认 2000")
    parser.add_argument("--concurrency", type=int, default=2, help="并发 bulk 写入线程数，默认 2")
    parser.add_argument("--max-retries", type=int, default=3, help="连接/超时重试次数，默认 3")
    parser.add_argument("--retry-wait-sec", type=int, default=5, help="每次重试前等待秒数，默认 5")
    parser.add_argument("--sync-aliases", action="store_true", help="同步别名（默认关闭）")
    parser.add_argument("--sleep-interval", type=float, default=0.1, help="每批次间休眠秒数，保护源ES")
    args = parser.parse_args()

    src_client = get_client(args.src_host, args.src_user, args.src_pass)
    dst_client = get_client(args.dst_host, args.dst_user, args.dst_pass)

    patterns = [p.strip() for p in args.patterns.split(",") if p.strip()]
    total_indexes = 0
    total_docs = 0

    for pattern in patterns:
        indices = list_indices(src_client, pattern)
        if not indices:
            print(f"[信息] 模式 {pattern} 没有匹配到任何索引")
            continue

        print(f"==== 模式 {pattern} ====")
        total_pattern_indexes = len(indices)
        for index_idx, index_name in enumerate(
            tqdm(indices, desc=f"同步模式 {pattern}", total=total_pattern_indexes, unit="index"),
            start=1,
        ):
            total_indexes += 1
            tqdm.write(f"[{index_idx}/{total_pattern_indexes}] 同步索引: {index_name}")
            start = time.time()
            try:
                copied = copy_index_search_after(
                    src_client=src_client,
                    dst_client=dst_client,
                    src_index=index_name,
                    dst_index=index_name,
                    batch_size=args.batch_size,
                    concurrency=args.concurrency,
                    max_retries=args.max_retries,
                    retry_wait_sec=args.retry_wait_sec,
                    sleep_interval=args.sleep_interval,
                )
                elapsed = time.time() - start
                total_docs += copied
                tqdm.write(f"  复制完成: {copied} 条文档, 用时 {elapsed:.1f} 秒")
            except Exception as e:
                tqdm.write(f"  [错误] 索引 {index_name} 同步失败: {e}")
                # 可以选择继续或退出，这里选择继续处理下一个索引

    print("==== 统计摘要 ====")
    print(f"- 目标模式数: {len(patterns)}")
    print(f"- 实际处理索引数: {total_indexes}")
    print(f"- 实际复制文档数: {total_docs}")
    
    if args.sync_aliases:
        print("\n---- 开始同步别名 ----")
        alias_count = sync_aliases(src_client, dst_client)
        print(f"- 同步别名数: {alias_count}")
    else:
        print("- 别名同步: 未启用（使用 --sync-aliases 可开启）")
        
    print("==== 完成 ====")
    return 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""
基于双字段排序 (Time + Unique ID) 的 Elasticsearch 增量补数脚本。

核心优化：
1. 使用 search_after 替代 scroll，无状态，低内存消耗。
2. 支持指定第二个排序字段 (如 trace_id)，确保同一时间点数据的顺序确定性，防止漏数。
3. 增加限流休眠，保护源集群。
"""

import argparse
import json
import os
import sys
import time
from typing import List, Optional
from datetime import datetime, timezone

from elasticsearch import Elasticsearch
from elasticsearch.helpers import parallel_bulk
from tqdm import tqdm


def get_client(hosts: str, user: str, password: str) -> Elasticsearch:
    parsed_hosts = [h.strip() for h in hosts.split(",") if h.strip()]
    if not parsed_hosts:
        raise ValueError("host 列表不能为空")
    return Elasticsearch(
        hosts=parsed_hosts,
        basic_auth=(user, password),
        request_timeout=60,
        max_retries=3,
        retry_on_timeout=True,
    )


def list_indices(client: Elasticsearch, pattern: str) -> List[str]:
    try:
        info = client.indices.get(index=pattern, expand_wildcards="open")
    except Exception as e:
        print(f"[跳过] 模式 {pattern} 获取索引失败: {e}")
        return []
    return sorted(info.keys())


def ensure_target_index(client: Elasticsearch, index_name: str) -> None:
    try:
        if not client.indices.exists(index=index_name):
            client.indices.create(index=index_name, body={})
    except Exception as e:
        print(f"[警告] 确保目标索引存在时出错: {e}")


def build_upsert_actions(hits, dst_index: str):
    for hit in hits:
        action = {"_index": dst_index, "_id": hit.get("_id")}
        if "_routing" in hit:
            action["_routing"] = hit["_routing"]
        if "_parent" in hit:
            action["_parent"] = hit["_parent"]
        yield {**action, "_source": hit.get("_source", {})}


def load_last_sync_time(state_file: Optional[str]) -> Optional[int]:
    """加载上次同步时间，返回毫秒时间戳"""
    if not state_file or not os.path.exists(state_file):
        return None
    try:
        with open(state_file, "r", encoding="utf-8") as fh:
            data = json.load(fh)
        last_sync = data.get("last_sync")
        if last_sync is not None:
            if isinstance(last_sync, str):
                return parse_time_to_millis(last_sync)
            return int(last_sync)
    except Exception as exc:
        print(f"[警告] 读取状态文件失败: {state_file} ({exc})")
    return None


def save_last_sync_time(state_file: Optional[str], value: int) -> None:
    if not state_file:
        return
    try:
        with open(state_file, "w", encoding="utf-8") as fh:
            json.dump({"last_sync": value}, fh, ensure_ascii=False, indent=2)
    except Exception as exc:
        print(f"[警告] 写入状态文件失败: {state_file} ({exc})")


def parse_time_to_millis(value: Optional[str]) -> Optional[int]:
    if value is None:
        return None
    value = str(value).strip()
    if value.isdigit():
        ts = int(value)
        return ts * 1000 if ts < 10**12 else ts
    try:
        if value.endswith('Z'):
            dt = datetime.fromisoformat(value.replace('Z', '+00:00'))
        else:
            dt = datetime.fromisoformat(value)
        if dt.tzinfo is None:
            dt = dt.replace(tzinfo=timezone.utc)
        return int(dt.timestamp() * 1000)
    except Exception:
        raise ValueError(f"无法解析时间格式: {value}")


def sync_index_incremental_dual_sort(
    src_client: Elasticsearch,
    dst_client: Elasticsearch,
    src_index: str,
    dst_index: str | None = None,
    time_field: str = "@timestamp",
    sort_field_2: str = "_id",  # 第二个排序字段，用于去重和确定顺序
    since_millis: Optional[int] = None,
    until_millis: Optional[int] = None,
    batch_size: int = 2000,
    refresh: bool = True,
    concurrency: int = 2,
    sleep_interval: float = 0.1,
) -> int:
    if since_millis is None:
        raise ValueError("增量同步必须提供 since 时间 (毫秒时间戳)")

    dst_index = dst_index or src_index
    ensure_target_index(dst_client, dst_index)

    total = 0
    page_num = 0
    search_after_value = None
    
    # 构建基础查询
    # 注意：sort 必须包含两个字段，且顺序一致
    query_body = {
        "query": {"range": {time_field: {"gte": since_millis}}},
        "sort": [
            {time_field: "asc"},
            {sort_field_2: "asc"} 
        ],
        "size": batch_size
    }
    
    if until_millis is not None:
        query_body["query"]["range"][time_field]["lt"] = until_millis

    try:
        while True:
            page_num += 1
            
            # 如果是第一页，直接搜索；后续页使用 search_after
            if search_after_value is None:
                resp = src_client.search(index=src_index, body=query_body)
            else:
                query_body["search_after"] = search_after_value
                resp = src_client.search(index=src_index, body=query_body)

            hits = resp.get("hits", {}).get("hits", [])
            if not hits:
                break

            # 提取最后一行的排序值 [time_val, id_val]
            last_hit = hits[-1]
            sort_values = last_hit.get("sort")
            
            if sort_values and len(sort_values) >= 2:
                # 确保存储的是正确的类型，ES 返回的 sort 值可能已经是处理过的
                search_after_value = sort_values
            else:
                # 极端情况下如果没有 sort 值（极少见），中断循环防止死循环
                print(f"[警告] {src_index} 第 {page_num} 批未返回 sort 值，停止同步")
                break
            
            # 写入目标集群
            for success, _info in parallel_bulk(
                dst_client,
                build_upsert_actions(hits, dst_index),
                chunk_size=batch_size,
                thread_count=max(1, concurrency),
                queue_size=max(1, concurrency * 2),
                refresh=False,
                raise_on_error=False,
            ):
                if not success:
                    print(f"[警告] {src_index} 有一批文档写入失败")

            total += len(hits)
            if page_num % 50 == 0:  # 每50页打印一次进度
                print(f"[进度] {src_index} 第 {page_num} 批完成，累计已补数 {total} 条")
            
            # 关键优化：休眠片刻，给源 ES 释放资源的机会
            time.sleep(sleep_interval)

    except Exception as e:
        print(f"[错误] 同步过程中发生异常: {e}")
        raise

    if refresh:
        try:
            dst_client.indices.refresh(index=dst_index)
        except Exception:
            pass

    return total


def main() -> int:
    parser = argparse.ArgumentParser(description="通过双字段排序做 Elasticsearch 增量补数")
    parser.add_argument("--src-host", default="http://src-es:9200")
    parser.add_argument("--src-user", default="elastic")
    parser.add_argument("--src-pass", default="src_password")
    parser.add_argument("--dst-host", default="http://127.0.0.1:9200")
    parser.add_argument("--dst-user", default="elastic")
    parser.add_argument("--dst-pass", default="dst2_password")
    parser.add_argument("--patterns", default="logs-*,metrics-*,audit-*")
    parser.add_argument("--time-field", default="@timestamp", help="主排序时间字段")
    parser.add_argument("--sort-field-2", default="_id", help="第二排序字段 (用于去重/确定顺序)，例如 trace_id 或 _id")
    parser.add_argument("--since", default=None, help="增量起点时间")
    parser.add_argument("--until", default=None, help="可选：增量终点时间")
    parser.add_argument("--state-file", default=None, help="可选：保存/读取最后一次同步时间")
    parser.add_argument("--batch-size", type=int, default=2000, help="每批读取文档数")
    parser.add_argument("--concurrency", type=int, default=2, help="并发写入线程数")
    parser.add_argument("--sleep-interval", type=float, default=0.1, help="每批次间休眠秒数")
    args = parser.parse_args()

    src_client = get_client(args.src_host, args.src_user, args.src_pass)
    dst_client = get_client(args.dst_host, args.dst_user, args.dst_pass)

    # 确定起始时间
    effective_since_str = args.since or load_last_sync_time(args.state_file)
    if isinstance(effective_since_str, int):
        effective_since_millis = effective_since_str
    elif isinstance(effective_since_str, str):
        effective_since_millis = parse_time_to_millis(effective_since_str)
    else:
        effective_since_millis = None

    if not effective_since_millis:
        print("[错误] 请提供 --since，或者提供 --state-file 并且其中已有 last_sync")
        return 2

    # 确定结束时间
    if args.until:
        effective_until_millis = parse_time_to_millis(args.until)
    else:
        effective_until_millis = int(time.time() * 1000)

    patterns = [p.strip() for p in args.patterns.split(",") if p.strip()]
    total_indexes = 0
    total_docs = 0

    for pattern in patterns:
        indices = list_indices(src_client, pattern)
        if not indices:
            print(f"[信息] 模式 {pattern} 没有匹配到任何索引")
            continue

        print(f"==== 模式 {pattern} ====")
        for index_idx, index_name in enumerate(
            tqdm(indices, desc=f"增量补数 {pattern}", total=len(indices), unit="index"),
            start=1,
        ):
            total_indexes += 1
            tqdm.write(f"[{index_idx}/{len(indices)}] 增量补数索引: {index_name}")
            try:
                copied = sync_index_incremental_dual_sort(
                    src_client=src_client,
                    dst_client=dst_client,
                    src_index=index_name,
                    dst_index=index_name,
                    time_field=args.time_field,
                    sort_field_2=args.sort_field_2,  # 传入第二个排序字段
                    since_millis=effective_since_millis,
                    until_millis=effective_until_millis,
                    batch_size=args.batch_size,
                    concurrency=args.concurrency,
                    sleep_interval=args.sleep_interval,
                )
                total_docs += copied
                tqdm.write(f"  补数完成: {copied} 条文档")
            except Exception as e:
                tqdm.write(f"  [错误] 索引 {index_name} 同步失败: {e}")

    if args.state_file:
        save_last_sync_time(args.state_file, effective_until_millis)
        print(f"[信息] 已将本次增量同步结束时间写入状态文件: {args.state_file} -> {effective_until_millis}")

    print("==== 统计摘要 ====")
    print(f"- 实际处理索引数: {total_indexes}")
    print(f"- 实际补数文档数: {total_docs}")
    print("==== 完成 ====")
    return 0


if __name__ == "__main__":
    sys.exit(main())