渣渣菜鸡的 ElasticSearch 源码解析 —— 启动流程

作者: zhisheng_blog | 来源:发表于2018-08-18 09:08 被阅读28次

005

关注我

zhisheng

转载请务必注明原创地址为：http://www.54tianzhisheng.cn/2018/08/11/es-code02/

前提

上篇文章写了 ElasticSearch 源码解析 —— 环境搭建，其中里面说了启动打开 server 模块下的 Elasticsearch 类：org.elasticsearch.bootstrap.Elasticsearch，运行里面的 main 函数就可以启动 ElasticSearch 了，这篇文章讲讲启动流程，因为篇幅会很多，所以分了两篇来写。

启动流程

main 方法入口

可以看到入口其实是一个 main 方法，方法里面先是检查权限，然后是一个错误日志监听器（确保在日志配置之前状态日志没有出现 error），然后是 Elasticsearch 对象的创建，然后调用了静态方法 main 方法（18 行），并把创建的对象和参数以及 Terminal 默认值传进去。静态的 main 方法里面调用 elasticsearch.main 方法。

public static void main(final String[] args) throws Exception {         //1、入口
    // 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)
    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();
    int status = main(args, elasticsearch, Terminal.DEFAULT); //2、调用Elasticsearch.main方法
    if (status != ExitCodes.OK) {
        exit(status);
    }
}

static int main(final String[] args, final Elasticsearch elasticsearch, final Terminal terminal) throws Exception {
    return elasticsearch.main(args, terminal);  //3、command main
}

1es01

因为 Elasticsearch 类是继承了 EnvironmentAwareCommand 类，EnvironmentAwareCommand 类继承了 Command 类，但是 Elasticsearch 类并没有重写 main 方法，所以上面调用的 elasticsearch.main 其实是调用了 Command 的 main 方法，代码如下：

/** Parses options for this command from args and executes it. */
public final int main(String[] args, Terminal terminal) throws Exception {
    if (addShutdownHook()) {                                                //利用Runtime.getRuntime().addShutdownHook方法加入一个Hook，在程序退出时触发该Hook
        shutdownHookThread = new Thread(() -> {
            try {
                this.close();
            } catch (final IOException e) {
                try (
                    StringWriter sw = new StringWriter();
                    PrintWriter pw = new PrintWriter(sw)) {
                    e.printStackTrace(pw);
                    terminal.println(sw.toString());
                } catch (final IOException impossible) {
                    // StringWriter#close declares a checked IOException from the Closeable interface but the Javadocs for StringWriter
                    // say that an exception here is impossible
                    throw new AssertionError(impossible);
                }
            }
        });
        Runtime.getRuntime().addShutdownHook(shutdownHookThread);
    }

    beforeMain.run();

    try {
        mainWithoutErrorHandling(args, terminal);//4、mainWithoutErrorHandling
    } catch (OptionException e) {
        printHelp(terminal);
        terminal.println(Terminal.Verbosity.SILENT, "ERROR: " + e.getMessage());
        return ExitCodes.USAGE;
    } catch (UserException e) {
        if (e.exitCode == ExitCodes.USAGE) {
            printHelp(terminal);
        }
        terminal.println(Terminal.Verbosity.SILENT, "ERROR: " + e.getMessage());
        return e.exitCode;
    }
    return ExitCodes.OK;
}

上面代码一开始利用一个勾子函数，在程序退出时触发该 Hook，该方法主要代码是 mainWithoutErrorHandling() 方法，然后下面的是 catch 住方法抛出的异常，方法代码如下：

/*** Executes the command, but all errors are thrown. */
void mainWithoutErrorHandling(String[] args, Terminal terminal) throws Exception {
    final OptionSet options = parser.parse(args);
    if (options.has(helpOption)) {
        printHelp(terminal);
        return;
    }
    if (options.has(silentOption)) {
        terminal.setVerbosity(Terminal.Verbosity.SILENT);
    } else if (options.has(verboseOption)) {
        terminal.setVerbosity(Terminal.Verbosity.VERBOSE);
    } else {
        terminal.setVerbosity(Terminal.Verbosity.NORMAL);
    }
    execute(terminal, options);//5、执行 EnvironmentAwareCommand 中的 execute()，（重写了command里面抽象的execute方法）
}

上面的代码从 3 ～ 14 行是解析传进来的参数并配置 terminal，重要的 execute() 方法，执行的是 EnvironmentAwareCommand 中的 execute() （重写了 Command 类里面的抽象 execute 方法），从上面那个继承图可以看到 EnvironmentAwareCommand 继承了 Command，重写的 execute 方法代码如下：

@Override
protected void execute(Terminal terminal, OptionSet options) throws Exception {
    final Map<String, String> settings = new HashMap<>();
    for (final KeyValuePair kvp : settingOption.values(options)) {
        if (kvp.value.isEmpty()) {
            throw new UserException(ExitCodes.USAGE, "setting [" + kvp.key + "] must not be empty");
        }
        if (settings.containsKey(kvp.key)) {
            final String message = String.format(
                Locale.ROOT, "setting [%s] already set, saw [%s] and [%s]",
                kvp.key, settings.get(kvp.key), kvp.value);
            throw new UserException(ExitCodes.USAGE, message);
        }
        settings.put(kvp.key, kvp.value);
    }
    //6、根据我们ide配置的 vm options 进行设置path.data、path.home、path.logs
    putSystemPropertyIfSettingIsMissing(settings, "path.data", "es.path.data");
    putSystemPropertyIfSettingIsMissing(settings, "path.home", "es.path.home");
    putSystemPropertyIfSettingIsMissing(settings, "path.logs", "es.path.logs");

    execute(terminal, options, createEnv(terminal, settings));//7、先调用 createEnv 创建环境
    //9、执行elasticsearch的execute方法，elasticsearch中重写了EnvironmentAwareCommand中的抽象execute方法
}

方法前面是根据传参去判断配置的，如果配置为空，就会直接跳到执行 putSystemPropertyIfSettingIsMissing 方法，这里会配置三个属性：path.data、path.home、path.logs 设置 es 的 data、home、logs 目录，它这里是根据我们 ide 配置的 vm options 进行设置的，这也是为什么我们上篇文章说的配置信息，如果不配置的话就会直接报错。下面看看 putSystemPropertyIfSettingIsMissing 方法代码里面怎么做到的：

/** Ensure the given setting exists, reading it from system properties if not already set. */
private static void putSystemPropertyIfSettingIsMissing(final Map<String, String> settings, final String setting, final String key) {
    final String value = System.getProperty(key);//获取key（es.path.data）找系统设置
    if (value != null) {
        if (settings.containsKey(setting)) {
            final String message =
                String.format(
                Locale.ROOT,
                "duplicate setting [%s] found via command-line [%s] and system property [%s]",
                setting, settings.get(setting), value);
            throw new IllegalArgumentException(message);
        } else {
            settings.put(setting, value);
        }
    }
}

执行这三个方法后：

2es02

跳出此方法，继续看会发现 execute 方法调用了方法，

 execute(terminal, options, createEnv(terminal, settings));

这里我们先看看 createEnv(terminal, settings) 方法：

protected Environment createEnv(final Terminal terminal, final Map<String, String> settings) throws UserException {
    final String esPathConf = System.getProperty("es.path.conf");//8、读取我们 vm options 中配置的 es.path.conf
    if (esPathConf == null) {
        throw new UserException(ExitCodes.CONFIG, "the system property [es.path.conf] must be set");
    }
    return InternalSettingsPreparer.prepareEnvironment(Settings.EMPTY, terminal, settings, getConfigPath(esPathConf));  //8、准备环境 prepareEnvironment
}

读取我们 ide vm options 中配置的 es.path.conf，同上篇文章也讲了这个一定要配置的，因为 es 启动的时候会加载我们的配置和一些插件。这里继续看下上面代码第 6 行的 prepareEnvironment 方法：

public static Environment prepareEnvironment(Settings input, Terminal terminal, Map<String, String> properties, Path configPath) {
    // just create enough settings to build the environment, to get the config dir
    Settings.Builder output = Settings.builder();
    initializeSettings(output, input, properties);
    Environment environment = new Environment(output.build(), configPath);

    //查看 es.path.conf 目录下的配置文件是不是 yml 格式的，如果不是则抛出一个异常
    if (Files.exists(environment.configFile().resolve("elasticsearch.yaml"))) {
        throw new SettingsException("elasticsearch.yaml was deprecated in 5.5.0 and must be renamed to elasticsearch.yml");
    }

    if (Files.exists(environment.configFile().resolve("elasticsearch.json"))) {
        throw new SettingsException("elasticsearch.json was deprecated in 5.5.0 and must be converted to elasticsearch.yml");
    }

    output = Settings.builder(); // start with a fresh output
    Path path = environment.configFile().resolve("elasticsearch.yml");
    if (Files.exists(path)) {
        try {
            output.loadFromPath(path);  //加载文件并读取配置文件内容
        } catch (IOException e) {
            throw new SettingsException("Failed to load settings from " + path.toString(), e);
        }
    }

    // re-initialize settings now that the config file has been loaded
    initializeSettings(output, input, properties);          //再一次初始化设置
    finalizeSettings(output, terminal);

    environment = new Environment(output.build(), configPath);

    // we put back the path.logs so we can use it in the logging configuration file
    output.put(Environment.PATH_LOGS_SETTING.getKey(), environment.logsFile().toAbsolutePath().normalize().toString());
    return new Environment(output.build(), configPath);
}

3es03

准备的环境如上图，通过构建的环境查看配置文件 elasticsearch.yml 是不是以 yml 结尾，如果是 yaml 或者 json 结尾的则抛出异常（在 5.5.0 版本其他两种格式过期了，只能使用 yml 格式），然后加载该配置文件并读取里面的内容（KV结构）。

跳出 createEnv 方法，我们继续看 execute 方法吧。

EnvironmentAwareCommand 类的 execute 方法代码如下：

protected abstract void execute(Terminal terminal, OptionSet options, Environment env) throws Exception;

这是个抽象方法，那么它的实现方法在 Elasticsearch 类中，代码如下：

@Override
protected void execute(Terminal terminal, OptionSet options, Environment env) throws UserException {
    if (options.nonOptionArguments().isEmpty() == false) {
        throw new UserException(ExitCodes.USAGE, "Positional arguments not allowed, found " + options.nonOptionArguments());
    }
    if (options.has(versionOption)) {
        final String versionOutput = String.format(
            Locale.ROOT,
            "Version: %s, Build: %s/%s/%s/%s, JVM: %s",
            Version.displayVersion(Version.CURRENT, Build.CURRENT.isSnapshot()),
            Build.CURRENT.flavor().displayName(),
            Build.CURRENT.type().displayName(),
            Build.CURRENT.shortHash(),
            Build.CURRENT.date(),
            JvmInfo.jvmInfo().version());
        terminal.println(versionOutput);
        return;
    }

    final boolean daemonize = options.has(daemonizeOption);
    final Path pidFile = pidfileOption.value(options);
    final boolean quiet = options.has(quietOption);

    // a misconfigured java.io.tmpdir can cause hard-to-diagnose problems later, so reject it immediately
    try {
        env.validateTmpFile();
    } catch (IOException e) {
        throw new UserException(ExitCodes.CONFIG, e.getMessage());
    }
    try {
        init(daemonize, pidFile, quiet, env);    //10、初始化
    } catch (NodeValidationException e) {
        throw new UserException(ExitCodes.CONFIG, e.getMessage());
    }
}

上面代码里主要还是看看 init(daemonize, pidFile, quiet, env); 初始化方法吧。

void init(final boolean daemonize, final Path pidFile, final boolean quiet, Environment initialEnv)
    throws NodeValidationException, UserException {
    try {
        Bootstrap.init(!daemonize, pidFile, quiet, initialEnv); //11、执行 Bootstrap 中的 init 方法
    } catch (BootstrapException | RuntimeException e) {
        // format exceptions to the console in a special way
        // to avoid 2MB stacktraces from guice, etc.
        throw new StartupException(e);
    }
}

init 方法

Bootstrap 中的静态 init 方法如下：

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();   //12、创建一个 Bootstrap 实例

    final SecureSettings keystore = loadSecureSettings(initialEnv);//如果注册了安全模块则将相关配置加载进来
    final Environment environment = createEnvironment(foreground, pidFile, keystore, initialEnv.settings(), initialEnv.configFile());   //干之前干过的事情
    try {
        LogConfigurator.configure(environment);   //13、log 配置环境
    } catch (IOException e) {
        throw new BootstrapException(e);
    }
    if (environment.pidFile() != null) {
        try {
            PidFile.create(environment.pidFile(), true);
        } catch (IOException e) {
            throw new BootstrapException(e);
        }
    }

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

        // fail if somebody replaced the lucene jars
        checkLucene();             //14、检查Lucene版本

// install the default uncaught exception handler; must be done before security is initialized as we do not want to grant the runtime permission setDefaultUncaughtExceptionHandler
        Thread.setDefaultUncaughtExceptionHandler(
            new ElasticsearchUncaughtExceptionHandler(() -> Node.NODE_NAME_SETTING.get(environment.settings())));

        INSTANCE.setup(true, environment);      //15、调用 setup 方法

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

        INSTANCE.start();         //26、调用 start 方法

        if (closeStandardStreams) {
            closeSysError();
        }
    } catch (NodeValidationException | RuntimeException e) {
        // disable console logging, so user does not see the exception twice (jvm will show it already)
        final Logger rootLogger = ESLoggerFactory.getRootLogger();
        final Appender maybeConsoleAppender = Loggers.findAppender(rootLogger, ConsoleAppender.class);
        if (foreground && maybeConsoleAppender != null) {
            Loggers.removeAppender(rootLogger, maybeConsoleAppender);
        }
        Logger logger = Loggers.getLogger(Bootstrap.class);
        if (INSTANCE.node != null) {
            logger = Loggers.getLogger(Bootstrap.class, Node.NODE_NAME_SETTING.get(INSTANCE.node.settings()));
        }
        // 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;
    }
}

该方法主要有：

1、创建 Bootstrap 实例

2、如果注册了安全模块则将相关配置加载进来

3、创建 Elasticsearch 运行的必须环境以及相关配置, 如将 config、scripts、plugins、modules、logs、lib、bin 等配置目录加载到运行环境中

4、log 配置环境，创建日志上下文

5、检查是否存在 PID 文件，如果不存在，创建 PID 文件

6、检查 Lucene 版本

7、调用 setup 方法（用当前环境来创建一个节点）

setup 方法

private void setup(boolean addShutdownHook, Environment environment) throws BootstrapException {
    Settings settings = environment.settings();//根据环境得到配置
    try {
        spawner.spawnNativeControllers(environment);
    } catch (IOException e) {
        throw new BootstrapException(e);
    }
    initializeNatives(
        environment.tmpFile(),
        BootstrapSettings.MEMORY_LOCK_SETTING.get(settings),
        BootstrapSettings.SYSTEM_CALL_FILTER_SETTING.get(settings),
        BootstrapSettings.CTRLHANDLER_SETTING.get(settings));
    // initialize probes before the security manager is installed
    initializeProbes();
    if (addShutdownHook) {
        Runtime.getRuntime().addShutdownHook(new Thread() {
            @Override
            public void run() {
                try {
                    IOUtils.close(node, spawner);
                    LoggerContext context = (LoggerContext) LogManager.getContext(false);
                    Configurator.shutdown(context);
                } catch (IOException ex) {
                    throw new ElasticsearchException("failed to stop node", ex);
                }
            }
        });
    }
    try {
        // look for jar hell
        final Logger logger = ESLoggerFactory.getLogger(JarHell.class);
        JarHell.checkJarHell(logger::debug);
    } catch (IOException | URISyntaxException e) {
        throw new BootstrapException(e);
    }
    // Log ifconfig output before SecurityManager is installed
    IfConfig.logIfNecessary();
    // install SM after natives, shutdown hooks, etc.
    try {
        Security.configure(environment, BootstrapSettings.SECURITY_FILTER_BAD_DEFAULTS_SETTING.get(settings));
    } catch (IOException | NoSuchAlgorithmException e) {
        throw new BootstrapException(e);
    }
    node = new Node(environment) {              //16、新建节点
        @Override
        protected void validateNodeBeforeAcceptingRequests(
            final BootstrapContext context,
            final BoundTransportAddress boundTransportAddress, List<BootstrapCheck> checks) throws NodeValidationException {
            BootstrapChecks.check(context, boundTransportAddress, checks);
        }
    };
}

上面代码最后就是 Node 节点的创建，这篇文章就不讲 Node 的创建了，下篇文章会好好讲一下 Node 节点的创建和正式启动 ES 节点的。