Alipay Tracer的编码要点

背景

• 一个分布式RPC调用，需要在被调用端记录调用过程中的TraceId以及其他上下文信息

编码要点

• 如果打点的位置在被调用端的主线程，直接用TracerContextUtil拿TraceId即可
• 如果被调用端在被调用过程中起了一个子线程并且需要在子线程中记录TraceId，那么就需要通过在主线程中建一个final对象做一下过渡，将Tracer上下文带入子线程，比如下面代码中的AbstractLogContext。然后在子线程内部去set这个上线文，这样就可以在子线程内部拿到TraceId了

@Override
    public void checkCorrectness(final QueryParameter queryParam, final QueryParameterDTO queryParameter,
            final OlapQueryModel queryModel, final IcubeRouteResultDTO icubeRouteResult, final String queryIcubeSql,
            final List<Map<String, Object>> icubeQueryResult) {
        final AbstractLogContext ctx = AbstractLogContext.get();
        final SoasUserInfo userInfo = UserInfoHolder.getUserInfo();
        // 用异步方式进行检查
        correctThreadPool.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    AbstractLogContext.set(ctx);
                    UserInfoHolder.setUserInfo(userInfo);

                    LogUtil.info(logger, "checkCorrectness start");

                    // 判断前置条件
                    // 1. check DRM 开关
                    if (!AlisisDynamicConfigUtil.isCheckIcubeCorrectness()) {
                        LogUtil.info(logger, "checkCorrectness drm close");
                        return;
                    }

                    // 2. 结果集大小小于limit 并且 小于 MAX_COMPARE_SIZE
                    if (queryParam.getLimit() == null || CollectionUtils.size(icubeQueryResult) >= queryParam
                            .getLimit() || CollectionUtils.size(icubeQueryResult) > MAX_COMPARE_SIZE) {
                        LogUtil.info(logger, "checkCorrectness limit queryLimit={0}, resultSize={1}",
                                queryParam.getLimit(), CollectionUtils.size(icubeQueryResult));
                        return;
                    }

                    // 3. 数据版本 不一致
                    Dataset dataset = queryParam.getDynamicCube().getDataset();
                    DSDTO dsInfo = connectionInfoProcesser.getDSInfo(dataset.getFacTable().getDsId());
                    OdpsTableVO tableByQueryOdps = odpsMetaDataService.getTableByQueryOdps(dsInfo.getConnectUrl(),
                            dsInfo.getProject(), dataset.getFacTable().getName());
                    LogUtil.info(logger, "getTableByQueryOdps: {0}", JSON.toJSONString(tableByQueryOdps));
                    Date icubeTime = icubeRouteResult.getCuboidSelect().getCuboid().getSourceTableCurrentModifiedTime();
                    Date odpsTime = null;
                    if (tableByQueryOdps != null) {
                        odpsTime = tableByQueryOdps.getDataModifyTime();
                    }

                    if (icubeTime == null || odpsTime == null
                            || !icubeTime.equals(odpsTime)) {
                        LogUtil.info(logger, "checkCorrectness data version icubeVersion={0}, odpsVersion={1}",
                                icubeTime, odpsTime);
                        return;
                    }

                    // 判断加速的数据版本是否一致
                    String icubeDsId = icubeRouteResult.getCuboidSelect().getCuboid().getBuildTableModel()
                            .getFactTable()
                            .getDsId();
                    String icubeTableName = icubeRouteResult.getCuboidSelect().getCuboid().getBuildTableModel()
                            .getFactTable()
                            .getTableName();
                    TableConnInfo tableInfo = new TableConnInfo();
                    tableInfo.setDsId(icubeDsId);
                    tableInfo.setTableName(icubeTableName);
                    List<TableConnInfo> tableList = new ArrayList<>();
                    tableList.add(tableInfo);
                    List<TableSpeedupInfo> speedupStatus = speedupManageService.getSpeedupStatus(tableList);
                    if (CollectionUtils.isNotEmpty(speedupStatus)) {
                        TableSpeedupInfo speedupDO = speedupStatus.get(0);
                        if (speedupDO.isSpeedupReady()) {
                            Date speedupVersion = speedupDO.getOdpsDataTime();
                            if (speedupVersion == null || !speedupVersion.equals(odpsTime)) {
                                LogUtil.info(logger,
                                        "checkCorrectness data version speedupVersion={0}, odpsVersion={1}",
                                        speedupVersion, odpsTime);
                                return;
                            }
                        }
                    }

                    // 查询ODPS数据
                    ContextUtil.setQueryFlag(QueryTypeEnum.SQL_GENERATION_ORIGINAL_ONLY.ordinal());
                    queryParameter.setQueryFlag(QueryTypeEnum.SQL_GENERATION_ORIGINAL_ONLY.ordinal());
                    String newQueryId = queryModel.getQueryConfigInfo().getQueryId()
                            + CORRECTNESS_CHECK_QUERY_ID_SUFFIX;
                    queryModel.getQueryConfigInfo().setQueryId(newQueryId);
                    QueryResultModel queryResultModel = queryHandler.queryForRawData(queryModel,
                            queryParameter, false);
                    if (queryResultModel == null) {
                        LogUtil.info(logger, "checkCorrectness query odps result null");
                        return;
                    }
                    List<Map<String, Object>> dataList = queryResultModel.getResultData();

                    // 后置处理
                    QueryUtils.convertDataTypeForOdps(dataList, queryParam, queryResultModel.getResultModel());

                    try {
                        LogUtil.info(logger, "checkCorrectness doCompare");
                        doCompare(dataList, icubeQueryResult);
                        LogUtil.info(logger, "checkCorrectness doCompare success");
                    } catch (IllegalArgumentException e) {
                        LogUtil.error(e, logger, "checkCorrectness faild, icubeSql=#{0}#, odpsSql=#{1}#", queryIcubeSql,
                                queryResultModel.getResultModel().getSqlText());
                        StringBuilder sb = new StringBuilder();
                        sb.append("checkCorrectness faild: ");
                        sb.append("[").append(TracerContextUtil.getTraceId()).append("]");
                        sb.append(e.getMessage()).append("\n");
                        sb.append("用户 ").append(UserInfoHolder.getUserInfo().getUserName());
                        sb.append("工号 ").append(UserInfoHolder.getUserId());
                        sb.append(" @ ").append(EnvUtil.getHostName()).append("\n");
                        sb.append("icubeSql: ").append(queryIcubeSql).append("\n");
                        sb.append("odpsSql: ").append(queryResultModel.getResultModel().getSqlText());
                        exceptionNotifyService.notifyIcube(sb.toString());
                    }
                } catch (Exception e) {
                    LogUtil.error(e, logger, "checkCorrectness error");
                }
            }
        });
    }