【CAT魔改】扩展cat-consumer实现自定义服务端收集日志

矫情吗；* 2023-01-22 11:53 17阅读 0赞

> 通过cat-consumer的源码修改，实现将CAT服务端收集到的日志按照自定义需求进行额外持久化，为后续的自定义日志审计打基础。

### 目录 ###

*   *   *  1. 前言
         *  2. 实现
         *  3. 原理
         *  4. 最后
         *  5. Links

### 1. 前言 ###

随着部门内部系统大型化，微服务化的进展，监控的重要性伴随着实际项目中盲人摸象式的问题排查愈发凸显。

试图改变他人的想法并不是推进某项制度的良策，你需要收起自己的骄傲，通过修改标准的制度流程来适应团队的实际，解决他们的痛点，将团队吸引过来，然后再慢慢引导他们走到标准路途上来。

不论你自身的切身感受有多强烈，如果无法说服团队进行监控的落地，那监控这件事在你们团队就是失败的，而这给你带来的将只有无尽的痛苦 —— 自身想法的不被认同，明知道是条不归路却被迫继续向前走。

### 2. 实现 ###

按照CAT编写者本人发表的[深度剖析开源分布式监控CAT][CAT] 中的介绍，我们最终将目光锁定到 cat-consumer项目上：  
![在这里插入图片描述][20210520102553147.png]

经过追踪调试，最终确定了扩展方式如下：  
![在这里插入图片描述][20210520103126562.png]  
核心类就只有这一个 `MessageTreePersistAnalyer` 。

// 注意这个package, cat-consumer的包结构非常清晰, 一个package就是一个 MessageTree 解析逻辑.
    package com.dianping.cat.consumer.messagetreepersist;
    /** * refer to {@code DumpAnalyzer} * */
    @Named(type = MessageAnalyzer.class, value = MessageTreePersistAnalyer.ID, instantiationStrategy = Named.PER_LOOKUP)
    public class MessageTreePersistAnalyer extends AbstractMessageAnalyzer<Object> implements LogEnabled { 
    	public static final String ID = "messagetree-persist";
    
    	private Logger m_logger;
    
    	private int m_discradSize = 50000000;
    
    	@Override
    	public synchronized void doCheckpoint(boolean atEnd) { 
    		m_logger.info("MessageTreePersistAnalyer.doCheckpoint, atEnd : " + atEnd);
    	}
    
    	@Override
    	public void enableLogging(Logger logger) { 
    		m_logger = logger;
    	}
    
    	@Override
    	public Object getReport(String domain) { 
    		throw new UnsupportedOperationException("This should not be called!");
    	}
    
    	@Override
    	public ReportManager<?> getReportManager() { 
    		return null;
    	}
    
    	@Override
    	public void initialize(long startTime, long duration, long extraTime) { 
    		super.initialize(startTime, duration, extraTime);
    		int hour = (int) TimeUnit.MILLISECONDS.toHours(startTime);
    
    		System.out.println(hour);
    	}
    
    	@Override
    	protected void loadReports() { 
    		// do nothing
    	}
    
    	@Override
    	public void process(MessageTree tree) { 
    		try { 
    			MessageId messageId = MessageId.parse(tree.getMessageId());
    
    			if (!shouldDiscard(messageId)) { 
    				processWithStorage(tree, messageId, messageId.getHour());
    			}
    		} catch (Exception ignored) { 
    			m_logger.error("process messageTree fail", ignored);
    		}
    	}
    
    	private boolean shouldDiscard(MessageId id) { 
    		int index = id.getIndex();
    
    		return index > m_discradSize;
    	}
    
    	/// ======================================= 以下为主要的扩展点
    	/* * 本方法用于判断当前的 MessageTree 是需要被处理 / 是本类所关心的 * * @see * com.dianping.cat.analysis.AbstractMessageAnalyzer#isEligable(com.dianping * .cat.message.spi.MessageTree) */
    	@Override
    	public boolean isEligable(MessageTree tree) { 
    		if (!(tree.getMessage() instanceof Transaction)) { 
    			return false;
    		}
    
    		final Transaction transaction = (Transaction) tree.getMessage();
    		// 最大化地削减非关心日志的处理, 只处理我们感兴趣的日志
    		return (transaction.getType().equals(CatConstants.TYPE_URL)
    				|| transaction.getType().equals(CatConstants.TYPE_URL_FORWARD))
    				&& (!transaction.getName().startsWith("/cat/"));
    	}
    
    	/** * 上面的 {@code isEligable(MessageTree tree)} 方法返回true时候, 本方法将被调用 * * @param tree * @param messageId * @param hour */
    	private void processWithStorage(MessageTree tree, MessageId messageId, int hour) { 
    		tree.setFormatMessageId(messageId);
    		// TODO 这里读者可以使用类似 Java SPI特性向外公开自定义扩展机制。实现可以参考本文底部链接, 或者 CAT 源生的 CatPropertyProvider 接口的实现逻辑。
    		Console.log(tree);
    	}
    
    }

CAT使用Plexus容器进行实例的构建和依赖管理，因此我们需要将上面扩展的`MessageTreePersistAnalyer` 注册到容器中

package com.dianping.cat.consumer.build;
    
    public class ComponentsConfigurator extends AbstractResourceConfigurator { 
    
    	@Override
    	public List<Component> defineComponents() { 
    		List<Component> all = new ArrayList<Component>();
    		......
    		all.addAll(defineBusinessComponents());
    
    		// 注册 defineMessageTreePersistComponents 到Plexus容器中
    		all.addAll(defineMessageTreePersistComponents());
    
    		all.add(A(AtomicMessageConfigManager.class));
    		......
    		
    		return all;
    	}
    }
    
    private Collection<Component> defineMessageTreePersistComponents() { 
    	final List<Component> all = new ArrayList<>();
    	all.add(A(MessageTreePersistAnalyer.class));
    
    	return all;
    }

让以上plexus容器配置生效：

# 项目根目录下执行如下命令：
    # 这一句生效, 底层实现依赖于 cat-consumer 项目pom.xml文件中的 plexus-maven-plugin PLUGIN。
    # 最终你将在 cat-consumer/src/main/resources/META-INF/plexus/components.xml 文件中看到 MessageTreePersistAnalyer 身影。
    mvn clean package

### 3. 原理 ###

最后让我们就上面的扩展实现原理，进行一点源码层面的解读。

通过追踪调试，我们最终获得如下堆栈图，并推导出如下结论：

1.  CAT 使用 `TcpSocketReceiver` 来进行 客户端日志的接收。
2.  下图中的`isEligable(MessageTree tree)`方法用于判断当前的 `MessageTree`(客户端日志) 是需要被处理 / 是本扩展类所关心的；这一步直接决定当前`MessageTree`是否存入队列。如下图，也说明了 接收客户端日志 和 上述判断逻辑是在同一条线程内同步执行的，因此我们自定义扩展的`MessageTreePersistAnalyer.isEligable(MessageTree tree)`方法，效率非常重要。  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xxemtjeDM_size_16_color_FFFFFF_t_70]

CAT中大量使用"生产者-消费者"模式来进行操作解耦，最大化地提高系统吞吐，提升服务在极限环境下的表现。

上面截图中，CAT已经将代表客户端日志的`MessageTree`经过筛选存放到专门的队列中，之后CAT将使用专门的线程来处理这些推入队列的`MessageTree`。于是我们可以得到如下堆栈，从中我们可以看到：

1.  `MessageTree`的消费步是异步执行的，实现了最大化的性能保障。
2.  该异步操作应该尽量保证不出现异常。  
    ![在这里插入图片描述][20210519191429223.png]

### 4. 最后 ###

对于之前基本没有监控体系的应用，在推进监控的过程我们应该以使用者的第一诉求为突破口，通过快速解决其切身感受到的痛苦来将加快监控体系的推进，最终达到新技术落地的目的。

### 5. Links ###

1.  [大众点评Cat–Server模块架构分析  
    ][Cat_Server_]
2.  [CAT魔改 - 独立化cat-client][CAT_ - _cat-client]

[CAT]: https://tech.meituan.com/2018/11/01/cat-in-depth-java-application-monitoring.html
[20210520102553147.png]: /images/20221020/42c1bd4260cb4807ab1067384bf52785.png
[20210520103126562.png]: /images/20221020/498cfc65bedc43c982437eeccd47f0bc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xxemtjeDM_size_16_color_FFFFFF_t_70]: /images/20221020/b6bf0d972ece4b429119d809e2924505.png
[20210519191429223.png]: /images/20221020/fe29c47661d246e7b42e01122c9493f2.png
[Cat_Server_]: https://blog.csdn.net/szwandcj/article/details/50992580
[CAT_ - _cat-client]: https://blog.csdn.net/lqzkcx3/article/details/115023572