Zipkin原理学习 -- Druid 追踪多种数据库 SQL 执行

墨蓝 2022-03-06 10:48 236阅读 0赞

在上一篇博客 [《Zipkin原理学习–日志追踪 MySQL 执行语句》][Zipkin_ MySQL] 中我们已经了解学习到 Zipkin 官方提供的针对 MySQL 数据库 sql 语句执行的追踪拦截器，现在我们基于数据库连接池 Druid 的 Filter 机制 写一个能支持多种数据库（mysql，postgresql、oracle、sql server 等）类型 SQL 执行日志追踪拦截器。

**Druid 过滤器 Filter**

DruidDataSource支持通过Filter-Chain模式进行扩展，类似Serlvet的Filter，扩展十分方便，你可以拦截任何JDBC的方法。

有两种配置Filter的方式，一种是配置filters属性，一种是配置proxyFilters属性。filters和proxyFilters的配置是组合关系，而不是替换关系。

配置filters属性  
配置filters属性比较简单，filters的类型是字符串，多个filter使用逗号隔开。例如：

filters属性的配置使用别名或者全类名，stat是com.alibaba.druid.filter.stat.StatFilter的别名。[ 内置Filter的别名 ][_Filter_] 查看内置Filter的别名。

配置proxyFilters属性  
proxyFilters的类型是List，使用proxyFilters配置，可以有更多的配置选项。

**Zipkin 日志追踪过滤器 TracingStatementFilter**

TracingStatementFilter 继承抽象类 FilterEventAdapter 实现以下接口：

protected void statementExecuteUpdateBefore(StatementProxy statement, String sql) {
        }
    
        protected void statementExecuteUpdateAfter(StatementProxy statement, String sql, int updateCount) {
        }
    
        protected void statementExecuteQueryBefore(StatementProxy statement, String sql) {
        }
    
        protected void statementExecuteQueryAfter(StatementProxy statement, String sql, ResultSetProxy resultSet) {
        }
    
        protected void statementExecuteBefore(StatementProxy statement, String sql) {
        }
    
        protected void statementExecuteAfter(StatementProxy statement, String sql, boolean result) {
        }
    
        protected void statementExecuteBatchBefore(StatementProxy statement) {
        }
    
        protected void statementExecuteBatchAfter(StatementProxy statement, int[] result) {
        }
    
        protected void statement_executeErrorAfter(StatementProxy statement, String sql, Throwable error) {
        }

实现以上接口，在执行SQL语句之前、之后或抛出异常时都会执行上面接口方法，这样就可以追踪 SQL 语句的执行过程了。  
完整 Zipkin 日志追踪代码：

public class TracingStatementFilter extends FilterEventAdapter {
    
    
        private String zipkinServiceName;
    
        public TracingStatementFilter(String zipkinServiceName){
            this.zipkinServiceName = zipkinServiceName;
        }
    
        public TracingStatementFilter(){
    
        }
    
    
        @Override
        protected void statementExecuteUpdateBefore(StatementProxy statement, String sql) {
    
            super.statementExecuteUpdateBefore(statement,sql);
    
            Before(statement,sql);
    
        }
    
        @Override
        protected void statementExecuteUpdateAfter(StatementProxy statement, String sql, int updateCount) {
    
            super.statementExecuteUpdateAfter(statement,sql,updateCount);
    
            After(statement,sql);
    
        }
    
        @Override
        protected void statementExecuteQueryBefore(StatementProxy statement, String sql) {
    
            super.statementExecuteQueryBefore(statement,sql);
    
            Before(statement,sql);
    
        }
    
        @Override
        protected void statementExecuteQueryAfter(StatementProxy statement, String sql, ResultSetProxy resultSet) {
    
            super.statementExecuteQueryAfter(statement,sql,resultSet);
    
            After(statement,sql);
    
    
        }
    
        @Override
        protected void statementExecuteBefore(StatementProxy statement, String sql) {
    
            super.statementExecuteBefore(statement,sql);
            Before(statement,sql);
    
        }
    
        @Override
        protected void statementExecuteAfter(StatementProxy statement, String sql, boolean result) {
    
            super.statementExecuteAfter(statement,sql,result);
    
            After(statement,sql);
    
    
    
    
        }
    
        @Override
        protected void statementExecuteBatchBefore(StatementProxy statement) {
    
            super.statementExecuteBatchBefore(statement);
    
            Before(statement , null);
    
    
    
        }
    
        @Override
        protected void statementExecuteBatchAfter(StatementProxy statement, int[] result) {
    
            super.statementExecuteBatchAfter(statement , result);
            After(statement , null);
        }
    
        @Override
        protected void statement_executeErrorAfter(StatementProxy statement, String sql, Throwable error) {
    
            super.statement_executeErrorAfter(statement , sql , error);
            ErrorAfter(statement , sql , error);
    
        }
    
    
        protected void Before(StatementProxy statement, String sql) {
    
            try {
                Span span = ThreadLocalSpan.CURRENT_TRACER.next();
                if (span == null || span.isNoop()) {
                    return;
                }
    
                if(sql == null){
                    sql = statement.getLastExecuteSql();
                }
                // Allow span names of single-word statements like COMMIT
    
                int spaceIndex = sql.indexOf(' ');
                span.kind(Span.Kind.CLIENT).name(spaceIndex == -1 ? sql : sql.substring(0, spaceIndex));
                span.tag("sql.query", sql);
    
                parseServerIpAndPort(statement,span);
                span.start();
            }catch (Exception e){
    
            }
    
        }
    
        protected void After(StatementProxy statement, String sql) {
    
            try {
                Span span = ThreadLocalSpan.CURRENT_TRACER.remove();
                if (span == null || span.isNoop()) {
                    return;
                }
                span.finish();
                return ;
            }catch (Exception e){
    
            }
    
        }
    
        protected void ErrorAfter(StatementProxy statement, String sql, Throwable error) {
    
            try {
                Span span = ThreadLocalSpan.CURRENT_TRACER.remove();
                if (span == null || span.isNoop()) {
                    return;
                }
    
                if (error instanceof SQLException) {
                    span.tag("error", Integer.toString(((SQLException) error).getErrorCode()));
                }
                span.finish();
    
                return ;
            }catch (Exception e){
    
            }
    
        }
    
    
        public void parseServerIpAndPort(StatementProxy statement, Span span) {
            try {
                URI url = URI.create(statement.getConnection().getMetaData().getURL().substring(5));
                if (getZipkinServiceName() == null || "".equals(getZipkinServiceName())) {
                    try {
                        zipkinServiceName = "DB"+url.getPath();
                    }catch (Exception e){
                        ;
                    }
                }
                span.remoteServiceName(getZipkinServiceName());
                String host = url.getHost();
                if (host != null) {
                    span.remoteIpAndPort(host, url.getPort());
                }
            } catch (Exception e) {
                // remote address is optional
            }
        }
    
        public String getZipkinServiceName() {
            return zipkinServiceName;
        }
    
        public void setZipkinServiceName(String zipkinServiceName) {
            this.zipkinServiceName = zipkinServiceName;
        }
    }

使用配置示例：

**总结**：  
简单来说就是利用 Druid 提供的过滤器机制，在 SQL 语句执行前后添加追踪日志，工程地址 [brave-instrumentation-druid][]

提交给官方的Pull ：[https://github.com/openzipkin/brave/pull/880][https_github.com_openzipkin_brave_pull_880]

[Zipkin_ MySQL]: https://tianjunwei.blog.csdn.net/article/details/88559499
[_Filter_]: https://github.com/alibaba/druid/wiki/%E5%86%85%E7%BD%AEFilter%E7%9A%84%E5%88%AB%E5%90%8D
[brave-instrumentation-druid]: https://github.com/IAMTJW/brave-instrumentation-druid
[https_github.com_openzipkin_brave_pull_880]: https://github.com/openzipkin/brave/pull/880