本文基于Elasticsearch6.8.5版本

ES使用的是Guice框架，依赖注入和暴露接口的方式和Spring差距较大，可先查看guice框架

节点启动过程：

org/elasticsearch/bootstrap/Elasticsearch.java（main）---》org/elasticsearch/node/Node.java（构造方法）加载插件&module---其中包括---》org/elasticsearch/xpack/ccr/Ccr.java（注册各种action）

CCR是插件的形式：

Ccr extends Plugin implements ActionPlugin, PersistentTaskPlugin, EnginePlugin, RepositoryPlugin

CCR的几个操作：

官网操作API：https://www.elastic.co/guide/en/elasticsearch/reference/7.x/ccr-apis.html

1.本地集群连接远程集群

PUT /_cluster/settings
{
  "persistent" : {
    "cluster" : {
      "remote" : {
        "leader" : {
          "seeds" : [
            "127.0.0.1:9300" 
          ]
        }
      }
    }
  }
}

2.在远程集群创建leader索引

远程集群创建leader索引，和正常创建索引操作一样，需要开启soft_deletes（7.x版本已经默认开启）

3.在本地集群创建follower索引

手动创建：需要指定远程集群和复制的远程索引

PUT /<follower_index>/_ccr/follow?wait_for_active_shards=1
{ 
   "remote_cluster" : "<remote_cluster>", 
   "leader_index" : "<leader_index>" 
} 

自动跟随创建：通过auto_follow API建立自动跟随的模板

PUT /_ccr/auto_follow/<auto_follow_pattern_name>
{
  "remote_cluster" : "<remote_cluster>",
  "leader_index_patterns" :
  [
    "<leader_index_pattern>"
  ],
  "follow_index_pattern" : "<follow_index_pattern>"
}

源码分析：

1.执行 _ccr/follow接口，手动创建一个follow索引，该接口会生成索引并完成全量同步

org.elasticsearch.xpack.ccr.rest.RestPutFollowAction类暴露该接口

/**
     * 暴露 {index}/_ccr/follow 接口
     * @param settings
     * @param controller
     */
    public RestPutFollowAction(Settings settings, RestController controller) {
        super(settings);
        controller.registerHandler(RestRequest.Method.PUT, "/{index}/_ccr/follow", this);
    }

通过import关联到org.elasticsearch.xpack.core.ccr.action.PutFollowAction类

import static org.elasticsearch.xpack.core.ccr.action.PutFollowAction.INSTANCE;

PutFollowAction动作在ccr.java中注册了处理类：TransportPutFollowAction

TransportMasterNodeAction将请求转发给follow集群的master节点

在该方法中org.elasticsearch.action.support.master.TransportMasterNodeAction.AsyncSingleAction#doStart

执行：transportService.sendRequest这一行，实现请求转发（如果当前节点就是master，则不转发）

-->follow集群的master节点执行TransportMasterNodeAction.masterOperation()

-->TransportPutFollowAction.createFollowerIndex()

-->restoreSnapshot() 从"_ccr_$remoteClusterName"的repository的"_latest_"快照恢复

-->afterRestoreStarted()

--->initiateFollowing() 该方法核心主要是构造一个ResumeFollowRequest请求，ResumeFollowRequest请求的handler会对每个主分片构造一个任务，去不断刷新leader的更新数据（由此开始增量同步，上面的流程主要是全量同步）

---->TransportResumeFollowAction.start() 开始恢复数据并且修改index状态

----->persistentTasksService.sendStartRequest( ShardFollowTask.NAME, shardFollowTask) 作用：Notifies the master node to create new persistent task and to assign it to a node（通知follow集群的主节点创建新的持久性任务---增量同步数据，long pull，并将其分配给各个节点）

// 执行sendStartRequest之前先create了一个task
final ShardFollowTask shardFollowTask = createShardFollowTask(shardId, clusterNameAlias, request.getParameters(),
                leaderIndexMetadata, followIndexMetadata, filteredHeaders);

// 循环，对每个shard产生task
for (int shardId = 0; shardId < numShards; shardId++) {
            String taskId = followIndexMetadata.getIndexUUID() + "-" + shardId;
            final ShardFollowTask shardFollowTask = createShardFollowTask(shardId, clusterNameAlias, request.getParameters(),
                leaderIndexMetadata, followIndexMetadata, filteredHeaders);
            persistentTasksService.sendStartRequest(taskId, ShardFollowTask.NAME, shardFollowTask, handler.getActionListener(shardId));
        }

到此，只剩下增量同步的逻辑，全部由ShardFollowTask完成；

PS：入口类Ccr在getPersistentTasksExecutor方法中，完成了对ShardFollowTasksExecutor的注册，该类为执行ShardFollowTask的执行器

如上sendStartRequest方法发送一个Task时，会触发到org.elasticsearch.cluster.service.ClusterApplierService#runTask监听，开始执行ShardFollowTasksExecutor中的方法;

------>org.elasticsearch.xpack.ccr.action.ShardFollowNodeTask#start 开启增量复制

------->org.elasticsearch.xpack.ccr.action.ShardFollowNodeTask#coordinateReads

---->org.elasticsearch.xpack.ccr.action.ShardFollowNodeTask#sendShardChangesRequest(long, int, long) 

----->org.elasticsearch.xpack.ccr.action.ShardFollowNodeTask#innerSendShardChangesRequest

作用：获取setting、mapping、translog、 leaderGlobalCheckpoint、leaderMaxSeqNo并更新

--->org.elasticsearch.xpack.ccr.action.ShardFollowNodeTask#handleReadResponse

--->org.elasticsearch.xpack.ccr.action.ShardFollowNodeTask#innerHandleReadResponse

--->org.elasticsearch.xpack.ccr.action.ShardFollowNodeTask#coordinateWrites

--->org.elasticsearch.xpack.ccr.action.ShardFollowNodeTask#sendBulkShardOperationsRequest

--->org.elasticsearch.xpack.ccr.action.ShardFollowNodeTask#

innerSendBulkShardOperationsRequest

作用：写入数据， 得到followerGlobalCheckpoint、followerMaxSeqNo并更新

2.执行 /auto_follow接口(当有新的集群创建时，自动触发全量同步)

1. 执行该API时，集群状态中会保存该pattern

2. Master节点上的AutoFollowCoordinator会一直监听remoteCluster的集群状态，如果发现新创建了索引，且符合pattern，则调用TransportPutFollowAction，执行与上述 _ccr/follow 接口一样的流程

请求接收细节流程：

在follow集群执行该api时代码执行流程（只关心CCR逻辑时可忽略）

1.接收到请求，经过netty的各个方法

2.org.elasticsearch.rest.RestController#dispatchRequest

(在RestController类中初始化NodeClient client，往后传递)

3.解析&处理该request

4.到达 org.elasticsearch.xpack.ccr.rest.RestPutFollowAction#getName

5.到org.elasticsearch.xpack.core.ccr.action.PutFollowAction#fromXContent 进一步处理request

6.执行client.execute到 org.elasticsearch.client.support.AbstractClient#execute

7.回到AbstractClient的实现类 org.elasticsearch.client.node.NodeClient#doExecute

8.到org.elasticsearch.client.node.NodeClient#executeLocally方法中执行transportAction(action).execute(request, listener); 到此关联到了transportAction

9.到org.elasticsearch.action.support.TransportAction#execute，继续执行到该行代码：requestFilterChain.proceed(task, actionName, request, listener);在该类继续执行到 this.action.filters[i].apply(task, actionName, request, listener, this);

10.到org.elasticsearch.xpack.security.action.filter.SecurityActionFilter#apply

执行到chain.proceed(task, action, request, listener);

11.到org.elasticsearch.action.support.TransportAction.RequestFilterChain#proceed

执行this.action.doExecute(task, request, listener);

12.到TransportMasterNodeAction类

org.elasticsearch.action.support.master.TransportMasterNodeAction#doExecute(org.elasticsearch.tasks.Task, Request, org.elasticsearch.action.ActionListener<Response>)

开启线程：new AsyncSingleAction(task, request, listener).start();

13.接下来是几个线程相关类

org.elasticsearch.common.util.concurrent.EsExecutors.DirectExecutorService#execute

org.elasticsearch.common.util.concurrent.AbstractRunnable#run

14.开始执行masterOperation方法，执行到org.elasticsearch.action.support.master.TransportMasterNodeAction#masterOperation(org.elasticsearch.tasks.Task, Request, org.elasticsearch.cluster.ClusterState, org.elasticsearch.action.ActionListener<Response>)

15.进一步找到masterOperation方法实现类TransportPutFollowAction

org.elasticsearch.xpack.ccr.action.TransportPutFollowAction#masterOperation

到此：关联到TransportPutFollowAction类

通过入口类的action绑定，完成在action中找到对应实现类的过程

重点：

增量同步是如何实现不断刷新的（long pull模式）？？？

1.coordinateReads方法读取leader集群信息以及operations

2.org.elasticsearch.xpack.ccr.action.ShardFollowNodeTask#start

coordinateReads->sendShardChangesRequest(long, int, long)->sendShardChangesRequest(long, int, long, java.util.concurrent.atomic.AtomicInteger)->handleReadResponse->innerHandleReadResponse(执行了coordinateWrites)->coordinateReads/sendShardChangesRequest

TODO：

http请求（ccr/follow）接收到后，follow集群节点开始全量同步，是以snapshot的模式去拉leader集群数据的，那么是在什么时候将leader集群伪装成snapshot的repository的？理论上应该是在Node初始化的时候...还未验证，后续再补充该逻辑...

PS:

如有缺失或错误，欢迎补充&指正...