flowchart TB
    subgraph "REST API 层"
        RestClusterHealthAction[RestClusterHealthAction<br/>REST健康接口]
        RestClusterStateAction[RestClusterStateAction<br/>REST状态接口]
        RestClusterSettingsAction[RestClusterSettingsAction<br/>REST设置接口]
    end

    subgraph "Transport Action 层"
        TransportClusterHealthAction[TransportClusterHealthAction<br/>健康查询Action]
        TransportClusterStateAction[TransportClusterStateAction<br/>状态查询Action]
        TransportCreateIndexAction[TransportCreateIndexAction<br/>索引创建Action]
        TransportClusterRerouteAction[TransportClusterRerouteAction<br/>重路由Action]
    end

    subgraph "集群服务层 (Cluster Service Layer)"
        ClusterService[ClusterService<br/>集群服务]
        MasterService[MasterService<br/>主节点服务]
        ClusterApplierService[ClusterApplierService<br/>状态应用服务]
    end

    subgraph "协调层 (Coordination Layer)"
        Coordinator[Coordinator<br/>集群协调器]
        PeerFinder[PeerFinder<br/>节点发现]
        JoinHelper[JoinHelper<br/>加入辅助器]
        PublicationTransportHandler[PublicationTransportHandler<br/>发布传输处理器]
        LeaderChecker[LeaderChecker<br/>主节点检查器]
        FollowersChecker[FollowersChecker<br/>跟随者检查器]
    end

    subgraph "状态管理层 (State Management)"
        ClusterState[ClusterState<br/>集群状态]
        CoordinationState[CoordinationState<br/>协调状态]
        ClusterStatePublisher[ClusterStatePublisher<br/>状态发布器]
        ClusterStateTaskQueue[ClusterStateTaskQueue<br/>状态更新任务队列]
    end

    subgraph "分片分配层 (Allocation Layer)"
        AllocationService[AllocationService<br/>分配服务]
        DesiredBalanceShardsAllocator[DesiredBalanceShardsAllocator<br/>期望平衡分配器]
        AllocationDeciders[AllocationDeciders<br/>分配决策器集]
        DiskThresholdDecider[DiskThresholdDecider<br/>磁盘阈值决策器]
        FilterAllocationDecider[FilterAllocationDecider<br/>过滤分配决策器]
    end

    subgraph "传输层 (Transport Layer)"
        TransportService[TransportService<br/>传输服务]
    end

    %% REST API 到 Transport Action
    RestClusterHealthAction -->|解析请求| TransportClusterHealthAction
    RestClusterStateAction -->|解析请求| TransportClusterStateAction
    RestClusterSettingsAction -->|解析请求| TransportCreateIndexAction

    %% Transport Action 到 Cluster Service
    TransportClusterHealthAction -->|读取状态| ClusterService
    TransportClusterStateAction -->|读取状态| ClusterService
    TransportCreateIndexAction -->|提交任务| MasterService
    TransportClusterRerouteAction -->|提交任务| MasterService

    %% Cluster Service 内部
    ClusterService -->|状态更新| MasterService
    ClusterService -->|状态应用| ClusterApplierService
    ClusterService -->|获取状态| ClusterState

    %% Master Service 到 Coordinator
    MasterService -->|任务队列| ClusterStateTaskQueue
    MasterService -->|发布状态| Coordinator
    MasterService -->|分片分配| AllocationService

    %% Coordinator 内部协调
    Coordinator -->|状态发布| ClusterStatePublisher
    Coordinator -->|节点发现| PeerFinder
    Coordinator -->|节点加入| JoinHelper
    Coordinator -->|维护协调状态| CoordinationState
    Coordinator -->|主节点检查| LeaderChecker
    Coordinator -->|跟随者检查| FollowersChecker

    %% ClusterStatePublisher 到传输层
    ClusterStatePublisher -->|序列化发送| PublicationTransportHandler
    PublicationTransportHandler -->|网络传输| TransportService

    %% AllocationService 内部
    AllocationService -->|调用分配器| DesiredBalanceShardsAllocator
    AllocationService -->|应用决策器| AllocationDeciders
    AllocationDeciders -->|磁盘检查| DiskThresholdDecider
    AllocationDeciders -->|过滤检查| FilterAllocationDecider

    %% 状态同步
    ClusterStatePublisher -.->|发布完成| ClusterApplierService
    ClusterApplierService -.->|更新本地状态| ClusterState

    style RestClusterHealthAction fill:#e1f5ff
    style TransportClusterHealthAction fill:#bbdefb
    style ClusterService fill:#90caf9
    style Coordinator fill:#64b5f6
    style MasterService fill:#42a5f5
    style AllocationService fill:#2196f3

flowchart LR
    subgraph "API接口层"
        HealthAPI[Cluster Health API]
        StateAPI[Cluster State API]
        SettingsAPI[Cluster Settings API]
        RerouteAPI[Cluster Reroute API]
    end

    subgraph "协调服务"
        Coordinator[Coordinator]
        JoinHelper[JoinHelper]
        PublicationHandler[PublicationTransportHandler]
    end

    subgraph "主节点服务"
        MasterService[MasterService]
        TaskQueue[TaskQueue]
    end

    subgraph "状态管理"
        ClusterState[ClusterState]
        CoordinationState[CoordinationState]
    end

    subgraph "分配服务"
        AllocationService[AllocationService]
        ShardsAllocator[ShardsAllocator]
        Deciders[AllocationDeciders]
    end

    subgraph "应用服务"
        ClusterApplier[ClusterApplierService]
        Listeners[ClusterStateListener]
    end

    %% 调用关系
    HealthAPI -->|查询| ClusterState
    StateAPI -->|查询| ClusterState
    SettingsAPI -->|提交任务| TaskQueue
    RerouteAPI -->|提交任务| TaskQueue

    TaskQueue -->|批量执行| MasterService
    MasterService -->|执行任务| AllocationService
    MasterService -->|发布状态| Coordinator

    Coordinator -->|处理加入| JoinHelper
    Coordinator -->|发布状态| PublicationHandler
    Coordinator -->|维护状态| CoordinationState

    AllocationService -->|调用分配器| ShardsAllocator
    AllocationService -->|应用决策| Deciders
    AllocationService -->|更新路由| ClusterState

    PublicationHandler -.->|2PC提交| ClusterApplier
    ClusterApplier -->|通知监听器| Listeners
    ClusterApplier -->|更新本地状态| ClusterState

public class Coordinator extends AbstractLifecycleComponent implements ClusterStatePublisher {
    // 模式: CANDIDATE（候选者）, LEADER（主节点）, FOLLOWER（跟随者）
    private Mode mode = Mode.CANDIDATE;

    // 当前任期
    private long currentTerm = 0;

    // 协调状态
    private final SetOnce<CoordinationState> coordinationState = new SetOnce<>();

    // 启动选举
    private void startElection() {
        synchronized (mutex) {
            if (mode == Mode.CANDIDATE) {
                final long electionTerm = getTermForNewElection();
                broadcastStartJoinRequest(getLocalNode(), electionTerm, getDiscoveredNodes());
            }
        }
    }

    // 成为候选者
    void becomeCandidate(String method) {
        mode = Mode.CANDIDATE;
        peerFinder.activate(coordinationState.get().getLastAcceptedState().nodes());
        preVoteCollector.update(getPreVoteResponse(), null);
    }

    // 成为主节点
    private void becomeLeader() {
        mode = Mode.LEADER;
        peerFinder.deactivate(getLocalNode());
        clusterFormationFailureHelper.stop();

        // 启动心跳和健康检查
        leaderHeartbeatService.start(coordinationState.get(), this::publishClusterState);
        followersChecker.updateFastResponseState(getCurrentTerm(), mode);
        lagDetector.startLagDetector(getCurrentTerm());
    }

    // 成为跟随者
    private void becomeFollower(DiscoveryNode leaderNode) {
        mode = Mode.FOLLOWER;
        peerFinder.deactivate(leaderNode);

        // 启动主节点健康检查
        leaderChecker.updateLeader(leaderNode);
    }
}

stateDiagram-v2
    [*] --> CANDIDATE: 启动

    CANDIDATE --> CANDIDATE: 选举超时<br/>重新发起选举
    CANDIDATE --> LEADER: 获得多数派投票
    CANDIDATE --> FOLLOWER: 发现更高任期的主节点

    LEADER --> CANDIDATE: 失去多数派支持
    LEADER --> FOLLOWER: 发现更高任期的主节点

    FOLLOWER --> CANDIDATE: 主节点心跳超时
    FOLLOWER --> FOLLOWER: 收到主节点心跳

    LEADER --> [*]: 节点停止
    FOLLOWER --> [*]: 节点停止
    CANDIDATE --> [*]: 节点停止

public class ClusterState implements Diffable<ClusterState> {
    // 版本号（严格递增）
    private final long version;

    // 集群 UUID
    private final String stateUUID;

    // 路由表（分片分配）
    private final RoutingTable routingTable;

    // 节点信息
    private final DiscoveryNodes nodes;

    // 元数据（索引、模板、设置等）
    private final Metadata metadata;

    // 集群范围的阻塞
    private final ClusterBlocks blocks;

    // 协调元数据（任期、投票配置等）
    private final CoordinationMetadata coordinationMetadata;
}

// 集群状态更新流程
public ClusterState execute(ClusterState currentState) {
    // 1. 基于当前状态创建 Builder
    ClusterState.Builder builder = ClusterState.builder(currentState);

    // 2. 应用变更
    builder.incrementVersion();
    builder.metadata(newMetadata);
    builder.routingTable(newRoutingTable);

    // 3. 构建新状态
    return builder.build();
}

public class MasterService extends AbstractLifecycleComponent {
    // 任务队列（按优先级）
    private final PrioritizedEsThreadPoolExecutor threadPoolExecutor;

    // 提交集群状态更新任务
    public <T> void submitStateUpdateTask(
        String source,
        T task,
        ClusterStateTaskConfig config,
        ClusterStateTaskExecutor<T> executor,
        ClusterStateTaskListener listener
    ) {
        // 1. 包装为任务
        TaskInputs taskInputs = new TaskInputs(task, source, listener);

        // 2. 加入队列
        threadPoolExecutor.execute(
            taskInputs,
            config.getThreadPool(),
            batchExecutionContext -> {
                // 3. 批量执行任务
                executeTasks(batchExecutionContext, executor);
            }
        );
    }

    // 批量执行任务
    private <T> void executeTasks(
        BatchExecutionContext<T> context,
        ClusterStateTaskExecutor<T> executor
    ) {
        ClusterState previousState = clusterService.state();

        // 1. 执行任务获取新状态
        ClusterTasksResult<T> result = executor.execute(previousState, context.tasks);
        ClusterState newState = result.resultingState;

        // 2. 发布新状态
        if (newState != previousState) {
            publishClusterState(newState, context);
        }

        // 3. 通知任务结果
        notifyListeners(context, result);
    }
}

flowchart TB
    Start[开始] --> UnassignedShards[获取未分配分片列表]
    UnassignedShards --> Primary{是主分片?}

    Primary -->|是| PrimaryAlloc[主分片分配]
    Primary -->|否| ReplicaAlloc[副本分片分配]

    PrimaryAlloc --> FindNodes[查找可用节点]
    ReplicaAlloc --> FindNodes

    FindNodes --> Deciders[应用分配决策器]

    Deciders --> CanAllocate{可以分配?}
    CanAllocate -->|是| Allocate[分配分片]
    CanAllocate -->|否| NextShard[下一个分片]

    Allocate --> Balance[重平衡]
    Balance --> End[结束]
    NextShard --> End

// 分配决策器接口
public interface AllocationDecider {
    Decision canAllocate(ShardRouting shardRouting, RoutingNode node, RoutingAllocation allocation);
    Decision canRemain(ShardRouting shardRouting, RoutingNode node, RoutingAllocation allocation);
    Decision canRebalance(ShardRouting shardRouting, RoutingAllocation allocation);
}

// 内置决策器示例
public class DiskThresholdDecider extends AllocationDecider {
    @Override
    public Decision canAllocate(ShardRouting shardRouting, RoutingNode node, RoutingAllocation allocation) {
        // 检查磁盘使用率
        if (diskUsagePercentage > highWatermark) {
            return Decision.NO;  // 磁盘不足,不允许分配
        }
        return Decision.YES;
    }
}

classDiagram
    class ClusterState {
        -long version
        -String stateUUID
        -ClusterName clusterName
        -DiscoveryNodes nodes
        -Metadata metadata
        -RoutingTable routingTable
        -ClusterBlocks blocks
        -Map~String,Custom~ customs
        -Map~String,CompatibilityVersions~ compatibilityVersions
        +builder(ClusterName)
        +diff(ClusterState)
        +incrementVersion()
        +getRoutingNodes()
    }

    class Metadata {
        -String clusterUUID
        -long version
        -Settings transientSettings
        -Settings persistentSettings
        -Map~String,IndexMetadata~ indices
        -Map~String,IndexTemplateMetadata~ templates
        -CoordinationMetadata coordinationMetadata
        +getTotalNumberOfShards()
        +builder()
    }

    class RoutingTable {
        -Map~String,IndexRoutingTable~ indicesRouting
        +index(String)
        +shardRoutingTable(ShardId)
        +hasIndex(String)
        +builder()
    }

    class DiscoveryNodes {
        -Map~String,DiscoveryNode~ nodes
        -String masterNodeId
        -String localNodeId
        +getMasterNode()
        +getLocalNode()
        +getSize()
        +getDataNodes()
    }

    class ClusterBlocks {
        -Set~ClusterBlock~ global
        -Map~String,Set~ClusterBlock~~ indices
        +hasGlobalBlock(ClusterBlockLevel)
        +hasIndexBlock(String, ClusterBlock)
    }

    class CoordinationMetadata {
        -long term
        -VotingConfiguration lastCommittedConfiguration
        -VotingConfiguration lastAcceptedConfiguration
        -Set~VotingConfigExclusion~ votingConfigExclusions
        +getTerm()
        +getLastCommittedConfiguration()
    }

    ClusterState --> Metadata
    ClusterState --> RoutingTable
    ClusterState --> DiscoveryNodes
    ClusterState --> ClusterBlocks
    Metadata --> CoordinationMetadata

classDiagram
    class Coordinator {
        -Mode mode
        -long currentTerm
        -Optional~DiscoveryNode~ lastKnownLeader
        -CoordinationState coordinationState
        -PeerFinder peerFinder
        -JoinHelper joinHelper
        -MasterService masterService
        +startElection()
        +becomeCandidate()
        +becomeLeader()
        +becomeFollower()
    }

    class CoordinationState {
        -long currentTerm
        -long lastAcceptedTerm
        -long lastAcceptedVersion
        -ClusterState lastAcceptedState
        -VotingConfiguration lastCommittedConfiguration
        -VotingConfiguration lastAcceptedConfiguration
        -Map~DiscoveryNode,Join~ joinVotes
        +handleJoin(Join)
        +handlePublishRequest(PublishRequest)
        +handleCommit(Commit)
    }

    class Mode {
        <<enumeration>>
        CANDIDATE
        LEADER
        FOLLOWER
    }

    class Join {
        -DiscoveryNode votingNode
        -DiscoveryNode masterNode
        -long term
        -long lastAcceptedTerm
        -long lastAcceptedVersion
    }

    class PublishRequest {
        -ClusterState clusterState
        +getAcceptedState()
    }

    class Commit {
        -long term
        -long version
    }

    Coordinator --> Mode
    Coordinator --> CoordinationState
    CoordinationState --> Join
    CoordinationState --> PublishRequest
    CoordinationState --> Commit

public class ClusterHealthRequest extends MasterNodeReadRequest<ClusterHealthRequest> {
    // 目标索引（可选，空表示整个集群）
    private String[] indices;

    // 索引选项
    private IndicesOptions indicesOptions = IndicesOptions.lenientExpandHidden();

    // 超时时间
    private TimeValue timeout = TimeValue.timeValueSeconds(30);

    // 等待集群达到指定状态
    private ClusterHealthStatus waitForStatus;

    // 等待无正在重新分配的分片
    private boolean waitForNoRelocatingShards = false;

    // 等待无正在初始化的分片
    private boolean waitForNoInitializingShards = false;

    // 等待指定数量的活跃分片
    private ActiveShardCount waitForActiveShards = ActiveShardCount.NONE;

    // 等待指定数量的节点
    private String waitForNodes = "";

    // 等待指定优先级的事件完成
    private Priority waitForEvents = null;
}

public class ClusterHealthResponse extends ActionResponse {
    // 集群名称
    private String clusterName;

    // 集群健康状态
    private ClusterHealthStatus status;

    // 是否超时
    private boolean timedOut;

    // 节点数量
    private int numberOfNodes;
    private int numberOfDataNodes;

    // 分片统计
    private int activePrimaryShards;
    private int activeShards;
    private int relocatingShards;
    private int initializingShards;
    private int unassignedShards;
    private int delayedUnassignedShards;

    // 待处理任务数量
    private int numberOfPendingTasks;
    private int numberOfInFlightFetch;

    // 活跃分片百分比
    private double activeShardsPercent;

    // 最长任务等待时间
    private TimeValue taskMaxWaitingTime;

    // 索引级别的健康状态（如果请求了特定索引）
    private Map<String, ClusterIndexHealth> indices;
}

# 1. 查询整个集群健康
GET /_cluster/health

# 2. 等待集群变为 green 状态（最多等待 30s）
GET /_cluster/health?wait_for_status=green&timeout=30s

# 3. 查询特定索引的健康
GET /_cluster/health/my_index

# 4. 等待所有分片活跃
GET /_cluster/health?wait_for_active_shards=all

# 5. 等待至少 3 个节点
GET /_cluster/health?wait_for_nodes=>=3

sequenceDiagram
    autonumber
    participant Client as HTTP Client
    participant RestHandler as RestClusterHealthAction
    participant NodeClient as NodeClient
    participant Transport as TransportClusterHealthAction
    participant ClusterService as ClusterService
    participant ClusterState as ClusterState<br/>(当前集群状态)
    participant Observer as ClusterStateObserver<br/>(可选,用于等待)

    Client->>RestHandler: GET /_cluster/health?wait_for_status=green&timeout=30s
    activate RestHandler

    Note over RestHandler: 1. 解析请求参数
    RestHandler->>RestHandler: fromRequest(request)<br/>- 解析 indices<br/>- 解析 timeout<br/>- 解析 waitForStatus<br/>- 解析 waitForNodes 等

    RestHandler->>RestHandler: 构建 ClusterHealthRequest

    Note over RestHandler: 2. 调用 NodeClient
    RestHandler->>NodeClient: client.admin().cluster().health(request)
    activate NodeClient

    NodeClient->>Transport: execute(ClusterHealthRequest)
    deactivate NodeClient
    activate Transport

    Note over Transport: 3. Transport Action 处理
    Transport->>Transport: masterOperation(task, request, state, listener)

    Transport->>ClusterService: clusterService.state()
    activate ClusterService
    ClusterService-->>Transport: 返回当前 ClusterState
    deactivate ClusterService

    Note over Transport: 4. 判断是否需要等待
    alt 需要等待 (wait_for_status 指定)
        Transport->>Transport: executeHealth(request, currentState, ...)
        Transport->>Transport: validateRequest(request, currentState, ...)

        alt 当前状态不满足条件
            Transport->>Observer: new ClusterStateObserver(...)<br/>waitForNextChange(...)
            activate Observer

            Note over Observer: 等待集群状态变更
            Observer->>Observer: 监听 ClusterStateListener

            loop 等待状态满足或超时
                Observer->>ClusterService: 监听状态变更事件
                ClusterService-->>Observer: onNewClusterState(newState)
                Observer->>Observer: acceptableClusterStatePredicate.test(newState)
            end

            Observer-->>Transport: 状态满足或超时
            deactivate Observer
        end
    end

    Note over Transport: 5. 计算集群健康
    Transport->>Transport: clusterHealth(request, clusterState, ...)<br/>- 解析索引名<br/>- 计算分片统计<br/>- 计算健康状态

    Transport->>ClusterState: 读取 nodes, routing, metadata
    activate ClusterState
    ClusterState-->>Transport: 节点信息、路由表、元数据
    deactivate ClusterState

    Transport->>Transport: 构建 ClusterHealthResponse<br/>- status (GREEN/YELLOW/RED)<br/>- numberOfNodes<br/>- activePrimaryShards<br/>- activeShards<br/>- unassignedShards

    Transport-->>RestHandler: ClusterHealthResponse
    deactivate Transport

    RestHandler->>RestHandler: RestToXContentListener.buildResponse()

    RestHandler-->>Client: 200 OK<br/>JSON Response
    deactivate RestHandler

sequenceDiagram
    autonumber
    participant N1 as Node 1<br/>(CANDIDATE)
    participant N2 as Node 2<br/>(FOLLOWER)
    participant N3 as Node 3<br/>(FOLLOWER)

    Note over N1,N3: 触发选举：心跳超时或节点启动

    N1->>N1: becomeCandidate()<br/>term = 1
    N1->>N1: ElectionScheduler 触发

    Note over N1,N3: Phase 1: Pre-Vote (预投票)

    N1->>N1: PreVoteCollector.start()

    par 并发发送预投票请求
        N1->>N2: PreVoteRequest<br/>(term=1, lastAcceptedTerm=0, lastAcceptedVersion=0)
        N2->>N2: 检查条件:<br/>1. term >= currentTerm<br/>2. lastAccepted >= own
        N2-->>N1: PreVoteResponse(granted=true)
    and
        N1->>N3: PreVoteRequest<br/>(term=1)
        N3->>N3: 检查条件
        N3-->>N1: PreVoteResponse(granted=true)
    end

    N1->>N1: 收集到 quorum (2/3)<br/>preVote 成功

    Note over N1,N3: Phase 2: Start Join (正式投票)

    N1->>N1: currentTerm++<br/>term = 2
    N1->>N1: JoinHelper.new CandidateJoinAccumulator()

    par 并发广播选举请求
        N1->>N2: StartJoinRequest<br/>(sourceNode=N1, term=2)
        N2->>N2: 检查 term (2 > 1)<br/>becomeFollower()
        N2->>N2: 构建 Join
        N2->>N1: Join(votingNode=N2, masterNode=N1, term=2)
        N1->>N1: joinAccumulator.addJoin(N2)
    and
        N1->>N3: StartJoinRequest<br/>(term=2)
        N3->>N3: becomeFollower()
        N3->>N3: 构建 Join
        N3->>N1: Join(votingNode=N3, term=2)
        N1->>N1: joinAccumulator.addJoin(N3)
    end

    N1->>N1: 检查 joinVotes.size() >= quorum<br/>2 votes >= 2 (quorum of 3)
    N1->>N1: becomeLeader()<br/>mode = LEADER

    Note over N1,N3: Phase 3: Publish Initial State

    N1->>N1: CoordinationState.handleStartJoin()<br/>更新 lastCommittedConfiguration

    N1->>N1: 构建初始集群状态:<br/>version++, masterNodeId=N1

    par 发布集群状态 (2-Phase Commit)
        N1->>N2: PublishRequest<br/>(ClusterState)
        N2->>N2: CoordinationState.handlePublishRequest()<br/>验证 term 和 version
        N2->>N2: 应用状态到本地
        N2-->>N1: PublishResponse(committed=false)
    and
        N1->>N3: PublishRequest<br/>(ClusterState)
        N3->>N3: handlePublishRequest()
        N3->>N3: 应用状态
        N3-->>N1: PublishResponse(committed=false)
    end

    N1->>N1: 等待 quorum ack (2/3)
    N1->>N1: 状态已提交

    par 发送 Commit 消息
        N1->>N2: Commit(term=2, version=N)
        N2->>N2: CoordinationState.handleCommit()<br/>标记状态为 committed
        N2->>N2: applyCommittedState()
    and
        N1->>N3: Commit(term=2, version=N)
        N3->>N3: handleCommit()
        N3->>N3: applyCommittedState()
    end

    Note over N1,N3: 选举完成，开始正常服务

    N1->>N1: LeaderHeartbeatService.start()

    loop 定期心跳
        N1->>N2: Heartbeat (PublishRequest with no-op state)
        N1->>N3: Heartbeat
    end

quorum = ⌊N/2⌋ + 1
例如：3个节点 → quorum = 2
     5个节点 → quorum = 3

sequenceDiagram
    autonumber
    participant Client as 客户端<br/>(创建索引请求)
    participant REST as REST Handler
    participant TAction as TransportCreateIndexAction
    participant MS as MasterService
    participant Coord as Coordinator<br/>(主节点)
    participant N1 as Node 1
    participant N2 as Node 2
    participant N3 as Node 3

    Client->>REST: PUT /my_index
    REST->>TAction: execute(CreateIndexRequest)

    TAction->>MS: submitStateUpdateTask(<br/>"create-index",<br/>CreateIndexTask,<br/>CreateIndexExecutor)

    MS->>MS: 加入任务队列<br/>(Priority.NORMAL)

    Note over MS: 任务调度器处理队列

    MS->>MS: executeTasks()<br/>批量执行同类型任务

    MS->>MS: CreateIndexExecutor.execute(<br/>currentState, tasks)

    MS->>MS: 1. 验证索引名
    MS->>MS: 2. 构建 IndexMetadata
    MS->>MS: 3. 调用 AllocationService.reroute()<br/>分配主分片

    MS->>MS: 4. 构建新 ClusterState<br/>version++, metadata, routingTable

    Note over Coord,N3: 2-Phase Commit 开始

    MS->>Coord: publishClusterState(newState)

    Coord->>Coord: ClusterStatePublisher.publish()

    Note over Coord,N3: Phase 1: Publish (发布)

    par 并发发布到所有节点
        Coord->>N1: PublishRequest<br/>(diff: ClusterState diff)
        N1->>N1: 1. 验证版本号<br/>(version > localVersion)
        N1->>N1: 2. 应用 diff 到本地<br/>newState = diff.apply(localState)
        N1->>N1: 3. 保存到内存<br/>(uncommitted)
        N1-->>Coord: PublishResponse<br/>(ack, currentTerm)
    and
        Coord->>N2: PublishRequest<br/>(diff)
        N2->>N2: 验证并应用 diff
        N2-->>Coord: PublishResponse(ack)
    and
        Coord->>N3: PublishRequest<br/>(diff)
        N3->>N3: 验证并应用 diff
        N3-->>Coord: PublishResponse(ack)
    end

    Coord->>Coord: 等待 quorum ack<br/>(2 out of 3)

    Coord->>Coord: applyCommittedState(newState)<br/>主节点本地提交

    Note over Coord,N3: Phase 2: Commit (提交)

    par 并发发送提交消息
        Coord->>N1: ApplyCommitRequest<br/>(term, version)
        N1->>N1: ClusterApplierService<br/>.onNewClusterState()
        N1->>N1: 1. 调用 ClusterStateApplier
        N1->>N1: 2. 更新 ClusterService.state()
        N1->>N1: 3. 调用 ClusterStateListener
        N1-->>Coord: ApplyCommitResponse
    and
        Coord->>N2: ApplyCommitRequest
        N2->>N2: onNewClusterState()
        N2->>N2: 应用和通知
        N2-->>Coord: ApplyCommitResponse
    and
        Coord->>N3: ApplyCommitRequest
        N3->>N3: onNewClusterState()
        N3->>N3: 应用和通知
        N3-->>Coord: ApplyCommitResponse
    end

    Coord-->>MS: 发布成功
    MS-->>TAction: onSuccess(newState)
    TAction-->>REST: CreateIndexResponse
    REST-->>Client: 200 OK

// 文件：MasterService.java:1572-1613
public void submitTask(String source, T task, TimeValue timeout) {
    queue.add(new Entry<>(source, taskHolder, ...));
    if (queueSize.getAndIncrement() == 0) {
        perPriorityQueue.execute(processor);  // 触发执行
    }
}

// 文件：AllocationService.java:583-592
private void reroute(RoutingAllocation allocation, RerouteStrategy rerouteStrategy) {
    rerouteStrategy.removeDelayMarkers(allocation);
    allocateExistingUnassignedShards(allocation);  // 分配主分片
    rerouteStrategy.execute(allocation);           // 执行再平衡
}

// 文件：ClusterApplierService.java:505-539
private void applyChanges(ClusterState previousState, ClusterState newState, ...) {
    // 1. 连接新节点
    connectToNodesAndWait(newState);

    // 2. 应用设置变更
    clusterSettings.applySettings(incomingSettings);

    // 3. 调用 ClusterStateApplier
    callClusterStateAppliers(clusterChangedEvent, ...);

    // 4. 更新本地状态（使其对外可见）
    state.set(newState);

    // 5. 通知监听器
    callClusterStateListeners(clusterChangedEvent, ...);
}

sequenceDiagram
    autonumber
    participant MS as MasterService
    participant AS as AllocationService
    participant Allocator as DesiredBalanceShardsAllocator
    participant Deciders as AllocationDeciders
    participant Disk as DiskThresholdDecider
    participant Filter as FilterAllocationDecider
    participant RT as RoutingTable

    Note over MS,RT: 触发分片分配：索引创建、节点加入/离开、手动 reroute

    MS->>AS: reroute(clusterState)
    activate AS

    Note over AS: 1. 构建 RoutingAllocation 上下文
    AS->>AS: new RoutingAllocation(<br/>deciders, routingNodes, clusterState, ...)

    Note over AS: 2. 清理无效状态
    AS->>AS: disassociateDeadNodes()<br/>- 移除已离开节点的分片<br/>- 标记分片为 unassigned

    Note over AS: 3. 分配现有未分配分片
    AS->>AS: allocateExistingUnassignedShards(allocation)

    AS->>RT: routingNodes.unassigned().iterator()
    activate RT
    RT-->>AS: UnassignedIterator
    deactivate RT

    loop 遍历未分配的主分片
        AS->>Allocator: allocateUnassigned(shardRouting, allocation, iterator)
        activate Allocator

        Note over Allocator: 查找可分配节点
        Allocator->>Allocator: 获取所有可用节点列表

        loop 对每个候选节点
            Allocator->>Deciders: canAllocate(shard, node, allocation)
            activate Deciders

            Note over Deciders: 依次应用所有决策器

            Deciders->>Disk: canAllocate(shard, node, allocation)
            activate Disk
            Disk->>Disk: 检查磁盘使用率<br/>- 低水位线: 85%<br/>- 高水位线: 90%<br/>- 洪水水位线: 95%
            alt 磁盘空间不足
                Disk-->>Deciders: Decision.NO("磁盘使用率过高")
            else 磁盘空间充足
                Disk-->>Deciders: Decision.YES
            end
            deactivate Disk

            Deciders->>Filter: canAllocate(shard, node, allocation)
            activate Filter
            Filter->>Filter: 检查节点属性过滤<br/>- index.routing.allocation.require.*<br/>- index.routing.allocation.include.*<br/>- index.routing.allocation.exclude.*
            Filter-->>Deciders: Decision (YES/NO/THROTTLE)
            deactivate Filter

            Note over Deciders: 其他决策器：<br/>- SameShardAllocationDecider<br/>- AwarenessAllocationDecider<br/>- ThrottlingAllocationDecider<br/>- ...

            Deciders-->>Allocator: Decision (YES/NO/THROTTLE)
            deactivate Deciders
        end

        Note over Allocator: 选择最佳节点
        Allocator->>Allocator: 根据决策结果选择节点<br/>- 优先选择 Decision.YES<br/>- 考虑负载均衡<br/>- 计算分片大小和节点容量

        alt 找到可分配节点
            Allocator->>RT: initialize(shard, targetNode)
            activate RT
            RT->>RT: 更新 ShardRouting 状态<br/>- currentNodeId = targetNode<br/>- state = INITIALIZING
            RT-->>Allocator: ok
            deactivate RT
            Allocator-->>AS: 分配成功
        else 未找到可分配节点
            Allocator->>Allocator: 记录分配失败原因
            Allocator-->>AS: 分配延迟或失败
        end
        deactivate Allocator
    end

    loop 遍历未分配的副本分片
        AS->>Allocator: allocateUnassigned(replicaShard, ...)
        Note over Allocator: 副本分片分配逻辑类似<br/>但需要确保与主分片不在同一节点
    end

    Note over AS: 4. 执行再平衡
    AS->>Allocator: allocate(allocation)
    activate Allocator

    Note over Allocator: 计算期望平衡状态
    Allocator->>Allocator: computeDesiredBalance()<br/>- 考虑所有分片<br/>- 计算每个节点的理想分片分布<br/>- 目标：最小化节点间负载差异

    Note over Allocator: 协调至期望状态
    Allocator->>Allocator: reconcile(desiredBalance, currentState)

    loop 移动分片以达到平衡
        Allocator->>Deciders: canRebalance(shard, allocation)
        Deciders-->>Allocator: Decision

        alt 允许再平衡
            Allocator->>RT: relocateShard(shard, targetNode)
            activate RT
            RT->>RT: 更新 ShardRouting<br/>- state = RELOCATING<br/>- relocatingNodeId = targetNode
            RT-->>Allocator: ok
            deactivate RT
        end
    end
    deactivate Allocator

    Note over AS: 5. 返回更新后的 RoutingTable
    AS->>AS: RoutingAllocation.routingTable()
    AS-->>MS: 更新后的 RoutingTable
    deactivate AS

    MS->>MS: 构建新 ClusterState<br/>with updated RoutingTable

// 文件：AllocationService.java:583-592
private void reroute(RoutingAllocation allocation, RerouteStrategy rerouteStrategy) {
    assert hasDeadNodes(allocation) == false : "dead nodes should be explicitly cleaned up";
    rerouteStrategy.removeDelayMarkers(allocation);
    allocateExistingUnassignedShards(allocation);  // 分配现有分片
    rerouteStrategy.execute(allocation);           // 执行策略（含再平衡）
    assert RoutingNodes.assertShardStats(allocation.routingNodes());
}

// 节点权重因素：
weight = α * shardCount + β * diskUsage + γ * writeLoad

// α, β, γ 是可配置的权重系数

STARTED → RELOCATING (源节点)
      ↓
INITIALIZING (目标节点)
      ↓
STARTED (目标节点) + STARTED 删除 (源节点)

// 文件：DesiredBalanceReconciler.java:155-166
private void moveShards() {
    // 2. 移动无法留在当前节点的分片（必须移动）
    // 3. 移动期望在其他节点的分片（再平衡）

    for (ShardRouting shard : assignment.nodeIds()) {
        if (canRemainOnCurrentNode(shard) == false) {
            // 强制移动
            moveToDesiredNode(shard);
        } else if (shouldRebalance(shard)) {
            // 可选移动（优化平衡）
            rebalanceToDesiredNode(shard);
        }
    }
}

sequenceDiagram
    autonumber
    participant NewNode as 新节点<br/>(Node N)
    participant Seed as 种子节点<br/>(Seed Node)
    participant Leader as 主节点<br/>(Leader)
    participant Coord as Leader.Coordinator
    participant JoinHelper as JoinHelper
    participant JoinQueue as Join Task Queue
    participant MS as MasterService
    participant Alloc as AllocationService

    Note over NewNode,Alloc: 新节点启动，尝试加入集群

    NewNode->>NewNode: 启动 Coordinator<br/>mode = CANDIDATE

    Note over NewNode: 1. 节点发现阶段
    NewNode->>NewNode: PeerFinder.activate()<br/>读取 discovery.seed_hosts 配置

    NewNode->>Seed: 发送节点发现请求<br/>(PeerFinder Request)
    Seed-->>NewNode: 返回已知节点列表<br/>(包括 Leader)

    NewNode->>NewNode: 更新已知节点列表<br/>discoveredNodes.add(Leader)

    Note over NewNode: 2. 检查主节点健康
    NewNode->>Leader: 发送 Ping 请求
    Leader-->>NewNode: Pong (term, mode=LEADER)

    NewNode->>NewNode: 识别到 Leader<br/>lastKnownLeader = Leader

    Note over NewNode: 3. 发起加入请求
    NewNode->>NewNode: 构建 JoinRequest<br/>- sourceNode: NewNode<br/>- compatibilityVersions<br/>- features<br/>- term: Leader.term

    NewNode->>Leader: JOIN_ACTION_NAME<br/>sendJoinRequest(JoinRequest)
    activate Leader

    Note over Leader: 4. Leader 接收加入请求
    Leader->>Coord: handleJoinRequest(JoinRequest)
    activate Coord

    Note over Coord: 5. 验证节点
    Coord->>Coord: validateJoinRequest(JoinRequest)

    Note over Coord: 验证步骤：
    Coord->>Coord: 1. 检查版本兼容性<br/>ensureVersionBarrier()

    Coord->>Coord: 2. 检查 transport 连接<br/>建立 STATE 通道

    alt 已是主节点
        Coord->>Coord: 3. 验证能否理解当前状态<br/>onJoinValidators.forEach()
    end

    Coord->>NewNode: 发送 JOIN_PING_ACTION<br/>检查 PING 通道健康
    NewNode-->>Coord: Pong

    alt 验证失败
        Coord-->>NewNode: ValidationException<br/>(版本不兼容/连接失败)
        NewNode->>NewNode: 记录失败，稍后重试
    else 验证成功
        Note over Coord: 6. 处理 Join 请求
        Coord->>Coord: CoordinationState.handleJoin(Join)<br/>- 记录节点投票<br/>- 检查是否达到 quorum

        Coord->>JoinHelper: JoinHelper.handleJoinRequest()
        activate JoinHelper

        JoinHelper->>JoinHelper: 将节点加入 pendingJoins 集合

        alt Join 数量达到 quorum
            Note over JoinHelper: 7. 提交加入任务
            JoinHelper->>JoinQueue: submitTask("node-join", JoinTask)
            activate JoinQueue

            JoinQueue->>MS: 加入任务队列<br/>Priority.URGENT
            deactivate JoinQueue
            activate MS

            Note over MS: 8. 执行加入任务
            MS->>MS: NodeJoinExecutor.execute(currentState, tasks)

            MS->>MS: 验证和更新：<br/>1. 再次检查版本兼容性<br/>2. 检查节点角色<br/>3. 更新 DiscoveryNodes<br/>4. 更新 node features

            alt 新节点首次加入
                MS->>MS: DiscoveryNodes.Builder<br/>.add(newNode)
            else 现有节点重新加入
                MS->>MS: 更新节点信息和 features
            end

            Note over MS: 9. 分片重新分配
            MS->>Alloc: allocationService.reroute(currentState)
            activate Alloc

            Note over Alloc: 新节点加入后重新分配分片：<br/>1. 可能分配新索引的主分片<br/>2. 可能分配副本分片<br/>3. 执行再平衡
            Alloc->>Alloc: allocateUnassignedShards()
            Alloc->>Alloc: balance()

            Alloc-->>MS: 更新后的 RoutingTable
            deactivate Alloc

            Note over MS: 10. 构建新 ClusterState
            MS->>MS: ClusterState.builder(currentState)<br/>.version++<br/>.nodes(newNodes)<br/>.routingTable(newRouting)<br/>.build()

            Note over MS: 11. 发布新状态
            MS->>Coord: publishClusterState(newState)
            deactivate MS
            deactivate JoinHelper
        end

        Note over Coord: 12. 两阶段提交发布状态
        Coord->>Coord: ClusterStatePublisher.publish(newState)

        par 并发发送到所有节点（包括新节点）
            Coord->>NewNode: PublishRequest<br/>(newState with newNode in DiscoveryNodes)
            Coord->>Leader: PublishRequest (本地应用)
            Coord->>Seed: PublishRequest
        end

        Note over NewNode: 13. 新节点接收集群状态
        NewNode->>NewNode: handlePublishRequest(PublishRequest)
        NewNode->>NewNode: 验证版本和 term
        NewNode->>NewNode: 应用 ClusterState<br/>- 更新 nodes<br/>- 更新 routingTable<br/>- 更新 metadata

        NewNode-->>Coord: PublishResponse (ack)
        Seed-->>Coord: PublishResponse (ack)

        Coord->>Coord: 等待 quorum ack

        Note over Coord: 14. 提交状态
        par 发送提交消息
            Coord->>NewNode: ApplyCommitRequest
            Coord->>Seed: ApplyCommitRequest
        end

        Note over NewNode: 15. 应用状态并成为 Follower
        NewNode->>NewNode: ClusterApplierService<br/>.onNewClusterState(newState)
        NewNode->>NewNode: becomeFollower(Leader)
        NewNode->>NewNode: LeaderChecker.updateLeader(Leader)<br/>启动主节点健康检查

        NewNode-->>Coord: ApplyCommitResponse
        Coord-->>Leader: 发布成功
        deactivate Coord
        Leader-->>NewNode: Join 成功
        deactivate Leader

        Note over NewNode: 16. 新节点完成加入
        NewNode->>NewNode: mode = FOLLOWER<br/>lastKnownLeader = Leader

        Note over NewNode: 17. 开始分片恢复
        loop 恢复分配给该节点的分片
            NewNode->>NewNode: 初始化分片<br/>从主分片或快照恢复数据
        end
    end

// 文件：PeerFinder.java
public void activate(final DiscoveryNodes lastAcceptedNodes) {
    // 从配置和上次状态获取种子节点
    final List<TransportAddress> seedAddresses = seedHostsProvider.getSeedAddresses(...);

    // 向每个种子节点发送探测请求
    for (TransportAddress seedAddress : seedAddresses) {
        transportService.sendRequest(
            seedAddress,
            DISCOVERY_REQUEST_ACTION_NAME,
            new PeersRequest(...),
            responseHandler
        );
    }
}

// 文件：JoinHelper.java:274-290
public void sendJoinRequest(DiscoveryNode destination, long term, Optional<Join> optionalJoin) {
    final JoinRequest joinRequest = new JoinRequest(
        transportService.getLocalNode(),  // sourceNode
        compatibilityVersions,             // 版本兼容性信息
        features,                          // 节点特性
        term,                              // 当前 term
        optionalJoin                       // 可选的投票
    );

    transportService.sendRequest(
        destination,
        JOIN_ACTION_NAME,
        joinRequest,
        responseHandler
    );
}

// 文件：Coordinator.java:703-735
private void validateJoinRequest(JoinRequest joinRequest, ActionListener<Void> validateListener) {
    // 验证版本
    final ClusterState stateForJoinValidation = getStateForJoinValidationService();
    if (stateForJoinValidation != null) {
        onJoinValidators.forEach(a -> a.accept(joinRequest.getSourceNode(), stateForJoinValidation));
        NodeJoinExecutor.ensureVersionBarrier(
            joinRequest.getSourceNode().getVersion(),
            stateForJoinValidation.getNodes().getMinNodeVersion()
        );
    }

    // 验证连接
    sendJoinValidate(joinRequest.getSourceNode(), validateListener);
}

// 文件：NodeJoinExecutor.java:157-182
public ClusterState execute(ClusterState currentState, List<JoinTask> joinTasks) {
    final DiscoveryNodes.Builder nodesBuilder = DiscoveryNodes.builder(currentState.nodes());
    final Map<String, Set<String>> nodeFeatures = new HashMap<>(currentState.nodeFeatures());
    boolean nodesChanged = false;

    for (JoinTask joinTask : joinTasks) {
        final DiscoveryNode node = joinTask.node();

        if (currentNodes.nodeExistsWithSameRoles(node)) {
            // 现有节点重新加入，更新特性
            if (!Objects.equals(nodeFeatures.get(node.getId()), joinTask.features())) {
                nodeFeatures.put(node.getId(), joinTask.features());
                nodesChanged = true;
            }
        } else {
            // 新节点加入
            CompatibilityVersions compatibilityVersions = joinTask.compatibilityVersions();
            Set<String> features = joinTask.features();

            // 再次验证版本
            ensureVersionBarrier(node.getVersion(), minClusterNodeVersion);
            CompatibilityVersions.ensureVersionsCompatibility(compatibilityVersions, ...);

            // 添加节点
            nodesBuilder.add(node);
            nodeFeatures.put(node.getId(), features);
            nodesChanged = true;
        }
    }

    if (nodesChanged == false) {
        return currentState;  // 无变化，返回原状态
    }

    // 构建新状态
    return ClusterState.builder(currentState)
        .nodes(nodesBuilder)
        .nodeFeatures(nodeFeatures)
        .build();
}

// 文件：ClusterApplierService.java:505-539
private void applyChanges(ClusterState previousState, ClusterState newState, ...) {
    // 1. 连接集群中的其他节点
    connectToNodesAndWait(newState);

    // 2. 应用集群设置
    clusterSettings.applySettings(newState.metadata().settings());

    // 3. 调用 ClusterStateApplier（前置处理）
    callClusterStateAppliers(clusterChangedEvent, ...);

    // 4. 更新本地状态
    state.set(newState);

    // 5. 通知监听器（后置处理）
    callClusterStateListeners(clusterChangedEvent, ...);
}

GET /_cluster/health

{
  "status": "green",  // green / yellow / red
  "number_of_nodes": 3,
  "number_of_data_nodes": 3,
  "active_primary_shards": 10,
  "active_shards": 20,
  "relocating_shards": 0,
  "initializing_shards": 0,
  "unassigned_shards": 0
}

GET /_cluster/stats

{
  "nodes": {
    "count": {
      "total": 3,
      "master": 3,
      "data": 3
    }
  },
  "indices": {
    "count": 5,
    "shards": {
      "total": 20,
      "primaries": 10
    }
  }
}

HTTP Client → RestClusterHealthAction → NodeClient →
TransportClusterHealthAction → ClusterService.state() →
计算健康状态 → 返回响应

HTTP Client → RestCreateIndexAction → TransportCreateIndexAction →
MasterService.submitTask() → 任务队列 →
CreateIndexExecutor.execute() → AllocationService.reroute() →
Coordinator.publishClusterState() → 两阶段提交 →
所有节点应用状态 → 返回响应

新节点启动 → PeerFinder 发现 → 发送 JoinRequest →
Leader 验证 → 提交 JoinTask → NodeJoinExecutor.execute() →
更新 DiscoveryNodes → 重新分配分片 →
发布新状态 → 新节点成为 Follower → 开始分片恢复

quorum = ⌊N/2⌋ + 1
- 3节点: quorum = 2
- 5节点: quorum = 3
- 7节点: quorum = 4