Method proxyRequest

proxy.go:285–352

proxyRequest proxies the given request to clickhouse and sends response to rw. srw is required only for setting non-200 status codes on timeouts or on client connection disconnects.

(s *scope, rw ResponseWriterWithCode, srw *statResponseWriter, req *http.Request)

// srw is required only for setting non-200 status codes on timeouts
// or on client connection disconnects.
func (rp *reverseProxy) proxyRequest(s *scope, rw ResponseWriterWithCode, srw *statResponseWriter, req *http.Request) {
	// wrap body into cachedReadCloser, so we could obtain the original
	// request on error.
	if _, ok := req.Body.(*cachedReadCloser); !ok {
		req.Body = &cachedReadCloser{
			ReadCloser: req.Body,
		}
	}

	timeout, timeoutErrMsg := s.getTimeoutWithErrMsg()
	ctx := context.Background()
	if timeout > 0 {
		var cancel context.CancelFunc
		ctx, cancel = context.WithTimeout(ctx, timeout)
		defer cancel()
	}

	// Cancel the ctx if client closes the remote connection,
	// so the proxied query may be killed instantly.
	ctx, ctxCancel := listenToCloseNotify(ctx, rw)
	defer ctxCancel()

	req = req.WithContext(ctx)

	startTime := time.Now()

	executeDuration, err := executeWithRetry(ctx, s, s.cluster.retryNumber, rp.rp.ServeHTTP, rw, srw, req, func(duration float64) {
		proxiedResponseDuration.With(s.labels).Observe(duration)
	}, func(labels prometheus.Labels) { retryRequest.With(labels).Inc() })

	statusCodesClickhouse.With(
		prometheus.Labels{
			"cluster":      s.cluster.name,
			"replica":      s.host.ReplicaName(),
			"cluster_node": s.host.Host(),
			"code":         strconv.Itoa(rw.StatusCode()),
		},
	).Inc()
	switch {
	case err == nil:
		return
	case errors.Is(err, context.Canceled):
		canceledRequest.With(s.labels).Inc()

		q := getQuerySnippet(req)
		log.Debugf("%s: remote client closed the connection in %s; query: %q", s, time.Since(startTime), q)
		if err := s.killQuery(); err != nil {
			log.Errorf("%s: cannot kill query: %s; query: %q", s, err, q)
		}
		srw.statusCode = 499 // See https://httpstatuses.com/499 .
	case errors.Is(err, context.DeadlineExceeded):
		timeoutRequest.With(s.labels).Inc()

		// Penalize host with the timed out query, because it may be overloaded.
		s.host.Penalize()

		q := getQuerySnippet(req)
		log.Debugf("%s: query timeout in %f; query: %q", s, executeDuration, q)