利用zookeeper监控flume进程报警

利用flume进行日志采集已在业界广泛使用,如果需要采集的log比较多,并且分布在多个服务器上,那么对flume agent的管理就是一个头疼的事情。
本篇主要介绍下利用zookeeper对flume进程进行监控,当flume异常退出时进行报警,以至于我们能够及时的发现故障解决故障。

思路很简单,其实开发起来也很简单,嘿嘿。。
主要是利用了zookeeper临时节点的功能,当flume agent启动时与zk建立连接创建一个临时节点,并周期的与zk进行心跳,当flume发生异常退出程序时,与zk的心跳失败,导致临时节点被zk删除,监听到这个事件之后就可以进行报警,达到监控的目的。

是不是很简单,但是如果只就这样去开发上线是有个bug的,当你正常退出flume进程时,zk上的临时节点也会被删除,同样会监听到此事件,也会报警。其实我们应该给每个flume agent在zk上创建一个临时节点的同时也应该创建一个永久性节点,这个永久性节点用来标识此agent的信息,当agent异常退出时,临时性节点会被删除但永久性节点不会被删除,这样当watcher监听到临时节点被删除,并且检测到永久性节点并没有被删除,那就可以断定agent是异常退出,需要报警,当agent正常退出时,会将永久性节点删除,同样临时节点也会被删除,这样watcher监听到临时节点被删除去检测永久性节点时,发现也被删除了,就可以断定此agent是正常退出,不需要报警。

此思路借鉴了HDFS HA机制,之所以利用zookeeper来进行监控是因为我们是将flume采集的log写入kafka,这样zookeeper已是我们的组件之一,重复利用现有的组件可以减少我们的维护量和开发量,并且zookeeper避免单点故障,也不用我们自己开发心跳模块和监听模块,开发量很小。

扯了那么多来点干货吧。

我没有直接在agent的启动流程中添加与zk的连接,感觉这样对flume的侵入性太大了,而是引入了watchdog,在watchdog中添加相关代码。

watchdog 是老版flume中的组件,用来监控agent进程,当agent异常退出之后进行重启。watchdog的具体代码可以参考github,我们在此基础上进行了修改。

在watchdog的startup中与zk建立连接,zk client用的是Curator
代码如下:

1
2
3
4
5
6
7
8
9
10
void startup() throws Exception {
...
// 与zk建立连接
CuratorFramework client = CuratorFrameworkFactory.newClient(ZK_ADDRESS, new ExponentialBackoffRetry(1000, 3));
client.start();
// 创建永久节点
client.create().withMode(CreateMode.PERSISTENT).forPath(PATH_P + "/test", new byte[0]);
// 创建临时节点
client.create().withMode(CreateMode.EPHEMERAL).forPath(PATH + "/test", new byte[0]);
}

agent端的代码就算结束了,我们需要再起一个服务来监听zk上节点的信息。

监听服务我使用的是Curator的PathCache来监听一个目录的所有子节点的delete事件。此服务在与zk建立连接时注册一个watcher,代码是在PathCache的example的基础上进行了修改,代码如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
public static void start() throws Exception
{
CuratorFramework client = null;
PathChildrenCache cache = null;
try
{
client = CuratorFrameworkFactory.newClient(ZK_ADDRESS, new ExponentialBackoffRetry(1000, 3));
client.start();
// path cache 类似 zk中的watcher
// in this example we will cache data. Notice that this is optional.
cache = new PathChildrenCache(client, PATH, true);
cache.start();
// 给cache注册监听器,对相应事件进行处理
addListener(cache);
}
finally
{
CloseableUtils.closeQuietly(cache);
CloseableUtils.closeQuietly(client);
}
}
private static void addListener(PathChildrenCache cache)
{
// a PathChildrenCacheListener is optional. Here, it's used just to log changes
PathChildrenCacheListener listener = new PathChildrenCacheListener()
{
@Override
public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception
{
switch ( event.getType() )
{
case CHILD_ADDED:
{
System.out.println("Node added: " + ZKPaths.getNodeFromPath(event.getData().getPath()));
break;
}
case CHILD_UPDATED:
{
System.out.println("Node changed: " + ZKPaths.getNodeFromPath(event.getData().getPath()));
break;
}
case CHILD_REMOVED:
{
System.out.println("判断永久节点是否存在,决定是否报警");
System.out.println("Node removed: " + ZKPaths.getNodeFromPath(event.getData().getPath()));
break;
}
}
}
};
cache.getListenable().addListener(listener);
}

这个服务比较简单,就一个类,打算随后将其集成到自动化部署平台中,上面只是伪代码,详细代码以后有时间再不全。

您的肯定,是我装逼的最大的动力!