Elasticsearch学习总结五 elasticSearch插件机制源码解析

摘要: es提供了很多插件，开发者可以灵活的使用各种插件方便的监控和操作es，下面我们来看看es源码是如何工作的

一.首先看看Elasticsearch中插件是如何安装的

我们安装好es后，如果要安装插件需要在 /usr/local/elasticsearch/bin的目录下使用plugin这个shell脚本，仔细看了下这个shell脚本，发现里面主要是运行了org.elasticsearch.plugins.PluginManager这个类
exec "$JAVA" $JAVA_OPTS $ES_JAVA_OPTS -Xmx64m -Xms16m -Delasticsearch -Des.path.home="$ES_HOME" $properties -cp "$ES_HOME/lib/*" org.elasticsearch.plugins.PluginManager $args

下面我们进入PluginManager这个类中看看是如何下载安装插件的，如何执行这个类，肯定应该有个main方法吧，果真如此，下面是pluginManager的main方法

public static void main(String[] args) {
..........
String url = null;
OutputMode outputMode = OutputMode.DEFAULT;
String pluginName = null;
TimeValue timeout = DEFAULT_TIMEOUT;
int action = ACTION.NONE;
if (args.length < 1) {
displayHelp(null);
}
try {
for (int c = 0; c < args.length; c++) {
String command = args[c];
switch (command) {
case "-u":
case "--url":
// deprecated versions:
case "url":
case "-url":
url = getCommandValue(args, ++c, "--url");
// Until update is supported, then supplying a URL implies installing
// By specifying this action, we also avoid silently failing without
//  dubious checks.
action = ACTION.INSTALL;
break;
case "-v":
case "--verbose":
// deprecated versions:
case "verbose":
case "-verbose":
outputMode = OutputMode.VERBOSE;
break;
case "-s":
case "--silent":
// deprecated versions:
case "silent":
case "-silent":
outputMode = OutputMode.SILENT;
break;
case "-i":
case "--install":
// deprecated versions:
case "install":
case "-install":
pluginName = getCommandValue(args, ++c, "--install");
action = ACTION.INSTALL;
break;
case "-r":
case "--remove":
// deprecated versions:
case "remove":
case "-remove":
pluginName = getCommandValue(args, ++c, "--remove");
action = ACTION.REMOVE;
break;
case "-t":
case "--timeout":
// deprecated versions:
case "timeout":
case "-timeout":
String timeoutValue = getCommandValue(args, ++c, "--timeout");
timeout = TimeValue.parseTimeValue(timeoutValue, DEFAULT_TIMEOUT);
break;
case "-l":
case "--list":
action = ACTION.LIST;
break;
case "-h":
case "--help":
displayHelp(null);
break;
default:
displayHelp("Command [" + command + "] unknown.");
// Unknown command. We break...
System.exit(EXIT_CODE_CMD_USAGE);
}
}
} catch (Throwable e) {
displayHelp("Error while parsing options: " + e.getClass().getSimpleName() +
": " + e.getMessage());
System.exit(EXIT_CODE_CMD_USAGE);
}

上面代码中，就是当你输入各种命令的时候，pluginManager都会根据响应的命令触发不同的动作，例如输入./plugin list 就会显示当前es安装了那些插件



当输入 plugin -install mobz/elasticsearch-head 命令的时候，pluginManager就会去获取当前插件名称， pluginName = getCommandValue(args, ++c, "--install"); 并且将 action赋值 action = ACTION.INSTALL;
当完成赋值后会继续执行如下代码，这部分代码主要构造PluginManger对象，并且根据action的动作去执行相应的操作，pluginManager.downloadAndExtract(pluginName);会根据插件名称去下载相应的插件

................
if (action > ACTION.NONE) {
int exitCode = EXIT_CODE_ERROR; // we fail unless it's reset
//构造pluginManger
PluginManager pluginManager = new PluginManager(initialSettings.v2(), url, outputMode, timeout);
switch (action) {
case ACTION.INSTALL:
try {
pluginManager.log("-> Installing " + Strings.nullToEmpty(pluginName) + "...");
pluginManager.downloadAndExtract(pluginName);//执行下载解压任务
exitCode = EXIT_CODE_OK;
} catch (IOException e) {
exitCode = EXIT_CODE_IO_ERROR;
pluginManager.log("Failed to install " + pluginName + ", reason: " + e.getMessage());
} catch (Throwable e) {
exitCode = EXIT_CODE_ERROR;
displayHelp("Error while installing plugin, reason: " + e.getClass().getSimpleName() +
": " + e.getMessage());
}
break;
.........................

PluginManager的构造函数主要是传入下载地址，延时时长等，下面看下downloadAndExtract下载安装是如何工作的

public void downloadAndExtract(String name) throws IOException {
.....
HttpDownloadHelper downloadHelper = new HttpDownloadHelper();
boolean downloaded = false;
HttpDownloadHelper.DownloadProgress progress;
if (outputMode == OutputMode.SILENT) {
progress = new HttpDownloadHelper.NullProgress();
} else {
progress = new HttpDownloadHelper.VerboseProgress(System.out);
}
.........
if (url != null) {
URL pluginUrl = new URL(url);
log("Trying " + pluginUrl.toExternalForm() + "...");
try {
downloaded = true;
downloadHelper.download(pluginUrl, pluginFile, progress, this.timeout);
} catch (ElasticsearchTimeoutException e) {
throw e;
} catch (Exception e) {
// ignore
log("Failed: " + ExceptionsHelper.detailedMessage(e));
}
}

主要是通过构建HttpDownloadHelper 实现http下载，并且实现了一个下载进度器，这就是我们安装插件的时候会有一些输出的原因，通过downloadHelper.download(pluginUrl, pluginFile, progress, this.timeout);可以真正的实现下载，这里就不在赘述了，当下载完毕后，会去执行解压和安装的过程，在解压文件时候越过父文件夹，将文件解压到指定目录，比如head文件夹中，至此下载安装基本完毕

ZipFile zipFile = null;
try {
zipFile = new ZipFile(pluginFile);
boolean removeTopLevelDir = topLevelDirInExcess(zipFile);
Enumeration<? extends ZipEntry> zipEntries = zipFile.entries();
while (zipEntries.hasMoreElements()) {
ZipEntry zipEntry = zipEntries.nextElement();
if (zipEntry.isDirectory()) {
continue;
}
String zipEntryName = zipEntry.getName().replace('\\', '/');
if (removeTopLevelDir) {
zipEntryName = zipEntryName.substring(zipEntryName.indexOf('/'));
}
File target = new File(extractLocation, zipEntryName);
FileSystemUtils.mkdirs(target.getParentFile());
Streams.copy(zipFile.getInputStream(zipEntry), new FileOutputStream(target));
}
log("Installed " + name + " into " + extractLocation.getAbsolutePath());
} catch (Exception e) {
log("failed to extract plugin [" + pluginFile + "]: " + ExceptionsHelper.detailedMessage(e));
return;
} finally {
if (zipFile != null) {
try {
zipFile.close();
} catch (IOException e) {
// ignore
}
}
pluginFile.delete();
}

二. elasticSearch是如何加载插件的

elasticsearch里面的组件基本都是用上面的方式进行模块化管理，elasticsearch对guice进行了简单的封装，通过ModulesBuilder类构建es的模块，一个es节点肯定够包括PluginsModule插件模块，下面我们看下插件模块是如何工作，加载我们的插件的

public class PluginsModule extends AbstractModule implements SpawnModules, PreProcessModule {

private final Settings settings;

private final PluginsService pluginsService;

public PluginsModule(Settings settings, PluginsService pluginsService) {
this.settings = settings;
this.pluginsService = pluginsService;
}

//预处理组件
public Iterable<? extends Module> spawnModules() {
List<Module> modules = Lists.newArrayList();
Collection<Class<? extends Module>> modulesClasses = pluginsService.modules();
for (Class<? extends Module> moduleClass : modulesClasses) {
modules.add(createModule(moduleClass, settings));
}
modules.addAll(pluginsService.modules(settings));
return modules;
}
..............

es中所有的组件都是继承自AbstractModule
插件调用层次为Modules->PluginsModule->PluginsService->XXPlugin；
都是调用void processModule(Module module)方法。
区别在于
1：PluginsModule实现的是PreProcessModule接口。
2：PluginsService是自己本来的方法。
3：XXPlugin实现的是Plugin接口。
虽然他们都有processModule方法，但不是实现的同一个类。最重要是的还是pluginService类，它真正负责加载的插件类

public PluginsService(Settings settings, Environment environment) {
super(settings);
this.environment = environment;
this.checkLucene = componentSettings.getAsBoolean("check_lucene", true);
.................................忽律部分代码
loadPluginsIntoClassLoader();
if (loadClasspathPlugins) {
tupleBuilder.addAll(loadPluginsFromClasspath(settings));
}
this.plugins = tupleBuilder.build();

// we load site plugins
ImmutableList<Tuple<PluginInfo, Plugin>> tuples = loadSitePlugins();
for (Tuple<PluginInfo, Plugin> tuple : tuples) {
sitePlugins.add(tuple.v1().getName());
}
.................................忽律部分代码

其中loadPluginsIntoClassLoader();负责classloader加载class,加载不在当前classpath中的class，使用当前setting中的classloader,这个构造函数负责加载plugin下面的所有的插件，并且使用当前的类加载器加载。

至此，插件的下载和加载全部完成。