带着疑问学源码，第五篇：Elasticsearch 节点启动分析
代码分析基于：https://github.com/jiankunking/elasticsearch
Elasticsearch 8.0.0-SNAPSHOT

目的

在看源码之前先梳理一下，自己对于节点启动流程疑惑的点：

节点启动都做了哪些检查？
节点启动都初始化了哪些内容？
当节点启动后，数据迁移是在哪里处理？

源码分析

先从启动脚本中找到启动类的入口:org.elasticsearch.bootstrap.Elasticsearch。

下面看一下org.elasticsearch.bootstrap.Elasticsearch,先看一下主入口函数：

/**
 * Main entry point for starting elasticsearch
 */
public static void main(final String[] args) throws Exception {
    // 根据jvm.options中读取：es.networkaddress.cache.ttl和es.networkaddress.cache.negative.ttl
    // 并覆盖JVM Security中的networkaddress.cache.ttl与networkaddress.cache.negative.ttl
    overrideDnsCachePolicyProperties();
    /*
     * We want the JVM to think there is a security manager installed so that if internal policy decisions that would be based on the
     * presence of a security manager or lack thereof act as if there is a security manager present (e.g., DNS cache policy). This
     * forces such policies to take effect immediately.
     */
    System.setSecurityManager(new SecurityManager() {

        @Override
        public void checkPermission(Permission perm) {
            // grant all permissions so that we can later set the security manager to the one that we want
        }

    });
    LogConfigurator.registerErrorListener();
    final Elasticsearch elasticsearch = new Elasticsearch();
    // 核心检查处理都在main(final String[] args, final Elasticsearch elasticsearch, final Terminal terminal)方法中
    int status = main(args, elasticsearch, Terminal.DEFAULT);
    if (status != ExitCodes.OK) {
        final String basePath = System.getProperty("es.logs.base_path");
        // It's possible to fail before logging has been configured, in which case there's no point
        // suggesting that the user look in the log file.
        if (basePath != null) {
            Terminal.DEFAULT.errorPrintln(
                "ERROR: Elasticsearch did not exit normally - check the logs at "
                    + basePath
                    + System.getProperty("file.separator")
                    + System.getProperty("es.logs.cluster_name") + ".log"
            );
        }
        exit(status);
    }
}

main的处理逻辑如下：

Elasticsearch main(final String[] args)=>
Elasticsearch main(final String[] args, final Elasticsearch elasticsearch, final Terminal terminal)=>
Command main(String[] args, Terminal terminal)=>
EnvironmentAwareCommand execute(Terminal terminal, OptionSet options)=>
Elasticsearch execute(Terminal terminal, OptionSet options, Environment env)=>
Bootstrap static void init(
            final boolean foreground,
            final Path pidFile,
            final boolean quiet,
            final Environment initialEnv)=>

下面看一下Bootstrap.init

 /**
 * This method is invoked by {@link Elasticsearch#main(String[])} to startup elasticsearch.
 */
static void init(
        final boolean foreground,
        final Path pidFile,
        final boolean quiet,
        final Environment initialEnv) throws BootstrapException, NodeValidationException, UserException {
    // force the class initializer for BootstrapInfo to run before
    // the security manager is installed
    BootstrapInfo.init();

    INSTANCE = new Bootstrap();

    final SecureSettings keystore = loadSecureSettings(initialEnv);
    final Environment environment = createEnvironment(pidFile, keystore, initialEnv.settings(), initialEnv.configFile());

    // the LogConfigurator will replace System.out and System.err with redirects to our logfile, so we need to capture
    // the stream objects before calling LogConfigurator to be able to close them when appropriate
    final Runnable sysOutCloser = getSysOutCloser();
    final Runnable sysErrorCloser = getSysErrorCloser();

    LogConfigurator.setNodeName(Node.NODE_NAME_SETTING.get(environment.settings()));
    try {
        LogConfigurator.configure(environment);
    } catch (IOException e) {
        throw new BootstrapException(e);
    }
    if (environment.pidFile() != null) {
        try {
            PidFile.create(environment.pidFile(), true);
        } catch (IOException e) {
            throw new BootstrapException(e);
        }
    }


    try {
        final boolean closeStandardStreams = (foreground == false) || quiet;
        if (closeStandardStreams) {
            final Logger rootLogger = LogManager.getRootLogger();
            final Appender maybeConsoleAppender = Loggers.findAppender(rootLogger, ConsoleAppender.class);
            if (maybeConsoleAppender != null) {
                Loggers.removeAppender(rootLogger, maybeConsoleAppender);
            }
            sysOutCloser.run();
        }

        // fail if somebody replaced the lucene jars
        // 检查 Lucene 版本，ES 各个版本对使用的 Lucene 版本是有要求的
        // 在这里检查Lucene版本以防止有人替换不兼容的jar包。
        checkLucene();

        // install the default uncaught exception handler; must be done before security is
        // initialized as we do not want to grant the runtime permission
        // setDefaultUncaughtExceptionHandler
        // 会根据不同的异常，设置不同的exit code
        // InternalError 128
        // OutOfMemoryError 127
        // StackOverflowError 126
        // UnknownError 125
        // IOError 124
        // 其它 1
        Thread.setDefaultUncaughtExceptionHandler(new ElasticsearchUncaughtExceptionHandler());

        // 检查启动es的用户
        // 检查JNA(系统调用)
        // 检查MEMORY_LOCK
        // 检查MaxNumberOfThreads
        // 检查MaxSizeVirtualMemory
        // 检查MaxFileSize
        // init lucene random seed
        // 注册JVM addShutdownHook(Node退出的时候，会用到)
        // 检查jar冲突
        // 初始化JVM Security
        // Node实例添加validateNodeBeforeAcceptingRequests，并初始化Node实例。
        INSTANCE.setup(true, environment);

        try {
            // any secure settings must be read during node construction
            IOUtils.close(keystore);
        } catch (IOException e) {
            throw new BootstrapException(e);
        }

        // 1、开始启动各子模块。
        // 子模块在Node类中创建、启动
        // 子模块的start方法基本就是初始化内部数据、创建线程池、启动线程池等操作。
        // 2、调用keepAliveThread.start()方法启动keepalive线程，线程本身不做具体的工作。
        // 主线程执行完启动流程后会退出，keepalive线程是唯一的用户线程，
        // 作用是保持进程运行。在Java程序中，至少要有一个用户线程。当用户线程数为零时退出进程。
        INSTANCE.start();

        // We don't close stderr if `--quiet` is passed, because that
        // hides fatal startup errors. For example, if Elasticsearch is
        // running via systemd, the init script only specifies
        // `--quiet`, not `-d`, so we want users to be able to see
        // startup errors via journalctl.
        if (foreground == false) {
            sysErrorCloser.run();
        }

    } catch (NodeValidationException | RuntimeException e) {
        // disable console logging, so user does not see the exception twice (jvm will show it already)
        final Logger rootLogger = LogManager.getRootLogger();
        final Appender maybeConsoleAppender = Loggers.findAppender(rootLogger, ConsoleAppender.class);
        if (foreground && maybeConsoleAppender != null) {
            Loggers.removeAppender(rootLogger, maybeConsoleAppender);
        }
        Logger logger = LogManager.getLogger(Bootstrap.class);
        // HACK, it sucks to do this, but we will run users out of disk space otherwise
        if (e instanceof CreationException) {
            // guice: log the shortened exc to the log file
            ByteArrayOutputStream os = new ByteArrayOutputStream();
            PrintStream ps = null;
            try {
                ps = new PrintStream(os, false, "UTF-8");
            } catch (UnsupportedEncodingException uee) {
                assert false;
                e.addSuppressed(uee);
            }
            new StartupException(e).printStackTrace(ps);
            ps.flush();
            try {
                logger.error("Guice Exception: {}", os.toString("UTF-8"));
            } catch (UnsupportedEncodingException uee) {
                assert false;
                e.addSuppressed(uee);
            }
        } else if (e instanceof NodeValidationException) {
            logger.error("node validation exception\n{}", e.getMessage());
        } else {
            // full exception
            logger.error("Exception", e);
        }
        // re-enable it if appropriate, so they can see any logging during the shutdown process
        if (foreground && maybeConsoleAppender != null) {
            Loggers.addAppender(rootLogger, maybeConsoleAppender);
        }

        throw e;
    }
}

下面看一下Node实例初始化及启动部分：

// 环境变量中携带的信息主要节点的配置信息：
  // dataFiles、configFile、pluginsFile、modulesFile等等
  // https://github.com/jiankunking/elasticsearch/blob/master/server/src/main/java/org/elasticsearch/env/Environment.java
  public Node(Environment environment) {
    this(environment, Collections.emptyList(), true);
  }

  /**
   * Constructs a node
   *
   * @param initialEnvironment         the initial environment for this node, which will be added to by plugins
   * @param classpathPlugins           the plugins to be loaded from the classpath
   * @param forbidPrivateIndexSettings whether or not private index settings are forbidden when creating an index; this is used in the
   *                                   test framework for tests that rely on being able to set private settings
   */
  protected Node(
    final Environment initialEnvironment,
    Collection<Class<? extends Plugin>> classpathPlugins,
    boolean forbidPrivateIndexSettings
  ) {
    final List<Closeable> resourcesToClose = new ArrayList<>(); // register everything we need to release in the case of an error
    boolean success = false;
    try {
      Settings tmpSettings = Settings
        .builder()
        .put(initialEnvironment.settings())
        .put(Client.CLIENT_TYPE_SETTING_S.getKey(), CLIENT_TYPE)
        .build();

      final JvmInfo jvmInfo = JvmInfo.jvmInfo();
      logger.info(
        "version[{}], pid[{}], build[{}/{}/{}/{}], OS[{}/{}/{}], JVM[{}/{}/{}/{}]",
        Build.CURRENT.getQualifiedVersion(),
        jvmInfo.pid(),
        Build.CURRENT.flavor().displayName(),
        Build.CURRENT.type().displayName(),
        Build.CURRENT.hash(),
        Build.CURRENT.date(),
        Constants.OS_NAME,
        Constants.OS_VERSION,
        Constants.OS_ARCH,
        Constants.JVM_VENDOR,
        Constants.JVM_NAME,
        Constants.JAVA_VERSION,
        Constants.JVM_VERSION
      );
      if (jvmInfo.getBundledJdk()) {
        logger.info(
          "JVM home [{}], using bundled JDK [{}]",
          System.getProperty("java.home"),
          jvmInfo.getUsingBundledJdk()
        );
      } else {
        logger.info("JVM home [{}]", System.getProperty("java.home"));
        deprecationLogger.deprecate(
          "no-jdk",
          "no-jdk distributions that do not bundle a JDK are deprecated and will be removed in a future release"
        );
      }
      logger.info(
        "JVM arguments {}",
        Arrays.toString(jvmInfo.getInputArguments())
      );
      if (Build.CURRENT.isProductionRelease() == false) {
        logger.warn(
          "version [{}] is a pre-release version of Elasticsearch and is not suitable for production",
          Build.CURRENT.getQualifiedVersion()
        );
      }

      if (logger.isDebugEnabled()) {
        logger.debug(
          "using config [{}], data [{}], logs [{}], plugins [{}]",
          initialEnvironment.configFile(),
          Arrays.toString(initialEnvironment.dataFiles()),
          initialEnvironment.logsFile(),
          initialEnvironment.pluginsFile()
        );
      }

      // 创建PluginsService，加载modules目录下的所有模块和plugins目录下的所有插件
      // https://github.com/jiankunking/elasticsearch/blob/master/server/src/main/java/org/elasticsearch/plugins/PluginsService.java
      this.pluginsService =
        new PluginsService(
          tmpSettings,
          initialEnvironment.configFile(),
          initialEnvironment.modulesFile(),
          initialEnvironment.pluginsFile(),
          classpathPlugins
        );
      final Settings settings = pluginsService.updatedSettings();

      final Set<DiscoveryNodeRole> additionalRoles = pluginsService
        .filterPlugins(Plugin.class)
        .stream()
        .map(Plugin::getRoles)
        .flatMap(Set::stream)
        .collect(Collectors.toSet());
      DiscoveryNode.setAdditionalRoles(additionalRoles);

      /*
       * Create the environment based on the finalized view of the settings. This is to ensure that components get the same setting
       * values, no matter they ask for them from.
       */
      this.environment =
        new Environment(settings, initialEnvironment.configFile());
      Environment.assertEquivalent(initialEnvironment, this.environment);
      nodeEnvironment = new NodeEnvironment(tmpSettings, environment);
      logger.info(
        "node name [{}], node ID [{}], cluster name [{}], roles {}",
        NODE_NAME_SETTING.get(tmpSettings),
        nodeEnvironment.nodeId(),
        ClusterName.CLUSTER_NAME_SETTING.get(tmpSettings).value(),
        DiscoveryNode
          .getRolesFromSettings(settings)
          .stream()
          .map(DiscoveryNodeRole::roleName)
          .collect(Collectors.toCollection(LinkedHashSet::new))
      );
      resourcesToClose.add(nodeEnvironment);
      localNodeFactory =
        new LocalNodeFactory(settings, nodeEnvironment.nodeId());

      // 调用各插件的getExecutorBuilders，获取ExecutorBuilder
      final List<ExecutorBuilder<?>> executorBuilders = pluginsService.getExecutorBuilders(
        settings
      );
      // 创建线程池
      final ThreadPool threadPool = new ThreadPool(
        settings,
        executorBuilders.toArray(new ExecutorBuilder[0])
      );
      resourcesToClose.add(
        () -> ThreadPool.terminate(threadPool, 10, TimeUnit.SECONDS)
      );

      final ResourceWatcherService resourceWatcherService = new ResourceWatcherService(
        settings,
        threadPool
      );
      resourcesToClose.add(resourceWatcherService);
      // adds the context to the DeprecationLogger so that it does not need to be injected everywhere
      HeaderWarning.setThreadContext(threadPool.getThreadContext());
      resourcesToClose.add(
        () -> HeaderWarning.removeThreadContext(threadPool.getThreadContext())
      );

      final List<Setting<?>> additionalSettings = new ArrayList<>();
      // register the node.data, node.ingest, node.master, node.remote_cluster_client settings here so we can mark them private
      additionalSettings.add(NODE_DATA_SETTING);
      additionalSettings.add(NODE_INGEST_SETTING);
      additionalSettings.add(NODE_MASTER_SETTING);
      additionalSettings.add(NODE_REMOTE_CLUSTER_CLIENT);
      additionalSettings.addAll(pluginsService.getPluginSettings());
      final List<String> additionalSettingsFilter = new ArrayList<>(
        pluginsService.getPluginSettingsFilter()
      );
      for (final ExecutorBuilder<?> builder : threadPool.builders()) {
        additionalSettings.addAll(builder.getRegisteredSettings());
      }
      // 创建NodeClient
      client = new NodeClient(settings, threadPool);

      // 创建各种***Service对象和各种模***Module对象
      final ScriptModule scriptModule = new ScriptModule(
        settings,
        pluginsService.filterPlugins(ScriptPlugin.class)
      );
      final ScriptService scriptService = newScriptService(
        settings,
        scriptModule.engines,
        scriptModule.contexts
      );
      AnalysisModule analysisModule = new AnalysisModule(
        this.environment,
        pluginsService.filterPlugins(AnalysisPlugin.class)
      );
      // this is as early as we can validate settings at this point. we already pass them to ScriptModule as well as ThreadPool
      // so we might be late here already

      final Set<SettingUpgrader<?>> settingsUpgraders = pluginsService
        .filterPlugins(Plugin.class)
        .stream()
        .map(Plugin::getSettingUpgraders)
        .flatMap(List::stream)
        .collect(Collectors.toSet());

      final SettingsModule settingsModule = new SettingsModule(
        settings,
        additionalSettings,
        additionalSettingsFilter,
        settingsUpgraders
      );
      scriptModule.registerClusterSettingsListeners(
        scriptService,
        settingsModule.getClusterSettings()
      );
      final NetworkService networkService = new NetworkService(
        getCustomNameResolvers(
          pluginsService.filterPlugins(DiscoveryPlugin.class)
        )
      );

      List<ClusterPlugin> clusterPlugins = pluginsService.filterPlugins(
        ClusterPlugin.class
      );
      final ClusterService clusterService = new ClusterService(
        settings,
        settingsModule.getClusterSettings(),
        threadPool
      );
      clusterService.addStateApplier(scriptService);
      resourcesToClose.add(clusterService);
      final Set<Setting<?>> consistentSettings = settingsModule.getConsistentSettings();
      if (consistentSettings.isEmpty() == false) {
        clusterService.addLocalNodeMasterListener(
          new ConsistentSettingsService(
            settings,
            clusterService,
            consistentSettings
          )
            .newHashPublisher()
        );
      }
      final IngestService ingestService = new IngestService(
        clusterService,
        threadPool,
        this.environment,
        scriptService,
        analysisModule.getAnalysisRegistry(),
        pluginsService.filterPlugins(IngestPlugin.class),
        client
      );
      final SetOnce<RepositoriesService> repositoriesServiceReference = new SetOnce<>();
      final ClusterInfoService clusterInfoService = newClusterInfoService(
        settings,
        clusterService,
        threadPool,
        client
      );
      final UsageService usageService = new UsageService();

      ModulesBuilder modules = new ModulesBuilder();
      final MonitorService monitorService = new MonitorService(
        settings,
        nodeEnvironment,
        threadPool
      );
      final FsHealthService fsHealthService = new FsHealthService(
        settings,
        clusterService.getClusterSettings(),
        threadPool,
        nodeEnvironment
      );
      final SetOnce<RerouteService> rerouteServiceReference = new SetOnce<>();
      final InternalSnapshotsInfoService snapshotsInfoService = new InternalSnapshotsInfoService(
        settings,
        clusterService,
        repositoriesServiceReference::get,
        rerouteServiceReference::get
      );
      final ClusterModule clusterModule = new ClusterModule(
        settings,
        clusterService,
        clusterPlugins,
        clusterInfoService,
        snapshotsInfoService,
        threadPool.getThreadContext()
      );
      modules.add(clusterModule);
      IndicesModule indicesModule = new IndicesModule(
        pluginsService.filterPlugins(MapperPlugin.class)
      );
      modules.add(indicesModule);

      SearchModule searchModule = new SearchModule(
        settings,
        pluginsService.filterPlugins(SearchPlugin.class)
      );
      List<BreakerSettings> pluginCircuitBreakers = pluginsService
        .filterPlugins(CircuitBreakerPlugin.class)
        .stream()
        .map(plugin -> plugin.getCircuitBreaker(settings))
        .collect(Collectors.toList());
      final CircuitBreakerService circuitBreakerService = createCircuitBreakerService(
        settingsModule.getSettings(),
        pluginCircuitBreakers,
        settingsModule.getClusterSettings()
      );
      pluginsService
        .filterPlugins(CircuitBreakerPlugin.class)
        .forEach(
          plugin -> {
            CircuitBreaker breaker = circuitBreakerService.getBreaker(
              plugin.getCircuitBreaker(settings).getName()
            );
            plugin.setCircuitBreaker(breaker);
          }
        );
      resourcesToClose.add(circuitBreakerService);
      modules.add(new GatewayModule());

      PageCacheRecycler pageCacheRecycler = createPageCacheRecycler(settings);
      BigArrays bigArrays = createBigArrays(
        pageCacheRecycler,
        circuitBreakerService
      );
      modules.add(settingsModule);
      List<NamedWriteableRegistry.Entry> namedWriteables = Stream
        .of(
          NetworkModule.getNamedWriteables().stream(),
          IndicesModule.getNamedWriteables().stream(),
          searchModule.getNamedWriteables().stream(),
          pluginsService
            .filterPlugins(Plugin.class)
            .stream()
            .flatMap(p -> p.getNamedWriteables().stream()),
          ClusterModule.getNamedWriteables().stream()
        )
        .flatMap(Function.identity())
        .collect(Collectors.toList());
      final NamedWriteableRegistry namedWriteableRegistry = new NamedWriteableRegistry(
        namedWriteables
      );
      NamedXContentRegistry xContentRegistry = new NamedXContentRegistry(
        Stream
          .of(
            NetworkModule.getNamedXContents().stream(),
            IndicesModule.getNamedXContents().stream(),
            searchModule.getNamedXContents().stream(),
            pluginsService
              .filterPlugins(Plugin.class)
              .stream()
              .flatMap(p -> p.getNamedXContent().stream()),
            ClusterModule.getNamedXWriteables().stream()
          )
          .flatMap(Function.identity())
          .collect(toList())
      );
      final MetaStateService metaStateService = new MetaStateService(
        nodeEnvironment,
        xContentRegistry
      );
      final PersistedClusterStateService lucenePersistedStateFactory = new PersistedClusterStateService(
        nodeEnvironment,
        xContentRegistry,
        bigArrays,
        clusterService.getClusterSettings(),
        threadPool::relativeTimeInMillis
      );

      // collect engine factory providers from plugins
      final Collection<EnginePlugin> enginePlugins = pluginsService.filterPlugins(
        EnginePlugin.class
      );
      final Collection<Function<IndexSettings, Optional<EngineFactory>>> engineFactoryProviders = enginePlugins
        .stream()
        .map(
          plugin ->
            (Function<IndexSettings, Optional<EngineFactory>>) plugin::getEngineFactory
        )
        .collect(Collectors.toList());

      final Map<String, IndexStorePlugin.DirectoryFactory> indexStoreFactories = pluginsService
        .filterPlugins(IndexStorePlugin.class)
        .stream()
        .map(IndexStorePlugin::getDirectoryFactories)
        .flatMap(m -> m.entrySet().stream())
        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

      final Map<String, IndexStorePlugin.RecoveryStateFactory> recoveryStateFactories = pluginsService
        .filterPlugins(IndexStorePlugin.class)
        .stream()
        .map(IndexStorePlugin::getRecoveryStateFactories)
        .flatMap(m -> m.entrySet().stream())
        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

      final List<IndexStorePlugin.IndexFoldersDeletionListener> indexFoldersDeletionListeners = pluginsService
        .filterPlugins(IndexStorePlugin.class)
        .stream()
        .map(IndexStorePlugin::getIndexFoldersDeletionListeners)
        .flatMap(List::stream)
        .collect(Collectors.toList());

      final Map<String, IndexStorePlugin.SnapshotCommitSupplier> snapshotCommitSuppliers = pluginsService
        .filterPlugins(IndexStorePlugin.class)
        .stream()
        .map(IndexStorePlugin::getSnapshotCommitSuppliers)
        .flatMap(m -> m.entrySet().stream())
        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

      final Map<String, Collection<SystemIndexDescriptor>> systemIndexDescriptorMap = pluginsService
        .filterPlugins(SystemIndexPlugin.class)
        .stream()
        .collect(
          Collectors.toUnmodifiableMap(
            plugin -> plugin.getClass().getSimpleName(),
            plugin -> plugin.getSystemIndexDescriptors(settings)
          )
        );
      final SystemIndices systemIndices = new SystemIndices(
        systemIndexDescriptorMap
      );

      final SystemIndexManager systemIndexManager = new SystemIndexManager(
        systemIndices,
        client
      );
      clusterService.addListener(systemIndexManager);

      final RerouteService rerouteService = new BatchedRerouteService(
        clusterService,
        clusterModule.getAllocationService()::reroute
      );
      rerouteServiceReference.set(rerouteService);
      clusterService.setRerouteService(rerouteService);

      final IndicesService indicesService = new IndicesService(
        settings,
        pluginsService,
        nodeEnvironment,
        xContentRegistry,
        analysisModule.getAnalysisRegistry(),
        clusterModule.getIndexNameExpressionResolver(),
        indicesModule.getMapperRegistry(),
        namedWriteableRegistry,
        threadPool,
        settingsModule.getIndexScopedSettings(),
        circuitBreakerService,
        bigArrays,
        scriptService,
        clusterService,
        client,
        metaStateService,
        engineFactoryProviders,
        indexStoreFactories,
        searchModule.getValuesSourceRegistry(),
        recoveryStateFactories,
        indexFoldersDeletionListeners,
        snapshotCommitSuppliers
      );

      final AliasValidator aliasValidator = new AliasValidator();

      final ShardLimitValidator shardLimitValidator = new ShardLimitValidator(
        settings,
        clusterService
      );
      final MetadataCreateIndexService metadataCreateIndexService = new MetadataCreateIndexService(
        settings,
        clusterService,
        indicesService,
        clusterModule.getAllocationService(),
        aliasValidator,
        shardLimitValidator,
        environment,
        settingsModule.getIndexScopedSettings(),
        threadPool,
        xContentRegistry,
        systemIndices,
        forbidPrivateIndexSettings
      );
      pluginsService
        .filterPlugins(Plugin.class)
        .forEach(
          p ->
            p
              .getAdditionalIndexSettingProviders()
              .forEach(
                metadataCreateIndexService::addAdditionalIndexSettingProvider
              )
        );

      final MetadataCreateDataStreamService metadataCreateDataStreamService = new MetadataCreateDataStreamService(
        threadPool,
        clusterService,
        metadataCreateIndexService
      );

      Collection<Object> pluginComponents = pluginsService
        .filterPlugins(Plugin.class)
        .stream()
        .flatMap(
          p ->
            p
              .createComponents(
                client,
                clusterService,
                threadPool,
                resourceWatcherService,
                scriptService,
                xContentRegistry,
                environment,
                nodeEnvironment,
                namedWriteableRegistry,
                clusterModule.getIndexNameExpressionResolver(),
                repositoriesServiceReference::get
              )
              .stream()
        )
        .collect(Collectors.toList());

      ActionModule actionModule = new ActionModule(
        settings,
        clusterModule.getIndexNameExpressionResolver(),
        settingsModule.getIndexScopedSettings(),
        settingsModule.getClusterSettings(),
        settingsModule.getSettingsFilter(),
        threadPool,
        pluginsService.filterPlugins(ActionPlugin.class),
        client,
        circuitBreakerService,
        usageService,
        systemIndices,
        getRestCompatibleFunction()
      );
      modules.add(actionModule);

      final RestController restController = actionModule.getRestController();
      final NetworkModule networkModule = new NetworkModule(
        settings,
        pluginsService.filterPlugins(NetworkPlugin.class),
        threadPool,
        bigArrays,
        pageCacheRecycler,
        circuitBreakerService,
        namedWriteableRegistry,
        xContentRegistry,
        networkService,
        restController,
        clusterService.getClusterSettings()
      );
      Collection<UnaryOperator<Map<String, IndexTemplateMetadata>>> indexTemplateMetadataUpgraders = pluginsService
        .filterPlugins(Plugin.class)
        .stream()
        .map(Plugin::getIndexTemplateMetadataUpgrader)
        .collect(Collectors.toList());
      final MetadataUpgrader metadataUpgrader = new MetadataUpgrader(
        indexTemplateMetadataUpgraders
      );
      final MetadataIndexUpgradeService metadataIndexUpgradeService = new MetadataIndexUpgradeService(
        settings,
        xContentRegistry,
        indicesModule.getMapperRegistry(),
        settingsModule.getIndexScopedSettings(),
        systemIndices,
        scriptService
      );
      if (DiscoveryNode.isMasterNode(settings)) {
        clusterService.addListener(
          new SystemIndexMetadataUpgradeService(systemIndices, clusterService)
        );
      }
      new TemplateUpgradeService(
        client,
        clusterService,
        threadPool,
        indexTemplateMetadataUpgraders
      );
      final Transport transport = networkModule.getTransportSupplier().get();
      Set<String> taskHeaders = Stream
        .concat(
          pluginsService
            .filterPlugins(ActionPlugin.class)
            .stream()
            .flatMap(p -> p.getTaskHeaders().stream()),
          Stream.of(Task.X_OPAQUE_ID)
        )
        .collect(Collectors.toSet());
      final TransportService transportService = newTransportService(
        settings,
        transport,
        threadPool,
        networkModule.getTransportInterceptor(),
        localNodeFactory,
        settingsModule.getClusterSettings(),
        taskHeaders
      );
      final GatewayMetaState gatewayMetaState = new GatewayMetaState();
      final ResponseCollectorService responseCollectorService = new ResponseCollectorService(
        clusterService
      );
      final SearchTransportService searchTransportService = new SearchTransportService(
        transportService,
        client,
        SearchExecutionStatsCollector.makeWrapper(responseCollectorService)
      );
      final HttpServerTransport httpServerTransport = newHttpTransport(
        networkModule
      );
      final IndexingPressure indexingLimits = new IndexingPressure(settings);

      final RecoverySettings recoverySettings = new RecoverySettings(
        settings,
        settingsModule.getClusterSettings()
      );
      RepositoriesModule repositoriesModule = new RepositoriesModule(
        this.environment,
        pluginsService.filterPlugins(RepositoryPlugin.class),
        transportService,
        clusterService,
        bigArrays,
        xContentRegistry,
        recoverySettings
      );
      RepositoriesService repositoryService = repositoriesModule.getRepositoryService();
      repositoriesServiceReference.set(repositoryService);
      SnapshotsService snapshotsService = new SnapshotsService(
        settings,
        clusterService,
        clusterModule.getIndexNameExpressionResolver(),
        repositoryService,
        transportService,
        actionModule.getActionFilters()
      );
      SnapshotShardsService snapshotShardsService = new SnapshotShardsService(
        settings,
        clusterService,
        repositoryService,
        transportService,
        indicesService
      );
      RestoreService restoreService = new RestoreService(
        clusterService,
        repositoryService,
        clusterModule.getAllocationService(),
        metadataCreateIndexService,
        metadataIndexUpgradeService,
        clusterService.getClusterSettings(),
        shardLimitValidator
      );

      final DiskThresholdMonitor diskThresholdMonitor = new DiskThresholdMonitor(
        settings,
        clusterService::state,
        clusterService.getClusterSettings(),
        client,
        threadPool::relativeTimeInMillis,
        rerouteService
      );
      clusterInfoService.addListener(diskThresholdMonitor::onNewInfo);

      final DiscoveryModule discoveryModule = new DiscoveryModule(
        settings,
        transportService,
        namedWriteableRegistry,
        networkService,
        clusterService.getMasterService(),
        clusterService.getClusterApplierService(),
        clusterService.getClusterSettings(),
        pluginsService.filterPlugins(DiscoveryPlugin.class),
        clusterModule.getAllocationService(),
        environment.configFile(),
        gatewayMetaState,
        rerouteService,
        fsHealthService
      );
      this.nodeService =
        new NodeService(
          settings,
          threadPool,
          monitorService,
          discoveryModule.getDiscovery(),
          transportService,
          indicesService,
          pluginsService,
          circuitBreakerService,
          scriptService,
          httpServerTransport,
          ingestService,
          clusterService,
          settingsModule.getSettingsFilter(),
          responseCollectorService,
          searchTransportService,
          indexingLimits,
          searchModule.getValuesSourceRegistry().getUsageService()
        );

      final SearchService searchService = newSearchService(
        clusterService,
        indicesService,
        threadPool,
        scriptService,
        bigArrays,
        searchModule.getFetchPhase(),
        responseCollectorService,
        circuitBreakerService
      );

      final List<PersistentTasksExecutor<?>> tasksExecutors = pluginsService
        .filterPlugins(PersistentTaskPlugin.class)
        .stream()
        .map(
          p ->
            p.getPersistentTasksExecutor(
              clusterService,
              threadPool,
              client,
              settingsModule,
              clusterModule.getIndexNameExpressionResolver()
            )
        )
        .flatMap(List::stream)
        .collect(toList());

      final PersistentTasksExecutorRegistry registry = new PersistentTasksExecutorRegistry(
        tasksExecutors
      );
      final PersistentTasksClusterService persistentTasksClusterService = new PersistentTasksClusterService(
        settings,
        registry,
        clusterService,
        threadPool
      );
      resourcesToClose.add(persistentTasksClusterService);
      final PersistentTasksService persistentTasksService = new PersistentTasksService(
        clusterService,
        threadPool,
        client
      );
      // 绑定各种服务模块的实例
      modules.add(
        b -> {
          b.bind(Node.class).toInstance(this);
          b.bind(NodeService.class).toInstance(nodeService);
          b.bind(NamedXContentRegistry.class).toInstance(xContentRegistry);
          b.bind(PluginsService.class).toInstance(pluginsService);
          b.bind(Client.class).toInstance(client);
          b.bind(NodeClient.class).toInstance(client);
          b.bind(Environment.class).toInstance(this.environment);
          b.bind(ThreadPool.class).toInstance(threadPool);
          b.bind(NodeEnvironment.class).toInstance(nodeEnvironment);
          b
            .bind(ResourceWatcherService.class)
            .toInstance(resourceWatcherService);
          b.bind(CircuitBreakerService.class).toInstance(circuitBreakerService);
          b.bind(BigArrays.class).toInstance(bigArrays);
          b.bind(PageCacheRecycler.class).toInstance(pageCacheRecycler);
          b.bind(ScriptService.class).toInstance(scriptService);
          b
            .bind(AnalysisRegistry.class)
            .toInstance(analysisModule.getAnalysisRegistry());
          b.bind(IngestService.class).toInstance(ingestService);
          b.bind(IndexingPressure.class).toInstance(indexingLimits);
          b.bind(UsageService.class).toInstance(usageService);
          b
            .bind(AggregationUsageService.class)
            .toInstance(
              searchModule.getValuesSourceRegistry().getUsageService()
            );
          b
            .bind(NamedWriteableRegistry.class)
            .toInstance(namedWriteableRegistry);
          b.bind(MetadataUpgrader.class).toInstance(metadataUpgrader);
          b.bind(MetaStateService.class).toInstance(metaStateService);
          b
            .bind(PersistedClusterStateService.class)
            .toInstance(lucenePersistedStateFactory);
          b.bind(IndicesService.class).toInstance(indicesService);
          b.bind(AliasValidator.class).toInstance(aliasValidator);
          b
            .bind(MetadataCreateIndexService.class)
            .toInstance(metadataCreateIndexService);
          b
            .bind(MetadataCreateDataStreamService.class)
            .toInstance(metadataCreateDataStreamService);
          b.bind(SearchService.class).toInstance(searchService);
          b
            .bind(SearchTransportService.class)
            .toInstance(searchTransportService);
          b
            .bind(SearchPhaseController.class)
            .toInstance(
              new SearchPhaseController(
                namedWriteableRegistry,
                searchService::aggReduceContextBuilder
              )
            );
          b.bind(Transport.class).toInstance(transport);
          b.bind(TransportService.class).toInstance(transportService);
          b.bind(NetworkService.class).toInstance(networkService);
          b
            .bind(UpdateHelper.class)
            .toInstance(new UpdateHelper(scriptService));
          b
            .bind(MetadataIndexUpgradeService.class)
            .toInstance(metadataIndexUpgradeService);
          b.bind(ClusterInfoService.class).toInstance(clusterInfoService);
          b.bind(SnapshotsInfoService.class).toInstance(snapshotsInfoService);
          b.bind(GatewayMetaState.class).toInstance(gatewayMetaState);
          b.bind(Discovery.class).toInstance(discoveryModule.getDiscovery());
          {
            processRecoverySettings(
              settingsModule.getClusterSettings(),
              recoverySettings
            );
            b
              .bind(PeerRecoverySourceService.class)
              .toInstance(
                new PeerRecoverySourceService(
                  transportService,
                  indicesService,
                  recoverySettings
                )
              );
            b
              .bind(PeerRecoveryTargetService.class)
              .toInstance(
                new PeerRecoveryTargetService(
                  threadPool,
                  transportService,
                  recoverySettings,
                  clusterService
                )
              );
          }
          b.bind(HttpServerTransport.class).toInstance(httpServerTransport);
          pluginComponents
            .stream()
            .forEach(p -> b.bind((Class) p.getClass()).toInstance(p));
          b
            .bind(PersistentTasksService.class)
            .toInstance(persistentTasksService);
          b
            .bind(PersistentTasksClusterService.class)
            .toInstance(persistentTasksClusterService);
          b.bind(PersistentTasksExecutorRegistry.class).toInstance(registry);
          b.bind(RepositoriesService.class).toInstance(repositoryService);
          b.bind(SnapshotsService.class).toInstance(snapshotsService);
          b.bind(SnapshotShardsService.class).toInstance(snapshotShardsService);
          b.bind(RestoreService.class).toInstance(restoreService);
          b.bind(RerouteService.class).toInstance(rerouteService);
          b.bind(ShardLimitValidator.class).toInstance(shardLimitValidator);
          b.bind(FsHealthService.class).toInstance(fsHealthService);
          b.bind(SystemIndices.class).toInstance(systemIndices);
        }
      );
      injector = modules.createInjector();

      // We allocate copies of existing shards by looking for a viable copy of the shard in the cluster and assigning the shard there.
      // The search for viable copies is triggered by an allocation attempt (i.e. a reroute) and is performed asynchronously. When it
      // completes we trigger another reroute to try the allocation again. This means there is a circular dependency: the allocation
      // service needs access to the existing shards allocators (e.g. the GatewayAllocator) which need to be able to trigger a
      // reroute, which needs to call into the allocation service. We close the loop here:
      clusterModule.setExistingShardsAllocators(
        injector.getInstance(GatewayAllocator.class)
      );

      List<LifecycleComponent> pluginLifecycleComponents = pluginComponents
        .stream()
        .filter(p -> p instanceof LifecycleComponent)
        .map(p -> (LifecycleComponent) p)
        .collect(Collectors.toList());
      resourcesToClose.addAll(pluginLifecycleComponents);
      resourcesToClose.add(
        injector.getInstance(PeerRecoverySourceService.class)
      );
      this.pluginLifecycleComponents =
        Collections.unmodifiableList(pluginLifecycleComponents);
      client.initialize(
        injector.getInstance(new Key<Map<ActionType, TransportAction>>() {}),
        transportService.getTaskManager(),
        () -> clusterService.localNode().getId(),
        transportService.getLocalNodeConnection(),
        transportService.getRemoteClusterService(),
        namedWriteableRegistry
      );
      this.namedWriteableRegistry = namedWriteableRegistry;

      logger.debug("initializing HTTP handlers ...");
      actionModule.initRestHandlers(() -> clusterService.state().nodes());
      logger.info("initialized");

      success = true;
    } catch (IOException ex) {
      throw new ElasticsearchException("failed to bind service", ex);
    } finally {
      if (!success) {
        IOUtils.closeWhileHandlingException(resourcesToClose);
      }
    }
  }

  /**
   * Start the node. If the node is already started, this method is no-op.
   */
  public Node start() throws NodeValidationException {
    // 将local node的state设为STARTED状态
    if (!lifecycle.moveToStarted()) {
      return this;
    }

    logger.info("starting ...");
    // plugins start
    pluginLifecycleComponents.forEach(LifecycleComponent::start);

    injector.getInstance(MappingUpdatedAction.class).setClient(client);
    injector.getInstance(IndicesService.class).start();
    injector.getInstance(IndicesClusterStateService.class).start();
    injector.getInstance(SnapshotsService.class).start();
    injector.getInstance(SnapshotShardsService.class).start();
    injector.getInstance(RepositoriesService.class).start();
    injector.getInstance(SearchService.class).start();
    injector.getInstance(FsHealthService.class).start();
    nodeService.getMonitorService().start();

    final ClusterService clusterService = injector.getInstance(
      ClusterService.class
    );

    final NodeConnectionsService nodeConnectionsService = injector.getInstance(
      NodeConnectionsService.class
    );
    nodeConnectionsService.start();
    clusterService.setNodeConnectionsService(nodeConnectionsService);

    injector.getInstance(GatewayService.class).start();
    Discovery discovery = injector.getInstance(Discovery.class);
    clusterService
      .getMasterService()
      .setClusterStatePublisher(discovery::publish);

    // Start the transport service now so the publish address will be added to the local disco node in ClusterService
    TransportService transportService = injector.getInstance(
      TransportService.class
    );
    transportService
      .getTaskManager()
      .setTaskResultsService(injector.getInstance(TaskResultsService.class));
    transportService
      .getTaskManager()
      .setTaskCancellationService(
        new TaskCancellationService(transportService)
      );
    transportService.start();
    assert localNodeFactory.getNode() != null;
    assert transportService
      .getLocalNode()
      .equals(
        localNodeFactory.getNode()
      ) : "transportService has a different local node than the factory provided";
    injector.getInstance(PeerRecoverySourceService.class).start();

    // Load (and maybe upgrade) the metadata stored on disk
    final GatewayMetaState gatewayMetaState = injector.getInstance(
      GatewayMetaState.class
    );
    gatewayMetaState.start(
      settings(),
      transportService,
      clusterService,
      injector.getInstance(MetaStateService.class),
      injector.getInstance(MetadataIndexUpgradeService.class),
      injector.getInstance(MetadataUpgrader.class),
      injector.getInstance(PersistedClusterStateService.class)
    );
    if (Assertions.ENABLED) {
      try {
        assert injector
          .getInstance(MetaStateService.class)
          .loadFullState()
          .v1()
          .isEmpty();
        final NodeMetadata nodeMetadata = NodeMetadata.FORMAT.loadLatestState(
          logger,
          NamedXContentRegistry.EMPTY,
          nodeEnvironment.nodeDataPaths()
        );
        assert nodeMetadata != null;
        assert nodeMetadata.nodeVersion().equals(Version.CURRENT);
        assert nodeMetadata.nodeId().equals(localNodeFactory.getNode().getId());
      } catch (IOException e) {
        assert false : e;
      }
    }
    // we load the global state here (the persistent part of the cluster state stored on disk) to
    // pass it to the bootstrap checks to allow plugins to enforce certain preconditions based on the recovered state.
    final Metadata onDiskMetadata = gatewayMetaState
      .getPersistedState()
      .getLastAcceptedState()
      .metadata();
    assert onDiskMetadata != null : "metadata is null but shouldn't"; // this is never null
    validateNodeBeforeAcceptingRequests(
      new BootstrapContext(environment, onDiskMetadata),
      transportService.boundAddress(),
      pluginsService
        .filterPlugins(Plugin.class)
        .stream()
        .flatMap(p -> p.getBootstrapChecks().stream())
        .collect(Collectors.toList())
    );

    clusterService.addStateApplier(transportService.getTaskManager());
    // start after transport service so the local disco is known
    discovery.start(); // start before cluster service so that it can set initial state on ClusterApplierService
    clusterService.start();
    assert clusterService
      .localNode()
      .equals(
        localNodeFactory.getNode()
      ) : "clusterService has a different local node than the factory provided";
    // start accepting incoming requests.
    // when the transport layer starts up it will block any incoming requests until this method is called.
    transportService.acceptIncomingRequests();
    // 一会着重看一下选举部分
    discovery.startInitialJoin();
    final TimeValue initialStateTimeout = INITIAL_STATE_TIMEOUT_SETTING.get(
      settings()
    );
    configureNodeAndClusterIdStateListener(clusterService);

    if (initialStateTimeout.millis() > 0) {
      final ThreadPool thread = injector.getInstance(ThreadPool.class);
      ClusterState clusterState = clusterService.state();
      ClusterStateObserver observer = new ClusterStateObserver(
        clusterState,
        clusterService,
        null,
        logger,
        thread.getThreadContext()
      );

      if (clusterState.nodes().getMasterNodeId() == null) {
        logger.debug(
          "waiting to join the cluster. timeout [{}]",
          initialStateTimeout
        );
        final CountDownLatch latch = new CountDownLatch(1);
        // Wait for the next cluster state which satisfies statePredicate
        observer.waitForNextChange(
          new ClusterStateObserver.Listener() {
            @Override
            public void onNewClusterState(ClusterState state) {
              latch.countDown();
            }

            @Override
            public void onClusterServiceClose() {
              latch.countDown();
            }

            @Override
            public void onTimeout(TimeValue timeout) {
              logger.warn(
                "timed out while waiting for initial discovery state - timeout: {}",
                initialStateTimeout
              );
              latch.countDown();
            }
          },
          state -> state.nodes().getMasterNodeId() != null,
          initialStateTimeout
        );

        try {
          latch.await();
        } catch (InterruptedException e) {
          throw new ElasticsearchTimeoutException(
            "Interrupted while waiting for initial discovery state"
          );
        }
      }
    }

    injector.getInstance(HttpServerTransport.class).start();

    if (WRITE_PORTS_FILE_SETTING.get(settings())) {
      TransportService transport = injector.getInstance(TransportService.class);
      writePortsFile("transport", transport.boundAddress());
      HttpServerTransport http = injector.getInstance(
        HttpServerTransport.class
      );
      writePortsFile("http", http.boundAddress());
    }

    logger.info("started");

    pluginsService
      .filterPlugins(ClusterPlugin.class)
      .forEach(ClusterPlugin::onNodeStarted);

    return this;
  }

总结

Node启动过程这种做的检查、初始化、加入集群都梳理清楚了，但节点加入集群后同步数据，在该部分没有找到。

这个后续在看集群管理的时候，再找一下这个问题的答案。

参考

https://www.modb.pro/db/33681
https://easyice.cn/archives/332

衣舞晨风

【Elasticsearch源码】节点启动分析

目的

源码分析

总结

参考