登录/ 注册
 找回密码
 立即注册
关闭

CSDN热搜

豆瓜网 精品问答 技术交流 资源下载
返回列表 发新帖
首页 › 业界区 › 业界 ›http流量镜像

http流量镜像

膏包 2025-7-5 06:58:32
http流量镜像

“流量镜像”是指将网络中的数据流量复制一份，并将这份复制流量发送到另一个目的地（如监控、分析或安全检测系统）。这项技术常用于网络安全、故障排查、业务灰度发布等场景。
主要应用场景

  • 安全监控与威胁检测
    将生产环境的流量镜像到安全分析设备（如IDS/IPS），用于实时监控和威胁检测。
  • 性能分析与故障排查
    将流量镜像到分析平台，对网络异常、延迟、丢包等问题进行实时排查和定位。
  • 灰度发布和A/B测试
    将真实用户流量镜像到新版本服务，进行灰度环境验证和兼容性测试，不影响真实用户体验。
  • 合规与审计
    对重要业务流量进行行为记录，以满足合规和审计要求。
VKProxy 目前只支持 http流量镜像, 需注意由于会再一次发送http 请求，请求body会临时暂存内存，所以无论内存还是请求延迟都会受到影响，特别body很大的请求
设置

大家可以在Metadata中设置缓存， 具体设置项如下

  • MirrorCluster
    镜像流量发送到的集群id
配置示例：
  1. {
  2.   "ReverseProxy": {
  3.     "Routes": {
  4.       "a": {
  5.         "Order": 0,
  6.         "Match": {
  7.             "Hosts": [ "api.com" ],
  8.             "Paths": [ "*" ]
  9.         },
  10.         "ClusterId": "apidemo",
  11.         "Metadata": {
  12.           "MirrorCluster": "apidemoMirror"
  13.         }
  14.       }
  15.     },
  16.     "Clusters": {
  17.       "apidemo": {
  18.         "LoadBalancingPolicy": "Hash",
  19.         "Metadata": {
  20.           "HashBy": "header",
  21.           "Key": "X-forwarded-For"
  22.         },
  23.         "Destinations": [
  24.           {
  25.             "Address": "https://xxx.lt"
  26.           }
  27.         ]
  28.       },
  29.       "apidemoMirror": {
  30.         "LoadBalancingPolicy": "Hash",
  31.         "Metadata": {
  32.           "HashBy": "header",
  33.           "Key": "X-forwarded-For"
  34.         },
  35.         "Destinations": [
  36.           {
  37.             "Address": "http://xxx.org/"
  38.           }
  39.         ]
  40.       }
  41.     }
  42.   }
  43. }
复制代码
具体实现

首先需要缓存body 内容，这里实现一个简单的 ReadBufferingStream
  1. public class ReadBufferingStream : Stream, IDisposable
  2. {
  3.     private readonly SparseBufferWriter<byte> bufferWriter;
  4.     protected Stream innerStream;
  6.     public ReadBufferingStream(Stream innerStream)
  7.     {
  8.         this.innerStream = innerStream;
  9.         bufferWriter = new SparseBufferWriter<byte>();
  10.     }
  12.     public override bool CanRead => innerStream.CanRead;
  14.     public override bool CanSeek => innerStream.CanSeek;
  16.     public override bool CanWrite => innerStream.CanWrite;
  18.     public override long Length => innerStream.Length;
  20.     public override long Position
  21.     {
  22.         get => innerStream.Position;
  23.         set => innerStream.Position = value;
  24.     }
  26.     public override int WriteTimeout
  27.     {
  28.         get => innerStream.WriteTimeout;
  29.         set => innerStream.WriteTimeout = value;
  30.     }
  32.     public Stream BufferingStream => bufferWriter.AsStream(true);
  34.     public override void Flush()
  35.     {
  36.         innerStream.Flush();
  37.     }
  39.     public override Task FlushAsync(CancellationToken cancellationToken)
  40.     {
  41.         return innerStream.FlushAsync(cancellationToken);
  42.     }
  44.     public override int Read(byte[] buffer, int offset, int count)
  45.     {
  46.         var res = innerStream.Read(buffer, offset, count);
  48.         // Zero-byte reads (where the passed in buffer has 0 length) can occur when using PipeReader, we don't want to accidentally complete the RequestBody logging in this case
  49.         if (count == 0)
  50.         {
  51.             return res;
  52.         }
  54.         bufferWriter.Write(buffer.AsSpan(offset, res));
  56.         return res;
  57.     }
  59.     public override async Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
  60.     {
  61.         var res = await innerStream.ReadAsync(buffer.AsMemory(offset, count), cancellationToken);
  63.         if (count == 0)
  64.         {
  65.             return res;
  66.         }
  68.         bufferWriter.Write(buffer.AsSpan(offset, res));
  70.         return res;
  71.     }
  73.     public override long Seek(long offset, SeekOrigin origin)
  74.     {
  75.         return innerStream.Seek(offset, origin);
  76.     }
  78.     public override void SetLength(long value)
  79.     {
  80.         innerStream.SetLength(value);
  81.     }
  83.     public override void Write(byte[] buffer, int offset, int count)
  84.     {
  85.         innerStream.Write(buffer, offset, count);
  86.     }
  88.     public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
  89.     {
  90.         return innerStream.WriteAsync(buffer, offset, count, cancellationToken);
  91.     }
  93.     public override ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default)
  94.     {
  95.         return innerStream.WriteAsync(buffer, cancellationToken);
  96.     }
  98.     public override void Write(ReadOnlySpan<byte> buffer)
  99.     {
  100.         innerStream.Write(buffer);
  101.     }
  103.     public override IAsyncResult BeginRead(byte[] buffer, int offset, int count, AsyncCallback? callback, object? state)
  104.     {
  105.         return innerStream.BeginRead(buffer, offset, count, callback, state);
  106.     }
  108.     public override IAsyncResult BeginWrite(byte[] buffer, int offset, int count, AsyncCallback? callback, object? state)
  109.     {
  110.         return innerStream.BeginWrite(buffer, offset, count, callback, state);
  111.     }
  113.     public override int EndRead(IAsyncResult asyncResult)
  114.     {
  115.         return innerStream.EndRead(asyncResult);
  116.     }
  118.     public override void EndWrite(IAsyncResult asyncResult)
  119.     {
  120.         innerStream.EndWrite(asyncResult);
  121.     }
  123.     public override async ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
  124.     {
  125.         var res = await innerStream.ReadAsync(buffer, cancellationToken);
  126.         if (buffer.IsEmpty)
  127.         {
  128.             return res;
  129.         }
  131.         bufferWriter.Write(buffer.Slice(0, res).Span);
  133.         return res;
  134.     }
  136.     public override ValueTask DisposeAsync()
  137.     {
  138.         return innerStream.DisposeAsync();
  139.     }
  141.     protected override void Dispose(bool disposing)
  142.     {
  143.         if (disposing)
  144.         {
  145.             bufferWriter.Dispose();
  146.         }
  147.     }
  148. }
复制代码
然后利用中间件进行镜像处理
  1. public class MirrorFunc : IHttpFunc
  2. {
  3.     private readonly IServiceProvider serviceProvider;
  4.     private readonly IHttpForwarder forwarder;
  5.     private readonly ILoadBalancingPolicyFactory loadBalancing;
  6.     private readonly IForwarderHttpClientFactory forwarderHttpClientFactory;
  7.     private readonly ProxyLogger logger;
  9.     public int Order => int.MinValue;
  11.     public MirrorFunc(IServiceProvider serviceProvider, IHttpForwarder forwarder, ILoadBalancingPolicyFactory loadBalancing, IForwarderHttpClientFactory forwarderHttpClientFactory, ProxyLogger logger)
  12.     {
  13.         this.serviceProvider = serviceProvider;
  14.         this.forwarder = forwarder;
  15.         this.loadBalancing = loadBalancing;
  16.         this.forwarderHttpClientFactory = forwarderHttpClientFactory;
  17.         this.logger = logger;
  18.     }
  20.     public RequestDelegate Create(RouteConfig config, RequestDelegate next)
  21.     {
  22.         if (config.Metadata == null || !config.Metadata.TryGetValue("MirrorCluster", out var mirrorCluster) || string.IsNullOrWhiteSpace(mirrorCluster)) return next;
  24.         return c => Mirror(c, mirrorCluster, next);
  25.     }
  27.     private async Task Mirror(HttpContext c, string mirrorCluster, RequestDelegate next)
  28.     {
  29.         var config = serviceProvider.GetRequiredService<IConfigSource<IProxyConfig>>();
  30.         if (config.CurrentSnapshot == null || config.CurrentSnapshot.Clusters == null || !config.CurrentSnapshot.Clusters.TryGetValue(mirrorCluster, out var cluster) || cluster == null)
  31.         {
  32.             await next(c);
  33.             return;
  34.         }
  36.         var originBody = c.Request.Body;
  37.         using var buffer = new ReadBufferingStream(originBody);
  38.         c.Request.Body = buffer;
  40.         try
  41.         {
  42.             await next(c);
  43.         }
  44.         finally
  45.         {
  46.             c.Request.Body = buffer.BufferingStream;
  47.             try
  48.             {
  49.                 var proxyFeature = c.Features.GetRequiredFeature<IReverseProxyFeature>();
  50.                 var origin = proxyFeature.SelectedDestination;
  51.                 var selectedDestination = loadBalancing.PickDestination(proxyFeature, cluster);
  52.                 proxyFeature.SelectedDestination = origin;
  53.                 if (selectedDestination != null)
  54.                 {
  55.                     cluster.InitHttp(forwarderHttpClientFactory);
  56.                     await forwarder.SendAsync(c, proxyFeature, selectedDestination, cluster, new NonHttpTransformer(proxyFeature.Route.Transformer));
  57.                 }
  58.             }
  59.             catch (Exception ex)
  60.             {
  61.                 logger.LogWarning(ex, "Mirror failed");
  62.             }
  63.             finally
  64.             {
  65.                 c.Request.Body = originBody;
  66.             }
  67.         }
  68.     }
  69. }
复制代码
所以说会再一次发送http 请求，请求body会临时暂存内存，所以无论内存还是请求延迟都会受到影响，特别body很大的请求
VKProxy 是使用c#开发的基于 Kestrel 实现 L4/L7的代理（感兴趣的同学烦请点个github小赞赞呢）

来源：豆瓜网用户自行投稿发布，如果侵权，请联系站长删除
http流量镜像

使用道具 举报

相关推荐
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册
发帖
膏包
2025-7-5 06:58:32

0

粉丝关注

10

主题发布

板块介绍填写区域，请于后台编辑
微信扫一扫