返回

通过 EF/Dapper 扩展实现数据库审计功能

相信大家都有过周末被电话“吵醒”的经历,这个时候,客服同事会火急火燎地告诉你,客户反馈生产环境上某某数据“异常”,然后你花费大量时间去排查这些错误数据,发现这是客户使用某一种“骚”操作搞出来的“人祸”。可更多的时候,你不会这么顺利,因为你缺乏有力的证据去支持你的结论。最终,你不情愿地去处理了这些错误数据。你开始反思,为什么没有一种流程去记录客户对数据的变更呢?为什么你总要花时间去和客户解释这些数据产生的原因呢?好了,这就要说到我们今天这篇博客的主题——审计。

什么是审计

结合本文引言中的描述的场景,当我们需要知道某条数据被什么人修改过的时候,或者是希望在数据变更的时候去通知某个人,亦或者是我们需要追溯一条数据的变更历史的时候,我们需要一种机制去记录数据表中的数据变更,这就是所谓的审计。而实际的业务中,可能会有类似,查询某一个员工一天内审批了多少单据的需求。你不要笑,人类常常如此无聊,就像我们有一个异常复杂的计费逻辑,虽然审计日志里记录了某个费用是怎么计算出来的,可花时间最多的地方,无一例外是需要开发去排查和解释的,对于这一点,我时常感觉疲于应对,这是我这篇文章里想要写审计的一个重要原因。

EF/EF Core 实体跟踪

EF 和 EF Core 里都提供了实体跟踪的功能,我的领导经常吐槽我,在操作数据库的时候,喜欢显式地调用repository.Update()方法,因为他觉得项目中的实体跟踪是默认打开的。可当你学习了Vue以后,你了解到Vue中是检测不到数组的某些变化的,所以,这个事情我持保留意见,显式调用就显式调用呗,万一哪天人家把实体跟踪给关闭了呢?不过,话说回来,实体跟踪确实可以帮我们做一点工作的,其中,就包括我们今天要说的审计功能。

EF 和 EF Core 中的实体追踪主要指 DbContext 类的 ChangeTracker,而通过 DetachChanges()方法,则可以获得那些变化了的实体的集合。所以,使用实体追踪来实现审计功能,本质上就是在 SaveChanges()方法调用前后,记录实体中每一个字段的变化情况。为此,我们考虑编写下面的类——AuditDbContextBase,顾名思义,这是一个审计相关的 DbContext 基类,所以,希望实现审计功能的 DbContext 都会继承这个类。这里,我们重写其 SaveChanges()方法,其基本定义如下:

 1public class AuditDbContextBase : DbContext, IAuditStorage
 2{
 3    public DbSet<AuditLog> AuditLog { get; set; }
 4    public AuditDbContextBase(DbContextOptions options, AuditConfig auditConfig) : base(options) { }
 5    public virtual Task BeforeSaveChanges() { }
 6    public virtual Task AfterSaveChanges() { }
 7
 8    public override async Task<int> SaveChangesAsync(bool acceptAllChangesOnSuccess,
 9         CancellationToken cancellationToken = default)
10    {
11        await BeforeSaveChanges();
12        var result = await base.SaveChangesAsync(acceptAllChangesOnSuccess, cancellationToken);
13        await AfterSaveChanges();
14        return result;
15    }
16
17    public void SaveAuditLogs(params AuditLog[] auditLogs)
18    {
19        AuditLog.AddRange(auditLogs);
20        base.SaveChangesAsync();
21    }
22}

接下来,就是去实现BeforeSaveChanges()AfterSaveChanges()两个方法:

 1//BeforeSaveChanges
 2public virtual Task BeforeSaveChanges()
 3{
 4    ChangeTracker.DetectChanges();
 5    _auditEntries = new List<AuditEntry>();
 6    foreach (var entityEntry in ChangeTracker.Entries())
 7    {
 8        if (entityEntry.State == EntityState.Detached 
 9            || entityEntry.State == EntityState.Unchanged)
10            continue;
11        if (entityEntry.Entity.GetType() == typeof(AuditLog))
12            continue;
13        if (_auditConfig.EntityFilters.Any(x => x(entityEntry)))
14            continue;
15
16        var auditEntry = new AuditEntry(entityEntry, _auditConfig);
17        _auditEntries.Add(auditEntry);
18    }
19
20    return Task.CompletedTask;
21}
22
23//AfterSaveChanges
24public virtual Task AfterSaveChanges()
25{
26    if (_auditEntries == null || !_auditEntries.Any())
27        return Task.CompletedTask;
28
29    _auditEntries.ForEach(auditEntry => auditEntry.UpdateTemporaryProperties());
30
31    var auditLogs = _auditEntries.Select(x => x.AsAuditLog()).ToArray();
32    if (!_auditConfig.AuditStorages.Any())
33        _auditConfig.AuditStorages.Add(this);
34    _auditConfig.AuditStorages.ForEach(
35        auditStorage => auditStorage.SaveAuditLogs(auditLogs)
36    );
37
38    return Task.CompletedTask;
39}

可以注意到,我们会在SaveChanges()方法执行前,通过ChangeTracker.DetectChanges()方法显式地捕获“变化",这些“变化”会被存储到一个临时的列表中。而在SaveChanges()方法执行后,则会更新那些只有在数据提交后才可以获得的“临时”数据,最典型的例子是自增的 ID,在数据提交前,我们是无法获得真正的 ID 的。这个列表中的内容最终会通过AsAuditLog()方法进行转化。下面是AuditEntry中的部分代码片段:

 1//SetValuesCollection
 2private void SetValuesCollection(List<PropertyEntry> properties)
 3{
 4    foreach (var property in properties)
 5    {
 6        var propertyName = property.Metadata.GetColumnName();
 7        if (_auditConfig.PropertyFilters.Any(x => x(_entityEntry, property)))
 8            continue;
 9
10        switch (OperationType)
11        {
12            case OperationType.Created:
13                NewValues[propertyName] = property.CurrentValue;
14            break;
15            case OperationType.Updated:
16                if (_auditConfig.IsIgnoreSameValue 
17                    && property.OriginalValue.ToString() == property.CurrentValue.ToString())
18                    continue;
19                OldValues[propertyName] = property.OriginalValue;
20                NewValues[propertyName] = property.CurrentValue;
21            break;
22            case OperationType.Deleted:
23                OldValues[propertyName] = property.OriginalValue;
24            break;
25        }
26    };
27}
28
29//AsAuditLog
30public AuditLog AsAuditLog()
31{
32    return new AuditLog()
33    {
34        Id = Guid.NewGuid().ToString("N"),
35        TableName = TableName,
36        CreatedBy = string.Empty,
37        CreatedDate = DateTime.Now,
38        NewValues = NewValues.Any() ? JsonConvert.SerializeObject(NewValues) : null,
39        OldValues = OldValues.Any() ? JsonConvert.SerializeObject(OldValues) : null,
40        ExtraData = ExtraData.Any() ? JsonConvert.SerializeObject(ExtraData) : null,
41        OperationType = (int)OperationType
42    };
43}

在此基础上,我们可以编写我们实际的 DbContext,这里以 CustomerContext 为例,当我们向其中添加、修改和删除 Customer 的时候,就会触发审计相关的逻辑,默认情况下,审计产生的数据 AuditLog 和 Customer 在同一个数据库上下文中,当然,我们可以通过注入 IAuditStore 来实现更精细的控制,例如,可以将审计日志输入到文本文件,甚至是 Mongodb 这样的非关系型数据库里,因为有依赖注入的存在,这些实现起来会非常的简单!

 1//注入AuditLog配置
 2services.AddAuditLog(config => 
 3    config
 4    .IgnoreTable<AuditLog>()
 5    .IgnoreProperty<AuditLog>(x => x.CreatedDate)
 6    .WithExtraData("Tags", ".NET Core")
 7    .WithStorage<FileAuditStorage>()
 8    .WithStorage<MongoAuditStorage>()
 9);
10
11//注入DbContext
12services.AddDbContext<CustomerContext>(options =>
13    options.UseSqlServer(Configuration.GetConnectionString("DefaultConnection")));
14
15//像平时一样使用EF
16var entity = _context.Customer.Where(x => x.Id == customer.Id).FirstOrDefault();
17entity.Name = customer.Name;
18entity.Email = customer.Email;
19entity.Address = customer.Address;
20entity.Tel = customer.Tel;
21_context.Customer.Update(entity);
22await _context.SaveChangesAsync();

下面是最终生成的审计日志信息:

审计日志表展示
审计日志表展示

Castle 动态代理

而对于像 Dapper 这种轻量级的 ORM,它本身没有类似 EF/EF Core 的 ChangeTracker 的设计,如果我们在项目中使用 Dapper,并且希望实现审计的相关功能,直观上看就会有一点困难。其实,平时在混合使用 EF/Dapper 的过程中,经常遇到的问题就是,如何确保传统的 ADO.NET 和 EF 在一个数据库事务中,如何确保 Dapper 和 EF 在一个数据库事务中等等。此时,我们就需要一点抽象,首先去实现一个 Dapper 的仓储模式,然后再借助 Castle 这类动态代理库实现对接口的拦截。这里以 Dapper 的扩展库 Dapper.Contrib 为例。首先,我们定义一个仓储接口 IRepository:

 1public interface  IRepository
 2{
 3    TEntity GetByID<TEntity>(object id) where TEntity : class;
 4
 5    TEntity GetByKeys<TEntity>(object keys) where TEntity : class;
 6
 7    TEntity QueryFirst<TEntity>(string sql, object param) where TEntity : class;
 8
 9    TEntity QuerySingle<TEntity>(string sql, object param) where TEntity : class;
10
11    [AuditLog(OperationType.Created)]
12    void Insert<TEntity>(params TEntity[] entities) where TEntity : class;
13
14    [AuditLog(OperationType.Updated)]
15    void Update<TEntity>(params TEntity[] entities) where TEntity : class;
16
17    [AuditLog(OperationType.Deleted)]
18    void Delete<TEntity>(params TEntity[] entities) where TEntity : class;
19
20    void Delete<TEntity>(params object[] ids) where TEntity : class;
21
22    IEnumerable<TEntity> GetByQuery<TEntity>(Expression<Func<TEntity,bool>> exps) where TEntity : class;
23
24    IEnumerable<TEntity> GetByQuery<TEntity>(string sql, object param) where TEntity : class;
25
26    IEnumerable<TEntity> GetAll<TEntity>() where TEntity : class;
27}

接下来,我们就可以在拦截器中实现数据审计功能,因为 Dapper 本身没有 ChangeTracker,所以,我们必须要在先从数据库中查出来 OldValue,所以,实际效率应该并不会特别高,这里权当做为大家扩展思路吧!

 1public class AuditLogInterceptor : IInterceptor
 2{
 3    public void Intercept(IInvocation invocation)
 4    {
 5        var repository = invocation.Proxy as IRepository;
 6        var entityType = GetEntityType(invocation);
 7        var tableName = GetTableName(entityType);
 8        var tableIdProperty = entityType.GetProperty("Id");
 9        var auditLogAttrs = invocation.Method.GetCustomAttributes(typeof(AuditLogAttribute), false);
10        if (auditLogAttrs == null || auditLogAttrs.Length == 0 || entityType == typeof(AuditLog))
11        {
12            invocation.Proceed();
13            return;
14        }
15
16        var auditLogAttr = (auditLogAttrs as AuditLogAttribute[])[0];
17        var auditLogs = new List<AuditLog>();
18        switch (auditLogAttr.OperationType)
19        {
20            case Domain.OperationType.Created:
21                auditLogs = GetAddedAuditLogs(invocation, tableName);
22            break;
23            case Domain.OperationType.Updated:
24                auditLogs = GetUpdatedAuditLogs(invocation, tableName, entityType, 
25                    tableIdProperty, repository);
26            break;
27            case Domain.OperationType.Deleted:
28                auditLogs = GetDeletedAuditLogs(invocation, tableName);
29            break;
30        }
31            
32        invocation.Proceed();
33        repository.Insert<AuditLog>(auditLogs.ToArray());
34    }
35}

同样地,这里需要需要使用Autofac将其注册到 IoC 容器中:

1builder.RegisterType<DapperRepository>().As<IRepository>()
2    .InterceptedBy(typeof(AuditLogInterceptor))
3    .EnableInterfaceInterceptors();
4builder.RegisterType<AuditLogInterceptor>();

思路延伸:领域事件

最近这段时间,对于数据同步这类“需求”略有感触,譬如某种单据在两个互为上下游的系统里流转,譬如不同系统间实时地对基础资料进行同步等。这类需求可能会通过ETLDBLink这类“数据库”手段实现,亦有可能是通过互相调用 API 的方式实现,再者无非是通过数据库实现类似消息队列的功能……而我个人,更推崇通过事件来处理,因为它更接近人类思考的本质,希望在适当的时机来“通知”对方,而论询实际上是一种相当低效的沟通方式。一个订单被创建,一条记录被修改,本质上都是一个特定事件,而在业务上对此感兴趣的任何第三方,都可以去订阅这个事件,这就是事件驱动的思想。

领域事件
领域事件

我拜读了几篇关于“领域驱动设计(DDD)”文章,了解到 DDD 中有领域事件和集成事件的概念。最直接的体会就是,DDD 是主张“充血模型”的,它把事件附加到实体上,最大的好处就是,可以让“发送(Dispatch)”事件的代码,集中地放在一个地方。而我们现在的业务代码,基本是高度耦合的,每次去添加一个事件的时候,最担心地就是遗漏了某个地方。按照 DDD 的思想,实现领域事件,最常用的伎俩是重写 DbContext 的 SaveChanges()方法,或者在 EF 中去指定 DbContext 的 Complate 事件。这里同样借助了 ChangeTracker 来实现:

1public class OrderContext : DbContext
2{
3    public async Task<bool> SaveChangesAsync(CancellationToken cancellationToken = default(CancellationToken))
4    {
5        var aggregateRoots = dbContext.ChangeTracker.Entries().ToList();
6        await _eventDispatcher.DispatchAsync(aggregateRoots,cancellationToken);
7        var result = await base.SaveChangesAsync();
8    }
9}

其中,_eventDispatcher作为事件分发器来分发事件,它实现了IEventDispatcher接口。相对应地,事件订阅者需要实现IDomainEventHandler接口。如果是最简单的进程内通信,那么你需要一个容器来管理IDomainEventIDomainEventHandler间的关系;而如果是不同微服务间的通信,那么你需要引入RabbitMQ或者kafka这类消息队列中间件。

 1public interface IDomainEvent
 2{
 3
 4}
 5
 6public interface IDomainEventHandler<in TDomainEvent>
 7        where TDomainEvent : IDomainEvent
 8{
 9    Task HandleAysnc(TDomainEvent @event, CancellationToken cancellationToken = default);
10}
11
12public interface IEventDispatcher
13{
14    Task DispatchAsync<TDomainEvent>(
15        TDomainEvent @event,
16        CancellationToken cancellationToken = default) where TDomainEvent :IDomainEvent;
17}

所以,你现在问我怎么样做数据同步好,我一定会说,通过事件来处理。因为这样,每一条数据的新增、更新、删除,都可以事件的形式发布出去,而关心这些数据的下游系统,则只需要订阅这些事件,该干嘛好嘛,何乐而不为呢?搞什么中间表,打什么标记,用数据库一遍遍地实现消息队列有意思吗?同样地,你会意识到,仓储模式,哪怕 ORM 换成 Dapper,我们一样可以去发布这些事件,增量同步自然是要比全量同步优雅而且高效的。最重要的是,程序员再不需要到处找地方埋点了,你看我博客更新频率这么低,不就是因为这些事情浪费了时间吗(逃?因为,全量 + 实时同步就是一个非常愚蠢的决定。

本文小结

本文分别针对EF CoreDapper实现了数据库审计的功能。对于前者,主要是通过重写 DbContext 的SaveChanges()方法来实现,而EFEF Core中的ChangeTracker则提供了一种获取数据库表记录变化前后值的能力。而对于后者,主要是实现了Dapper的仓储模式,在此基础上结合Castle的动态代理功能,对仓储接口进行拦截,以此实现审计日志的记录功能。整体来看,后者对代码的侵入性要更小一点,理论上我们可以实现EFEF Core的仓储模式,这样两者在实现上会更接近一点,当然,更直接的方案是去拦截SaveChanges()方法,这和我们使用继承的目的是一样的,由于 Dapper 本身没有ChangeTracker,所以,在处理Update()相关的仓储接口时,都需要先查询一次数据库,这一点是这个方案里最大的短板。而顺着这个方案扩展下去,我们同样可以挖掘出一点DDD领域事件的意味,这就变得很有意思了,不是吗?这篇博客就先写到这里吧……再见

Built with Hugo
Theme Stack designed by Jimmy