Zipkin-mysqlConnector8.0 sql拦截器

• • 背景
  • Mysql驱动升级8.0
  • • Zipkin Mysql拦截器升级

最近在做微服务框架Springboot的版本升级，其中最为复杂的便是兼容性测试。
引起兼容性问题的几个主要原因：

• 依赖jar包方法不存在：No Such Method Error。解决办法为我们需要熟悉：maven依赖jar包优先级；
• 基础框架包的兼容性。比如：Springboot与Spring Cloud的兼容性。
• 框架升级会导致某些依赖版本发生变化，从而影响已经存在的代码或外部jar包。

做好这件事情，需要有以下知识储备：

• 熟悉maven依赖jar包优先级，推荐阅读
• 依赖问题解决办法，点我阅读

Connector8.0 sql拦截器，官方文档

官方提供了2个java文件，实现zipkin的SQL拦截：
TracingExceptionInterceptor.java

package brave.mysql8;

import brave.Span;
import brave.propagation.ThreadLocalSpan;
import com.mysql.cj.exceptions.ExceptionInterceptor;
import com.mysql.cj.log.Log;
import java.sql.SQLException;
import java.util.Properties;


public class TracingExceptionInterceptor implements ExceptionInterceptor {

  @Override public ExceptionInterceptor init(Properties properties, Log log) {
    String queryInterceptors = properties.getProperty("queryInterceptors");
    if (queryInterceptors == null ||
      !queryInterceptors.contains(TracingQueryInterceptor.class.getName())) {
      throw new IllegalStateException(
        "TracingQueryInterceptor must be enabled to use TracingExceptionInterceptor.");
    }
    return new TracingExceptionInterceptor();
  }

  @Override public void destroy() {
    // Don't care
  }

  
  @Override public Exception interceptException(Exception e) {
    Span span = ThreadLocalSpan.CURRENT_TRACER.remove();
    if (span == null || span.isNoop()) return null;

    span.error(e);
    if (e instanceof SQLException) {
      span.tag("error", Integer.toString(((SQLException) e).getErrorCode()));
    }

    span.finish();

    return null;
  }
}

racingQueryInterceptor.java

package brave.mysql8;

import brave.Span;
import brave.propagation.ThreadLocalSpan;
import com.mysql.cj.MysqlConnection;
import com.mysql.cj.Query;
import com.mysql.cj.interceptors.QueryInterceptor;
import com.mysql.cj.jdbc.JdbcConnection;
import com.mysql.cj.log.Log;
import com.mysql.cj.protocol.Resultset;
import com.mysql.cj.protocol.ServerSession;
import java.net.URI;
import java.sql.SQLException;
import java.util.Properties;
import java.util.function.Supplier;

import static brave.Span.Kind.CLIENT;


public class TracingQueryInterceptor implements QueryInterceptor {

  
  @Override
  public  T preProcess(Supplier sqlSupplier, Query interceptedQuery) {
    // Gets the next span (and places it in scope) so code between here and postProcess can read it
    Span span = ThreadLocalSpan.CURRENT_TRACER.next();
    if (span == null || span.isNoop()) return null;

    String sql = sqlSupplier.get();
    int spaceIndex = sql.indexOf(' '); // Allow span names of single-word statements like COMMIT
    span.kind(CLIENT).name(spaceIndex == -1 ? sql : sql.substring(0, spaceIndex));
    span.tag("sql.query", sql);
    parseServerIpAndPort(connection, span);
    span.start();
    return null;
  }

  private MysqlConnection connection;
  private boolean interceptingExceptions;

  @Override
  public  T postProcess(Supplier sql, Query interceptedQuery,
    T originalResultSet, ServerSession serverSession) {
    if (interceptingExceptions && originalResultSet == null) {
      // Error case, the span will be finished in TracingExceptionInterceptor.
      return null;
    }
    Span span = ThreadLocalSpan.CURRENT_TRACER.remove();
    if (span == null || span.isNoop()) return null;

    span.finish();

    return null;
  }

  
  static void parseServerIpAndPort(MysqlConnection connection, Span span) {
    try {
      URI url = URI.create(connection.getURL().substring(5)); // strip "jdbc:"
      String remoteServiceName = connection.getProperties().getProperty("zipkinServiceName");
      if (remoteServiceName == null || "".equals(remoteServiceName)) {
        String databaseName = getDatabaseName(connection);
        if (databaseName != null && !databaseName.isEmpty()) {
          remoteServiceName = "mysql-" + databaseName;
        } else {
          remoteServiceName = "mysql";
        }
      }
      span.remoteServiceName(remoteServiceName);
      String host = getHost(connection);
      if (host != null) {
        span.remoteIpAndPort(host, url.getPort() == -1 ? 3306 : url.getPort());
      }
    } catch (Exception e) {
      // remote address is optional
    }
  }

  private static String getDatabaseName(MysqlConnection connection) throws SQLException {
    if (connection instanceof JdbcConnection) {
      return ((JdbcConnection) connection).getCatalog();
    }
    return "";
  }

  private static String getHost(MysqlConnection connection) {
    if (!(connection instanceof JdbcConnection)) return null;
    return ((JdbcConnection) connection).getHost();
  }

  @Override
  public boolean executeTopLevelOnly() {
    return true;  // True means that we don't get notified about queries that other interceptors issue
  }

  @Override
  public QueryInterceptor init(MysqlConnection mysqlConnection, Properties properties,
    Log log) {
    String exceptionInterceptors = properties.getProperty("exceptionInterceptors");
    TracingQueryInterceptor interceptor = new TracingQueryInterceptor();
    interceptor.connection = mysqlConnection;
    interceptor.interceptingExceptions = exceptionInterceptors != null &&
      exceptionInterceptors.contains(TracingExceptionInterceptor.class.getName());
    if (!interceptor.interceptingExceptions) {
      log.logWarn("TracingExceptionInterceptor not enabled. It is highly recommended to "
        + "enable it for error logging to Zipkin.");
    }
    return interceptor;
  }

  @Override
  public void destroy() {
    // Don't care
  }
}