Elasticsearch使用syslog发送Watcher告警事件

Elasticsearch从5.0版本开始，把Watcher插件整合到了官方的X-Pack插件中。Watcher是Elasticsearch内置的计划任务管理器，定期执行某些脚本或查询语句，将结果用于告警。具体介绍见Alerting on Cluster and Index Events。

可但是、但可是，Watcher的告警模块十分蛋疼，设计了一堆Email、HipChat、Slack、PagerDuty、Jira等乱七八糟的告警方式，却连最基本的syslog、FTP、TCP等告警方式都没有！内置的Logging告警也只是把告警信息和Elasticsearch的其他日志写到同一个文件中，无法分离出来。

所以只能靠自己了！我们可以自己编写一个告警action，并加入到X-Pack中。这里我们以5.2.0版本为例。

什么！你没有license用不了Watcher？请戳X-Pack破解试用。那个……如果破解搞不定的话下面也不用看了。

0. 开发前准备

首先我们需要将X-Pack-5.2.0.zip下载下来，找到X-Pack-5.2.0.jar，用Luyten进行反编译。我们要用到的代码在

org.elasticsearch.xpack.watcher.actions

下，不会写的话可以参考其他action。

完整代码见GitHub上的actionssyslog，这是一个IntelliJ IDEA的工程。编译时需要将X-Pack-5.2.0.jar加入附加依赖包：File->Project Structure->Libraries->”+”->Java->/path/to/X-Pack-5.2.0.jar。

1. 代码

1.1 Watcher.java和ActionBuilders.java

这两个是注册action的类，打开可以看到里面注册了各种action。把我们的SyslogAction也加进去：

import org.elasticsearch.xpack.watcher.actions.syslog.*;
...
public class Watcher implements ActionPlugin, ScriptPlugin
{
...
public Collection<Object> createComponents(final Clock clock, final ScriptService scriptService, final InternalClient internalClient, final SearchRequestParsers searchRequestParsers, final XPackLicenseState licenseState, final HttpClient httpClient, final NamedXContentRegistry xContentRegistry, final Collection<Object> components) {
...
actionFactoryMap.put("slack", new SlackActionFactory(this.settings, templateEngine, this.getService(SlackService.class, components)));
actionFactoryMap.put("pagerduty", new PagerDutyActionFactory(this.settings, templateEngine, this.getService(PagerDutyService.class, components)));
actionFactoryMap.put("syslog", new SyslogActionFactory(this.settings, templateEngine));
...
}
...
}

以及：

package org.elasticsearch.xpack.watcher.actions;
...
public final class ActionBuilders
{
...
public static PagerDutyAction.Builder pagerDutyAction(final IncidentEvent.Template event) {
return PagerDutyAction.builder(event);
}
public static SyslogAction.Builder syslogAction(final String app, final String host, final int port , final String facility, final String level, final TextTemplate text) {
return SyslogAction.builder(app, host, port, facility, level, text);
}
}

1.2 SyslogActionFactory.java

Watcher.java中调用了SyslogActionFactory，这个工厂类相对简单，SyslogAction.parser解析我们XPUT到Elasticsearch的参数来生成SycloAction对象，然后交给ExecutableSyslogAction的类去执行action。

package org.elasticsearch.xpack.watcher.actions.syslog;
...
public class SyslogActionFactory extends ActionFactory {
private final TextTemplateEngine templateEngine;

public SyslogActionFactory(final Settings settings, final TextTemplateEngine templateEngine) {
super(Loggers.getLogger(ExecutableSyslogAction.class, settings, new String[0]));
this.templateEngine = templateEngine;
}

@Override
public ExecutableSyslogAction parseExecutable(final String watchId, final String actionId, final XContentParser parser) throws IOException {
return new ExecutableSyslogAction(SyslogAction.parse(watchId, actionId, parser), this.actionLogger, this.templateEngine);
}
}

1.3 ExecutableSyslogAction.java

这个类也相对简单，在

execute

方法中发送告警信息。

package org.elasticsearch.xpack.watcher.actions.syslog;
...
public class ExecutableSyslogAction extends ExecutableAction<SyslogAction> {
private final TextTemplateEngine engine;

public ExecutableSyslogAction(final SyslogAction action, final Logger logger, final TextTemplateEngine templateEngine) {
super(action, logger);
this.engine = templateEngine;
}

@Override
public Action.Result execute(final String actionId, final WatchExecutionContext ctx, final Payload payload) throws Exception {
final Map<String, Object> model = Variables.createCtxModel(ctx, payload);
final String loggedText = engine.render(action.text, model);
if (ctx.simulateAction(actionId)) {
return new SyslogAction.Result.Simulated(loggedText);
}
action.send(loggedText);
return new SyslogAction.Result.Success(loggedText);
}
}

1.4 SyslogAction.java

这个类实现了Action接口，内容比较多，就不贴代码了。比较重要的方法包括：

parse：从Watcher的CreateAPI中的参数构建出SyslogAction类；

Results：向Watcher反馈每次action执行情况的接口；

toXContent：将SyslogAction转为Elasticsearch文档；

send：这个不是Action的接口方法，可以自定义名称及行为，每个action其实都有一个类似的方法，具体执行自己的action操作。

这里尤其需要注意的是权限问题。Elasticsearch启用了Java安全机制，所以我们在安装X-Pack时会看到类似以下提示：

@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@

@ WARNING: plugin requires additional permissions @

@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@

* java.lang.RuntimePermission accessClassInPackage.com.sun.activation.registries

* java.lang.RuntimePermission getClassLoader

* java.lang.RuntimePermission setContextClassLoader

* java.lang.RuntimePermission setFactory

* java.security.SecurityPermission createPolicy.JavaPolicy

* java.security.SecurityPermission getPolicy

* java.security.SecurityPermission putProviderProperty.BC

* java.security.SecurityPermission setPolicy

* java.util.PropertyPermission * read,write

* java.util.PropertyPermission sun.nio.ch.bugLevel write

* javax.net.ssl.SSLPermission setHostnameVerifier

See http://docs.oracle.com/javase/8/docs/technotes/guides/security/permissions.html

for descriptions of what these permissions allow and the associated risks.

Continue with installation? [y/N]y

可见，插件中的很多行为都需要专门进行授权并确认。如果不进行专门授权，我们的SyslogAction甚至没有绑定端口和发送数据的权限！详见官方说明Help for plugin authors。因此才会有了这么一段代码：

public void send(final String loggedText) throws Exception {
UdpSyslogMessageSender sender = AccessController.doPrivileged(new PrivilegedExceptionAction<UdpSyslogMessageSender>() {
public UdpSyslogMessageSender run() throws Exception {
return new UdpSyslogMessageSender();
}
});
...
}

2. 部署

我们在X-Pack破解试用提到过，将X-Pack-5.2.0.jar作为依赖包，就无需编译整个X-Pack。因此这里编译只生成了SyslogAction相关的几个class文件。

用压缩文件管理器打开X-Pack-5.2.0.jar，将生产成的class文件按路径加入到jar包中，ActionBuilders.class和Watcher.class需要覆盖原有的文件。

将打包好的X-Pack-5.2.0.jar连同依赖的java-syslog-client-1.0.8.jar，一起拷贝到X-Pack插件的根目录，例如：/usr/share/elasticsearch/plugins/x-pack/。

修改该目录下的plugin-security.policy文件，增加一行：

permission java.net.SocketPermission "localhost:0", "listen,resolve";

所有Master节点上都需要部署我们定制的X-Pack-5.2.0.jar，Data节点不需要。

最后重启Elasticsearch集群即可。

3. 测试

我们新建一个每10秒触发一次的action作为测试：

PUT _xpack/watcher/watch/syslog_example
{
"trigger" : {
"schedule" : { "interval" : "10s" }
},
"actions" : {
"hello_syslog" : {
"syslog" : {
"app" : "elastic"
"host" : "127.0.0.1",
"port" : 514,
"facility" : "local7",
"level" : "warning",
"text" : "executed at {{ctx.execution_time}}"
}
}
}
}

说明：

这个例子中”syslog”除text外各项参数的值就是默认值；

“level”的可选值包括EMERGENCY、ALERT、CRITICAL、ERROR、NOTICE、INFORMATIONAL、DEBUG；

text不能为空，可以引用Elasticsearch事件中的字段。

在接收Syslog的服务器上（这里就是本机）tcpdump一下就能看到事件：

tcpdump -i lo udp port 514

最后，如果想偷懒的话，我有一个打包好的版本：X-Pack-5.2.0-mvpboss1004.jar。