基于Hadoo的日志收集框架---Chukwa的源码分析(适配器、代理)
1. 接口、实现类简介
(1)Chukwa使用适配器(Adaptor)实现对各种输入(Streaming、Log File)的监控
??? org.apache.hadoop.chukwa.datacollection.adaptor.Adaptor
??? 适配器接口
// 启动适配器void start(String adaptorID, String type, long offset, ChunkReceiver dest)// 返回适配器状态String getCurrentStatus()// 返回适配器类型String getType()// 参数处理，返回流的名称String parseArgs(String datatype, String params, AdaptorManager c)// 关闭适配器，关闭之前应推送所有数据long shutdown(AdaptorShutdownPolicy shutdownPolicy)
?
??? org.apache.hadoop.chukwa.datacollection.adaptor.AbstractAdaptor
??? 适配器的实现类，是一个抽象类，只实现了适配器接口中的start方法和parseArgs方法
// 只是做了对属性的赋值操作，启动方法放在了重载的start(long offset)方法中void start(String adaptorID, String type, long offset, ChunkReceiver dest)// 抽象方法start由AbstractAdaptor的子类实现适配器的启动过程abstract void start(long offset)// 同样的方式实现了parseArgs方法String parseArgs(String d, String s, AdaptorManager c)abstract String parseArgs(String s)// 使用AdaptorManager实现Adaptor的注销和停止void deregisterAndStop()
?
??? org.apache.hadoop.chukwa.datacollection.adaptor.filetailer.LWFTAdaptor
??? 继承自AbstractAdaptor，实现了使用tail的方式监测文件增量内容的功能
// 启动一个文件监测线程(FileTailer)，并设置当前读取位置(fileReadOffset)void start(long offset)// 停止文件监测线程(FileTailer),并返回已经读取的长度long shutdown(AdaptorShutdownPolicy shutdownPolicy)// 监测文件增量内容synchronized boolean tailFile()// 从字节数组中提取数据extractRecords(ChunkReceiver eq, long buffOffsetInFile, byte[] buf)
?
??? org.apache.hadoop.chukwa.datacollection.adaptor.filetailer.FileTailingAdaptor
??? 继承自LWFTAdaptor
// 重写了停止的方法long shutdown(AdaptorShutdownPolicy shutdownPolicy)// 重写了监测文件增量内容的方法synchronized boolean tailFile()
?
??? org.apache.hadoop.chukwa.datacollection.adaptor.filetailer.CharFileTailingAdaptorUTF8
??? 继承自FileTailingAdaptor
// 重写了提取数据的方法int extractRecords(ChunkReceiver eq, long buffOffsetInFile, byte[] buf)
?
(2)Chukwa使用代理(Agent)实现对各种适配器(Adaptor)的管理
??? org.apache.hadoop.chukwa.datacollection.agent.AdaptorManager
??? 适配器管理者接口
// 返回属于此管理者的适配器数量int adaptorCount()// 根据ID停止某个适配器long stopAdaptor(String id, AdaptorShutdownPolicy mode)// 根据ID获取适配器对象Adaptor getAdaptor(String id)// 命令处理String processAddCommand(String cmd)// 以map形式返回适配器列表Map<String, String> getAdaptorList()// 提交报告String reportCommit(Adaptor src, long uuid)
?
??? org.apache.hadoop.chukwa.datacollection.agent.ChukwaAgent
??? 适配器管理者的实现类
?
2. 启动、处理流程
(1)代理的启动流程
??? org.apache.hadoop.chukwa.datacollection.agent.ChukwaAgent
// 存储adaptor及其偏移量Map<Adaptor, Offset> adaptorPositions// 存储adaptor及其IDMap<String, Adaptor> adaptorsByName// 记录恢复的次数int checkpointNumber// 是否启用checkpoint机制boolean DO_CHECKPOINT_RESTORE// checkpoint文件前缀名String CHECKPOINT_BASE_NAME// checkpoint存储目录File checkpointDir/** * 1. 创建守护进程"Agent",将进程号写入相应的pid文件 * 2. 校验参数，如果参数为"-help",则显示帮助信息后退出 * 3. 根据系统设置(CHUKWA_HOME、CHUKWA_CONF_DIR)加载配置文件(chukwa-agent-conf.xml)，并根据配置文件新建ChukwaAgent实例 * 4. 检测是否有另外一个Agent在运行，如果有则退出 * 5. 启动Connector,如果第一个参数中为"local"，则启动ConsoleOutConnector，否则启动HttpConnector */public static void main(String[] args)/** * 根据配置文件初始化ChukwaAgent * 1. 从配置文件中加载配置项 * chukwaAgent.checkpoint.enabled 是否启动checkpoint机制 * chukwaAgent.checkpoint.namecheckpoint文件前缀名 * chukwaAgent.checkpoint.intervalcheckpoint间隔时间 * 2. 如果启动了checkpoint机制则从checkpoint文件恢复启动adaptor * 如果initialAdaptors不为空且存在，则执行文件中的加载adaptors的命令 * 3. 获取AgentControlSocketListener实例，将服务绑定在127.0.0.1:9093上 * 4. 如果设置了checkpoint间隔时间且checkpoint目录存在，则启动定时器，以间隔一定时间将adaptors信息写入到checkpoint文件 */public ChukwaAgent(Configuration conf)/** * 1. 遍历存储checkpoint文件的目录，查找以CHECKPOINT_BASE_NAME为前缀的文件 * 2. 截取文件名中的数字，对其加一做为下一次checkpoint的编号 * 3. 从文件中加载启动adaptor */private boolean restoreFromCheckpoint()/** * 1. 逐行读取文件内容并执行 */private void readAdaptorsFile(File checkpoint)/** * 1. 使用正则表达式匹配命令是否符合如下格式 * "add [name =] <adaptor_class_name> <datatype> <adaptor specific params> <initial offset>" * [name = ]为可选参数，名称中不能有空格或'=', 要以'adaptor_'开头 * checkpoint文件示例： * ADD adaptor_90ac9ec5db2de67ebd7ad5a5ec260d26 = org.apache.hadoop.chukwa.datacollection.adaptor.filetailer.CharFileTailingAdaptorUTF8 TestLog1 0 /home/matrix/Program/project/log/testlog1 57810 * initial_adaptors文件示例： * add filetailer.CharFileTailingAdaptorUTF8 TestLog1 0 /home/matrix/Program/project/log/testlog1 0 * 2. 分别获取adaptorID、adaptorClassName、dataType、params、offset * 3. 根据adaptorClassName由适配器工厂得到适配器对象 * 4. 根据dataType、params获取流名称，并由此和adaptorClassName、dataType构建adaptorID(如果adaptorID为null) * 以adaptorClassName、dataType、params作为摘要信息完成哈希计算，并将结果以十六进制无符号整数值的字符串形式做为adaptorID * 5. 如果adaptorsByName中已有此adaptorID为key的对象则直接返回adaptorID * 否则将其添加到adaptorsByName、adaptorPositions，然后启动此adaptor，返回adaptorID */public String processAddCommandE(String cmd)/** * 停止代理的方法 * 1. 停止AgentControlSocketListener，不再处理telnet连接请求 * 2. 如果启动了checkpoint定时器则将其停止并将最后一次文件增量信息写入checkpoint文件 * 3. 遍历adaptorsByName，依次停止adaptor */public void shutdown(boolean exit)/** * 将adaptor信息写入checkpoint文件 * 1. 遍历adaptorsByName，格式化adaptor信息 * 2. 将adaptor信息写入新的checkpoint文件 * 3. 删除旧的checkpoint文件并将执行次数加1 */private void writeCheckpoint()
?
(2)代理控制线程
??? org.apache.hadoop.chukwa.datacollection.agent.AgentControlSocketListener
??? 此线程提供了可通过telnet连接到agent进行控制的功能
/** * 启动一个线程，循环监听连接请求，有请求到达则使用内部类ListenThread处理 * 默认telnet端口为9093，可通过"chukwaAgent.control.port"设置 */public void run()
??? org.apache.hadoop.chukwa.datacollection.agent.AgentControlSocketListener$ListenThread
??? 用于处理用户对agent的操作命令
/** * 构造器，设置超时时间为60秒 */ListenThread(Socket conn)/** * 接收输入，并调用processCommand()方法处理命令 */public void run()/** * 处理命令,支持如下操作: * add [adaptorname] [args] [offset] -- start an adaptor * shutdown [adaptornumber] -- graceful stop * stop [adaptornumber] -- abrupt stop * list -- list running adaptors * close -- close this connection * stopagent -- stop the whole agent process * stopall -- stop all adaptors * reloadCollectors -- reload the list of collectors * help -- print this message */public void processCommand(String cmd, PrintStream out)?
?
?
(3)适配器的启动流程
??? org.apache.hadoop.chukwa.datacollection.adaptor.filetailer.CharFileTailingAdaptorUTF8
??? 以此类为例:
/** * 1. 此方法继承自org.apache.hadoop.chukwa.datacollection.adaptor.AbstractAdaptor * 在AbstractAdaptor中的实现只是做了赋值操作，具体实现交给了重载的方法start(long offset) */void start(String adaptorID, String type, long offset, ChunkReceiver dest)/** * 1. 此方法继承自org.apache.hadoop.chukwa.datacollection.adaptor.filetailer.FileTailingAdaptor * 在FileTailingAdaptor中的实现又调用了其父类的实现，启动了一个FileTailer实例 */public void start(long offset) /** * 1. 此方法继承自org.apache.hadoop.chukwa.datacollection.adaptor.filetailer.FileTailingAdaptor * 2. 检测被监测的文件是否存在、是否可读 * 3. 获取被监测的文件长度，与adaptor初始化时设置的已读文件长度(fileReadOffset)做比较 * 4. 如果存在增量内容则调用slurp(long len, RandomAccessFile reader)方法 */public synchronized boolean tailFile()/** * 1. 此方法继承自org.apache.hadoop.chukwa.datacollection.adaptor.filetailer.LWFTAdaptor * 2. 获取已读文件长度和现有文件长度的差值,即增量内容大小，判断是否已超过最大读取大小(MAX_READ_SIZE) * MAX_READ_SIZE 默认为 128*1024，可通过"chukwaAgent.fileTailingAdaptor.maxReadSize"设置 * 3. 新建一个字节数组，长度即为增量内容的长度，如果超过了最大值则设为最大值 * 用此数组在reader(监测文件)中以fileReadOffset(已读长度)为起点读取数据 * 4. 调用extractRecords(ChunkReceiver eq, long buffOffsetInFile, byte[] buf)方法提取数据 */protected boolean slurp(long len, RandomAccessFile reader)/** * 1. 遍历buf，记录换行符的位置，获取最后一个换行符的位置 * 2. 使用数据类型、监测文件路径、文件偏移量、字节数组、adaptor构建ChunkImpl对象，添加到ChunkReceiver中 */protected int extractRecords(ChunkReceiver eq, long buffOffsetInFile, byte[] buf)
?
(4)文件监测线程
??? org.apache.hadoop.chukwa.datacollection.adaptor.filetailer.FileTailer
??? 共享线程(针对一个agent服务器上的所有adaptor)
// 存储adaptor，因为此列表中的数据大多在启动时添加完成且每隔一定时间就要遍历列表// 以获取所有监测文件的增量内容，所以这里使用CopyOnWriteArrayListList<LWFTAdaptor> adaptors// 默认间隔时间为2秒int DEFAULT_SAMPLE_PERIOD_MS = 1000 * 2// 间隔时间，由配置项"chukwaAgent.adaptor.context.switch.time"获取int SAMPLE_PERIOD_MS// 最大间隔时间为1分钟public static final int MAX_SAMPLE_PERIOD = 60 * 1000/** * 遍历adaptors，调用adaptor的tailFile方法 */public void run()
?
(5)基本数据结构
??? org.apache.hadoop.chukwa.Chunk
??? Chukwa中最基本的数据结构，用于收集数据，记录所收集数据的来源、格式、内容、大小等信息
?
??? org.apache.hadoop.chukwa.ChunkImpl
??? 实现了Chunk接口
// 生成当前对象的主机名String source// 数据源名称String streamName// 数据类型String dataType// 标签String tags// 数据内容byte[] data// 记录了每行记录的偏移量int[] recordEndOffsets// 序列号(文件偏移量)long seqID
