流量回放工具之GoReplay output_http 源码分析

-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.

// 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
}

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 <- &response{resp, 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