Apache Camel基于RoutePolicy实现审计功能

短命女 2022-11-05 14:53 140阅读 0赞

> 基于Apache Camel提供的扩展快速实现日志审计功能。

### 0. 目录 ###

*   *   *  1. 前言
         *  2. 可行性方案分析
         *  3. 具体实现
         *  4. 实现原理
         *  5. Links

### 1. 前言 ###

公司部门内部基于Apache Camel创建了一个"数据共享交互平台"，现在需要对所有流经平台的请求进行记录以提供审计功能，初看之下感觉这个需求对Apache Camel来说简直易如反掌，但一来是受限于过去半年已经对Apache Camel少有接触，二来则是在对于Apache Camel的残留记忆中， Apache Camel提供的扩展功能有些多（log / event / intercept / trace / wireTap等等），反而给人一种"无从选择"的感觉。

本文接下来的部分将就"审计"功能实现过程中的一些可行性方案和最终实行的方案进行介绍。

本文使用的Apache Camel版本为 `2.24.3`。

### 2. 可行性方案分析 ###

思路大致如下三种：

1.  借鉴Camel内置的`CamelInternalProcessor.InstrumentationAdvice`，该功能默认只提供了计数器式的统计，并没有对每个请求的入参，出参等信息进行收集，不过得益于Camel强大的可扩展性，通过阅读源码`DefaultRouteContext.commit()`的实现，我们发现了一种实现的可能性。这个也是本文将要重点讨论的内容。
2.  借助Camel内置的`CamelInternalProcessor.UnitOfWorkProcessorAdvice` 和 `RouteLifecycleAdvice`，这个和上面有异曲同工之妙，因此这里只是稍微提及下思路。借助`UnitOfWork`（部分核心实现细节位于`RouteLifecycleAdvice`中的 `uow.beforeRoute` / `uow.afterRoute`，再确切点说是`SynchronizationRouteAware`接口实现类 ）。
3.  借鉴框架 [openhub-framework][] 来实现，该框架的核心底层也是基于Apache Camel的，其实现思路是借助Apache Camel中提供的Event机制来完成。本框架完全开源，有兴趣的读者可以自行查阅。

最终我们选择了第一种思路来实现：

1.  相较于方案三，我们已经有了自己的一套框架，贸然进行迁移工作量大，加之需求工期紧张，且过往对于该框架我们只有很粗浅的了解，这使得本套方案更不现实。当然我们也可以只借鉴其这部分的思路，例如借鉴其request / response 的数据结构来对 camel 中的关键数据信息进行记录。
2.  相较于方案二，方案一的实现逻辑更为简单，本着"简单就是美"的原则，我们选择方案一。

### 3. 具体实现 ###

正如上面小节本提到的，本文先给出第一种思路的代码实现，然后再进行原理性分析。

// ================ 因为RoutePolicy是基于Route的, 所以为了做到全局统一配置, 这里我们通过实现RoutePolicyFactory接口来实现
    // RoutePolicyFactoryInstrumentationImpl将为用户自定义的每个Route添加一个RoutePolicyAdvice, 实现了类似切面的逻辑统一存放。
    class RoutePolicyFactoryInstrumentationImpl implements RoutePolicyFactory { 
    	static final String EXCHANGE_PROERTIES_KEY_STOPWATCH_FOR_METRIC = "METRIC-STOPWATCH";
    
    	private final RoutePolicy policy = new RoutePolicyInstrumentationImpl();
    
    	@Override
    	public RoutePolicy createRoutePolicy(CamelContext camelContext, String routeId, RouteDefinition route) { 
    		return policy;
    	}
    
    	// =========================================
    	static class RoutePolicyInstrumentationImpl extends org.apache.camel.support.RoutePolicySupport
    	      implements RoutePolicy { 
    
    		@Override
    		public void onExchangeBegin(Route route, Exchange exchange) { 
    			// 参考 CamelInternalProcessor.InstrumentationAdvice 内部类
    			final StopWatch answer = new StopWatch();
    			// ====================== 这里要使用栈进行存储, 至于原因:
    			// === 因为可能存在支线route, 例如 from().to("direct:500").to(xxxx) , 这里的"direct:500"会导致重建exchange, 也就是这里会再次被调用
    			if(exchange.getProperty(EXCHANGE_PROERTIES_KEY_STOPWATCH_FOR_METRIC) == null){ 
    				exchange.setProperty(EXCHANGE_PROERTIES_KEY_STOPWATCH_FOR_METRIC, new ArrayDeque<StopWatch>());
    			}
    			@SuppressWarnings("unchecked")
    			Deque<StopWatch> deque = exchange.getProperty(EXCHANGE_PROERTIES_KEY_STOPWATCH_FOR_METRIC, Deque.class);
    			deque.push(answer);
    		}
    
    		@Override
    		public void onExchangeDone(Route route, Exchange exchange) { 
    			@SuppressWarnings("unchecked")
    			Deque<StopWatch> deque  = exchange.getProperty(EXCHANGE_PROERTIES_KEY_STOPWATCH_FOR_METRIC, Deque.class);
    			if(deque.size() > 1){ 
    				// 主Route还没走完
    				StopWatch subSW = deque.poll();
    				Console.log("支exchange {} 耗時 {} millis; routeId [ {} ]", exchange.getExchangeId(), subSW.taken(), route.getId());
    				return;
    			}
    			
    			
    			Console.log("主exchange {} 耗時 {} millis; routeId [ {} ]", exchange.getExchangeId(), deque.poll().taken(), route.getId());
    		}
    
    	}
    }
    
    // ====================== 配置
    @Component
    public class RoutePolicySample extends RouteBuilder { 
    
    	@Override
    	public void configure() throws Exception { 
    	    // 全局配置, 避免需要在每个Route定义上添加
    		getContext().addRoutePolicyFactory(new RoutePolicyFactoryInstrumentationImpl());
    
    		from("rest:get,post:rp")//
    		      .routeId("RoutePolicy") //
    		      // 针对单个Route进行配置
    		      // .routePolicy(new RoutePolicyFactoryInstrumentationImpl.RoutePolicyInstrumentationImpl())//
    		      .process(customProcessor)//
    		      .to("direct:500")//
    		      .to("stream:err");
    
    		from("direct:500") //
    		      .to("stream:out");
    		
    	}
    }

以上就是核心实现了，代码很简单。当然目前是只提供了整个Route的执行耗时统计，但很明显我们可以基于某些内部约定来进行所有经过平台请求的入/出参等等进行记录入库。之后就能根据这些入库的数据来进行审计功能的实现。

效果如下：  
![执行效果图][20210307121547180.png]

### 4. 实现原理 ###

相较于上面的实现，相信需要这篇文章的童鞋应该更关注其底层实现原理和细节。下面就让我们来看看。

如我们之前已经介绍过的[Apache Camel源码研究之启动  
][Apache Camel_]，Apache Camel内部将用户定义的每个`RouteDefinition`中的`outputs字段`部分组装为一个`CamelInternalProcessor`实例，每个Route的主体逻辑处理部分都被封装在了这个实例中，而关于这个实例的组成结构则在[Apache Camel源码研究之CamelInternalProcessor][Apache Camel_CamelInternalProcessor]中进行了专门的介绍，而我们本文所关注的功能实现也正是以此为基础。

// DefaultRouteContext.commit()
    // 以下列出的是 CamelInternalProcessor 的 Advice 的组装顺序，我们的方案正是基于其提供的扩展。
    public void commit() { 
    	// now lets turn all of the event driven consumer processors into a single route
    	if (!eventDrivenProcessors.isEmpty()) { 
    		Processor target = Pipeline.newInstance(getCamelContext(), eventDrivenProcessors);
    
    		// force creating the route id so its known ahead of the route is started
    		String routeId = route.idOrCreate(getCamelContext().getNodeIdFactory());
    
    		// and wrap it in a unit of work so the UoW is on the top, so the entire route will be in the same UoW
    		CamelInternalProcessor internal = new CamelInternalProcessor(target);
    		// 首先是 UnitOfWorkProcessorAdvice
    		internal.addAdvice(new CamelInternalProcessor.UnitOfWorkProcessorAdvice(this));
    
            // ========== 这里就是Apache Camel提供的可扩展点, 也是本文产生的基础
    		// and then optionally add route policy processor if a custom policy is set
    		List<RoutePolicy> routePolicyList = getRoutePolicyList();
    		if (routePolicyList != null && !routePolicyList.isEmpty()) { 
    			......
    
    			internal.addAdvice(new CamelInternalProcessor.RoutePolicyAdvice(routePolicyList));
    		}
    
    		// wrap in route inflight processor to track number of inflight exchanges for the route
    		internal.addAdvice(new CamelInternalProcessor.RouteInflightRepositoryAdvice(camelContext.getInflightRepository(), routeId));
    
            // 这个就是我们本次所借鉴的InstrumentationAdvice
    		// wrap in JMX instrumentation processor that is used for performance stats
    		internal.addAdvice(new CamelInternalProcessor.InstrumentationAdvice("route"));
    
    		// wrap in route lifecycle
    		internal.addAdvice(new CamelInternalProcessor.RouteLifecycleAdvice());
    		
    		// === 存在rest接口情况下，对入参和出参进行序列/反序列化
    		// wrap in REST binding
    		if (route.getRestBindingDefinition() != null) { 
    			try { 
    				internal.addAdvice(route.getRestBindingDefinition().createRestBindingAdvice(this));
    			} catch (Exception e) { 
    				throw ObjectHelper.wrapRuntimeCamelException(e);
    			}
    		}
    		
    		// === 对本Route的入/出参进行契约
    		// wrap in contract
    		if (route.getInputType() != null || route.getOutputType() != null) { 
    			Contract contract = new Contract();
    			if (route.getInputType() != null) { 
    				contract.setInputType(route.getInputType().getUrn());
    				contract.setValidateInput(route.getInputType().isValidate());
    			}
    			if (route.getOutputType() != null) { 
    				contract.setOutputType(route.getOutputType().getUrn());
    				contract.setValidateOutput(route.getOutputType().isValidate());
    			}
    			internal.addAdvice(new ContractAdvice(contract));
    			// make sure to enable data type as its in use when using input/output types on routes
    			camelContext.setUseDataType(true);
    		}
    
    		......
    
    		// invoke init on route policy
    		if (routePolicyList != null && !routePolicyList.isEmpty()) { 
    			for (RoutePolicy policy : routePolicyList) { 
    				policy.onInit(edcr);
    			}
    		}
    
    		routes.add(edcr);
    	}
    }

这里先老生常谈一句：**“Apache Camel的社区真是太活跃了！”**，相较于笔者在 [Apache Camel源码研究之CamelInternalProcessor][Apache Camel_CamelInternalProcessor]中的结构图，上面代码说明Apache Camel已经额外添加了数个Advice以实现不同的功能（例如对restful接口进行参数序列化/反序列化的`RestBindingAdvice`，以及对入参/出参进行直接约束的`ContractAdvice`等）。

上面的代码逻辑最终会组装如[Apache Camel源码研究之CamelInternalProcessor][Apache Camel_CamelInternalProcessor]类似的请求处理链路图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xxemtjeDM_size_16_color_FFFFFF_t_70_pic_center]

上图中，我们将要插入的扩展RoutePolicyAdvice位于`UnitOfWorkProcessorAdvice`和`RouteInflightRepositoryAdvice`之间，这是要优先于Apache Camel内置的`InstrumentationAdvice`的。

而且按照Apache Camel整体架构设计来说，其处理的所有请求都将流经这些Advice，因此对于本文所涉及的需求，本扩展是完全可以满足的。

### 5. Links ###

1.  [为什么推荐使用Apache Camel作EAI？][Apache Camel_EAI]
2.  [Apache Camel源码研究之启动  
    ][Apache Camel_]
3.  [Apache Camel源码研究之CamelInternalProcessor][Apache Camel_CamelInternalProcessor]
4.  [Office - RoutePolicy][]

[openhub-framework]: https://openhubframework.atlassian.net/wiki/spaces/OHF/pages/33606/Request+response+tracking
[20210307121547180.png]: /images/20221023/d0866268e7c54679af65ff255105bfd5.png
[Apache Camel_]: https://blog.csdn.net/lqzkcx3/article/details/103263042
[Apache Camel_CamelInternalProcessor]: https://blog.csdn.net/lqzkcx3/article/details/104126561
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xxemtjeDM_size_16_color_FFFFFF_t_70_pic_center]: /images/20221023/e404894950324429b8a032ee965f48c3.png
[Apache Camel_EAI]: https://blog.csdn.net/lqzkcx3/article/details/103104559
[Office - RoutePolicy]: https://camel.apache.org/manual/latest/route-policy.html