首页
栏目
ERP
链路追踪在ERP系统中的应用实践

链路追踪在ERP系统中的应用实践

2020-04-29
Admin

源宝导读：随着ERP的部署架构越来越复杂，对运维监控、问题排查等工作增加了难度，本文将介绍通过引入链路追踪技术，提高ERP系统问题排查效率，支撑更全面监控系统运行情况的实践过程。

一、导读

随着ERP的部署架构越来越复杂，微应用分布式部署架构在给用户带来高性能高稳定的同时，给运维监控、问题排查带来了一定难度，特别是后端服务的内部调用由于缺少日志，难以快速定位问题根因。借助链路追踪并结合异常日志所记录的链路id，可以方便定位整个异常链路，更快找到问题的原因。链路追踪还有一个好处，针对性能慢的页面，原来很难快速定位到慢的具体点，通过借助链路追踪，能快速掌握每个请求执行了哪些操作，每个操作消耗了多长时间，精准定位性能问题。

二、链路追踪介绍

分布式系统变得日趋复杂，越来越多的组件开始走向分布式化，如微服务、分布式数据库、分布式缓存等，使得后台服务构成了一种复杂的分布式网络。

在服务能力提升的同时，复杂的网络结构也使问题定位更加困难。在一个请求在经过诸多服务过程中，出现了某一个调用失败的情况，查询具体的异常由哪一个服务引起的就变得十分抓狂，问题定位和处理效率是也会非常低。

分布式链路追踪就是将一次分布式请求还原成调用链路，将一次分布式请求的调用情况集中展示，比如各个服务节点上的耗时、请求具体到达哪台机器上、每个服务节点的请求状态等等。

目前业界的链路追踪系统，如Twitter的Zipkin，Uber的Jaeger，国内比较流行的SkyWalking。

三、ERP中链路追踪的实现

什么是 Diagnostics

在.NET Core中实现链路追踪非常简单，因为在 .NET Core 中 .NET 团队设计了一个全新的 DiagnosticSource，新的 DiagnosticSource 非常的简单，它允许你在生产环境记录丰富的 payload 数据，然后你可以在另外一个消费者中消费感兴趣的记录。

ERP因为是.NET Framework实现链路追踪复杂一些，对此为了实现链路追踪，我们将Diagnostics应用到ERP中，手动创建生产者，然后在具体的链路追踪中创建对应消费者采集数据。

Diagnostics 就是提供一组功能使我们能够很方便的可以记录在应用程序运行期间发生的关键性操作以及他们的执行时间等，使管理员可以查找特别是生产环境中出现问题所在的根本原因。

在应用程序出现问题的时候，特别是出现可用性或者性能问题的时候，开发人员或者IT人员经常会对这些问题花费大量的时间来进行诊断，很多时候生产环境的问题都无法复现，这可能会对业务造成很大的影响。

目前平台实现了两种对接SkyWalking和Jaeger，可以方便的在配置中心启用，接入自建服务或云服务都是支持。

SkyWalking对接

SkyWalking目前对于.NET Core有着很好的支持，但是对于.NET Framework因为本身Diagnostics的原因，只有一个很简单的ASP.NET请求客户端。

其它的客户端我们都做了一些改造，但有一些我们也尽量标准，如SqlClient、HttpClient。以下介绍一下Redis的支持，完全自己实现客户端。

首先要创建生产者DiagnosticListener，然后在Redis的相关操作里执行具体事件。

DiagnosticListener需要引入System.Diagnostics.DiagnosticSource这个dll,以下为简单实现执行前事件：


internal static class RedisDiagnosticListenerExtensions
{
    /// 
    /// 定义监听的名称
    /// 
    public const string DiagnosticListenerName = "RedisDataDiagnosticListener";
 
    /// 
    /// 前缀
    /// 
    public const string CacheDataPrefix = "Mysoft.Map6.Cache.";
 
    /// 
    /// 执行前事件名称
    /// 
    public const string CacheBeforeExecuteName = CacheDataPrefix + nameof(RedisExecuteBefore);
 
    public static Guid RedisExecuteBefore(this DiagnosticListener @this, string operation,
         string endpoint, string cacheKey, long? cacheLength = null)
    {
        if (@this.IsEnabled(CacheBeforeExecuteName))
        {
            Guid operationId = Guid.NewGuid();
            @this.Write(
                CacheBeforeExecuteName,
                new
                {
                    OperationId = operationId,
                    Operation = operation,
                    Endpoint = endpoint,
                    CacheKey = cacheKey,
                    CacheLength = cacheLength,
                    Timestamp = Stopwatch.GetTimestamp()
                });
 
 
            return operationId;
        }
        else
            return Guid.Empty;
    }
}

然后在具体行为里面执行事件，以下写入Redis缓存为例：


private bool WriteToRedis(string key, byte[] value, TimeSpan expires)
{
    var operationId = s_diagnosticListener.RedisExecuteBefore(nameof(WriteToRedis), _redisProvider.GetCurrentEndPoint().ToString(), key, cacheLength: value?.Length);
    try
    {
        var result = RetryExecute(() => _redisProvider.GetRedis().GetDatabase().StringSet(key, value, expires));
 
        s_diagnosticListener.RedisExecuteAfter(operationId, nameof(WriteToRedis), _redisProvider.GetCurrentEndPoint().ToString(),  key);
 
        return result;
    }
    catch (Exception ex)
    {
 
        s_diagnosticListener.RedisExecuteError(operationId, ex, endpoint: _redisProvider.GetCurrentEndPoint().ToString(), cacheKey: key, cacheLength: value?.Length);
 
        throw;
    }
}

在写入Redis缓存前执行RedisExecuteBefore传入基本信息，然后写入Redis缓存完成再执行RedisExecuteAfter，如果写入Redis缓存发生异常就会执行RedisExecuteError。

最终在消费实现的时候，根据对应的信息构造链路跟踪信息，在After和Error中写入链路信息到队列。具体逻辑如下：


internal class RedisDiagnosticProcessor : ITracingDiagnosticProcessor
 {
 
     public static string ComponentName = "StackExchange.Redis";
 
     private readonly ITracingContext _tracingContext;
 
     private readonly IExitSegmentContextAccessor _contextAccessor;
 
     public string ListenerName => RedisDiagnosticStrings.DiagnosticListenerName;
 
     public RedisDiagnosticProcessor(ITracingContext tracingContext,
         IExitSegmentContextAccessor contextAccessor)
     {
         _tracingContext = tracingContext;
         _contextAccessor = contextAccessor;
     }
 
     private static string ResolveOperationName(string operation)
     {
         return $"{RedisDiagnosticStrings.CacheDataPrefix}  {operation}";
     }
     [DiagnosticName(RedisDiagnosticStrings.CacheBeforeExecuteName)]
     public void CacheExecuteBefore([Property(Name = "Endpoint")] string endpoint,
         [Property(Name = "CacheKey")] string cacheKey, [Property(Name = "CacheLength")]long? cacheLength, [Property(Name = "Operation")]string operation)
     {
         var context = _tracingContext.CreateExitSegmentContext(ResolveOperationName(operation), endpoint);
         context.Span.SpanLayer = SpanLayer.CACHE;
         context.Span.Component = new StringOrIntValue(ComponentName);
         context.Span.AddTag(CacheTags.CACHEKEY, cacheKey);
         if (cacheLength != null)
         {
             context.Span.AddTag(CacheTags.CACHELENGTH, cacheLength.Value);
         }
         context.Span.AddTag(CacheTags.OPERATION, operation);
     }
 
     [DiagnosticName(RedisDiagnosticStrings.CacheAfterExecuteName)]
     public void CacheExecuteAfter([Property(Name = "CacheLength")]long? cacheLength)
     {
         var context = _contextAccessor.Context;
         if (context != null)
         {
             if (cacheLength != null)
             {
                 context.Span.AddTag(CacheTags.CACHELENGTH, cacheLength.Value);
             }
             _tracingContext.Release(context);
         }
     }
 
     [DiagnosticName(RedisDiagnosticStrings.CacheErrorExecuteName)]
     public void CacheExecuteError([Property(Name = "Exception")] Exception ex)
     {
         var context = _contextAccessor.Context;
         if (context != null)
         {
             context.Span.ErrorOccurred(ex);
             _tracingContext.Release(context);
         }
     }
 }

ListenerName 就是上面RedisDiagnosticListener定义的名称。因为CacheExecuteBefore、CacheExecuteAfter这些方法都是通用的，同样会应用于读取Redis缓存和移除Redis缓存。

SkyWalking会构造一个 SegmentContext，在CacheExecuteBefore的时候构造Context信息，然后CacheExecuteAfter或CacheExecuteError发布Context 信息。

同时在SkyWalking的.NET客户端里会维护一个队列，将链路Context信息缓存在其中，例如当满足一定条件：如达到1000条或5秒钟等条件，就会推送至SkyWalking服务端。这个在客户端初始化的时候是可以配置的。

Jaeger对接

Jaeger的对接，在客户端上稍微做了一些改动。由于其中一个库OpenTracing.Contrib没有Framework版，平台单独编译的一个Framework版本。

然后Jaeger客户端是完美支持OpenTracing标准，平台采集数据的是标准OpenTracing格式，对接其它客户端也是会非常容易。

OpenTracing 是与后台无关的一套接口，被跟踪的服务只需要调用这套接口，就可以被任何实现这套接口的跟踪后台（比如Zipkin, Jaeger等等）支持，而作为一个跟踪后台，只要实现了个这套接口，就可以跟踪到任何调用这套接口的服务。

标准化了对跟踪最小单位Span的管理：定义了开始Span，结束Span和记录Span耗时的API。Span的定义可以参照开源分布式跟踪系统Zipkin介绍（架构篇）
标准化了进程间跟踪数据传递的方式：定义了一套API方便跟踪数据的传递
标准化了进程内当前Span的管理：定义了存储和获取当前Span的API

以下是Redis对应监听的实现：


internal class RedisDiagnosticObserver : DiagnosticListenerObserver
{
    public static string ComponentName = "StackExchange.Redis";
    private readonly ITracer _tracer;
 
    /// 
    public RedisDiagnosticObserver(ILoggerFactory loggerFactory, ITracer tracer,
        IOptions genericEventOptions) : base(loggerFactory, tracer, genericEventOptions.Value)
    {
        _tracer = tracer;
    }
 
    /// 
    protected override string GetListenerName() => RedisDiagnosticStrings.DiagnosticListenerName;
 
    /// 
    protected override void OnNext(string eventName, object untypedArg)
    {
        switch (eventName)
        {
            case RedisDiagnosticStrings.CacheBeforeExecuteName:
                RedisExecuteBefore(untypedArg);
                break;
            case RedisDiagnosticStrings.CacheAfterExecuteName:
                RedisExecuteAfter(untypedArg);
                break;
            case RedisDiagnosticStrings.CacheErrorExecuteName:
                RedisExecuteError(untypedArg);
                break;
        }
    }
}

以上对应的具体方法省略，跟SkyWalking类似，其中主要构建的是Scope，然后内部是Span，这些内容都是OpenTracing中的对象，最终再引入Jaeger的客户端即可完成接入。

ERP整个链路数据流转如下图：

四、与ERP结合

由于ERP基于.NET Framework，有很多场景是需要自己调整。具体在ERP中如何接入调整的，下面可以看看，目前初始化使用的动态HttpModule注入，然后在HttpModule中进行初始化。

动态注入，利用System.Web.PreApplicationStartMethod特性在程序启动时执行方法，然后使用DynamicModuleUtility中RegisterModule方法注册HttpModule。要使用RegisterModule方法，需要引入Microsoft.Web.Infrastructure类库。


[assembly: System.Web.PreApplicationStartMethod(typeof(InstrumentModuleFactory), nameof(InstrumentModuleFactory.Create))]
namespace Mysoft.Map6.OpenTracing.Startup
{
    public class InstrumentModuleFactory
    {
        public static void Create()
        {
            DynamicModuleUtility.RegisterModule(typeof(InstrumentModule));
        }
    }
}