流量回放工具之GoReplay output-http 源码分析

Posted 2022-01-06 zuozewei

文章目录

• • 前言
  • 参数说明
  • HTTP 输出工作数量
  • HTTP 输出队列
  • 代码逻辑调用图

前言

GoReplay 对数据流的抽象出了两个概念，即用 输入（input ） 和 输出（output ） 来表示数据来源与去向，统称为 plugin，用介于输入和输出模块之间的中间件实现拓展机制。
​
output_http.go：主要是 HTTP 输出的插件，实现 HTTP 协议， 实现 io.Writer 接口，最后根据配置注册到 Plugin.outputs 队列里。

参数说明

-output-http value  //转发进入的请求到一个http地址上
        Forwards incoming requests to given http address.
                # Redirect all incoming requests to staging.com address 
                gor --input-raw :80 --output-http http://staging.com
  -output-http-elasticsearch string  //把请求和响应状态发送到 ElasticSearch
        Send request and response stats to ElasticSearch:
                gor --input-raw :8080 --output-http staging.com --output-http-elasticsearch 'es_host:api_port/index_name'
  -output-http-queue-len int //http输出队列大小
        Number of requests that can be queued for output, if all workers are busy. default = 1000 (default 1000)
  -output-http-redirects int  // 设置多少次重定向被允许，默认忽略
        Enable how often redirects should be followed.
  -output-http-response-buffer value  //最大接收响应大小(缓冲区)
        HTTP response buffer size, all data after this size will be discarded.
  -output-http-skip-verify
        Don't verify hostname on TLS secure connection.
  -output-http-stats  //每5秒钟输出一次输出队列的状态
        Report http output queue stats to console every N milliseconds. See output-http-stats-ms
  -output-http-stats-ms int
        Report http output queue stats to console every N milliseconds. default: 5000 (default 5000)
  -output-http-timeout duration  //指定 http 的 request/response 超时时间，默认是 5 秒 
        Specify HTTP request/response timeout. By default 5s. Example: --output-http-timeout 30s (default 5s)
  -output-http-track-response
        If turned on, HTTP output responses will be set to all outputs like stdout, file and etc.
  -output-http-worker-timeout duration
        Duration to rollback idle workers. (default 2s)
  -output-http-workers int  //gor默认是动态的扩展工作者数量，你也可以指定固定数量的工作者
        Gor uses dynamic worker scaling. Enter a number to set a maximum number of workers. default = 0 = unlimited.
  -output-http-workers-min int
        Gor uses dynamic worker scaling. Enter a number to set a minimum number of workers. default = 1.

默认情况下，Gor 创建一个动态工作池：
它从 10 开始，并在 HTTP 输出队列长度大于 10 时创建更多的 HTTP 输出协程。创建的协程数量（N）等于该工作时间的队列长度检查并发现其长度大于10.每次将消息写入 HTTP 输出队列时都检查队列长度。在产生 N 个协程的请求得到满足之前，不会再有协程创建。如果动态协程池当时不能处理消息，它将睡眠 100 毫秒。如果动态工作协程无法处理消息2秒钟，则会死亡。可以使用 --output-http-workers=20 选项指定固定数量的协程。

HTTP 输出工作数量

NewHTTPOutput 默认情况：

// NewHTTPOutput constructor for HTTPOutput
// Initialize workers
func NewHTTPOutput(address string, config *HTTPOutputConfig) PluginReadWriter 
	o := new(HTTPOutput)
	var err error
	config.url, err = url.Parse(address)
	if err != nil 
		log.Fatal(fmt.Sprintf("[OUTPUT-HTTP] parse HTTP output URL error[%q]", err))
	
	if config.url.Scheme == "" 
		config.url.Scheme = "http"
	
	config.rawURL = config.url.String()
	if config.Timeout < time.Millisecond*100 
		config.Timeout = time.Second
	
	if config.BufferSize <= 0 
		config.BufferSize = 100 * 1024 // 100kb
	
	if config.WorkersMin <= 0 
		config.WorkersMin = 1
	
	if config.WorkersMin > 1000 
		config.WorkersMin = 1000
	
	if config.WorkersMax <= 0 
		config.WorkersMax = math.MaxInt32 // idealy so large
	
	if config.WorkersMax < config.WorkersMin 
		config.WorkersMax = config.WorkersMin
	
	if config.QueueLen <= 0 
		config.QueueLen = 1000
	
	if config.RedirectLimit < 0 
		config.RedirectLimit = 0
	
	if config.WorkerTimeout <= 0 
		config.WorkerTimeout = time.Second * 2
	
	o.config = config
	o.stop = make(chan bool)
	//是否收集统计信息，统计输出间隔是多少
	if o.config.Stats 
		o.queueStats = NewGorStat("output_http", o.config.StatsMs)
	

	o.queue = make(chan *Message, o.config.QueueLen)
	if o.config.TrackResponses 
		o.responses = make(chan *response, o.config.QueueLen)
	
	// it should not be buffered to avoid races
	o.stopWorker = make(chan struct)

	if o.config.ElasticSearch != "" 
		o.elasticSearch = new(ESPlugin)
		o.elasticSearch.Init(o.config.ElasticSearch)
	
	o.client = NewHTTPClient(o.config)
	o.activeWorkers += int32(o.config.WorkersMin)
	for i := 0; i < o.config.WorkersMin; i++ 
		go o.startWorker()
	
	go o.workerMaster()
	return o

配置后启动 httpclient：

o.client = NewHTTPClient(o.config)
	o.activeWorkers += int32(o.config.WorkersMin)
	for i := 0; i < o.config.WorkersMin; i++ 
		go o.startWorker()
	

启动多个发送协程：

func (o *HTTPOutput) startWorker() 
	for 
		select 
		case <-o.stopWorker:
			return
		case msg := <-o.queue:
			o.sendRequest(o.client, msg)
		
	

​

执行发送：

​

func (o *HTTPOutput) sendRequest(client *HTTPClient, msg *Message) 
	if !isRequestPayload(msg.Meta) 
		return
	

	uuid := payloadID(msg.Meta)
	start := time.Now()
	resp, err := client.Send(msg.Data)
	stop := time.Now()

	if err != nil 
		Debug(1, fmt.Sprintf("[HTTP-OUTPUT] error when sending: %q", err))
		return
	
	if resp == nil 
		return
	

	if o.config.TrackResponses 
		o.responses <- &responseresp, uuid, start.UnixNano(), stop.UnixNano() - start.UnixNano()
	

	if o.elasticSearch != nil 
		o.elasticSearch.ResponseAnalyze(msg.Data, resp, start, stop)
	

发送细节，各种配置生效点：

// Send sends an http request using client create by NewHTTPClient
func (c *HTTPClient) Send(data []byte) ([]byte, error) 
	var req *http.Request
	var resp *http.Response
	var err error

	req, err = http.ReadRequest(bufio.NewReader(bytes.NewReader(data)))
	if err != nil 
		return nil, err
	
	// we don't send CONNECT or OPTIONS request
	if req.Method == http.MethodConnect 
		return nil, nil
	

	if !c.config.OriginalHost 
		req.Host = c.config.url.Host
	

	// fix #862
	if c.config.url.Path == "" && c.config.url.RawQuery == "" 
		req.URL.Scheme = c.config.url.Scheme
		req.URL.Host = c.config.url.Host
	 else 
		req.URL = c.config.url
	

	// force connection to not be closed, which can affect the global client
	req.Close = false
	// it's an error if this is not equal to empty string
	req.RequestURI = ""

	resp, err = c.Client.Do(req)
	if err != nil 
		return nil, err
	
	if c.config.TrackResponses 
		return httputil.DumpResponse(resp, true)
	
	_ = resp.Body.Close()
	return nil, nil

HTTP 输出队列

​

队列用在哪儿呢？

代码逻辑调用图

相关系列：

• 性能工具之常见流量复制工具
• 流量回放工具之 goreplay 安装及初级使用
• 流量回放工具之 goreplay 核心源码分析（VIP）
• 流量回放工具之 GoReplay output-http-stats（HTTP请求统计） 源码分析
• 流量回放工具之GoReplay output-http 源码分析

​