| 84 | } |
| 85 | |
| 86 | func (rp *reverseProxy) ServeHTTP(rw http.ResponseWriter, req *http.Request) { |
| 87 | startTime := time.Now() |
| 88 | s, status, err := rp.getScope(req) |
| 89 | if err != nil { |
| 90 | q := getQuerySnippet(req) |
| 91 | err = fmt.Errorf("%q: %w; query: %q", req.RemoteAddr, err, q) |
| 92 | respondWith(rw, err, status) |
| 93 | return |
| 94 | } |
| 95 | |
| 96 | // WARNING: don't use s.labels before s.incQueued, |
| 97 | // since `replica` and `cluster_node` may change inside incQueued. |
| 98 | if err := s.incQueued(); err != nil { |
| 99 | limitExcess.With(s.labels).Inc() |
| 100 | q := getQuerySnippet(req) |
| 101 | err = fmt.Errorf("%s: %w; query: %q", s, err, q) |
| 102 | respondWith(rw, err, http.StatusTooManyRequests) |
| 103 | return |
| 104 | } |
| 105 | defer s.dec() |
| 106 | |
| 107 | log.Debugf("%s: request start", s) |
| 108 | requestSum.With(s.labels).Inc() |
| 109 | |
| 110 | req.Body = &statReadCloser{ |
| 111 | ReadCloser: req.Body, |
| 112 | bytesRead: requestBodyBytes.With(s.labels), |
| 113 | } |
| 114 | srw := &statResponseWriter{ |
| 115 | ResponseWriter: rw, |
| 116 | bytesWritten: responseBodyBytes.With(s.labels), |
| 117 | } |
| 118 | |
| 119 | req, origParams := s.decorateRequest(req) |
| 120 | |
| 121 | // wrap body into cachedReadCloser, so we could obtain the original |
| 122 | // request on error. |
| 123 | req.Body = &cachedReadCloser{ |
| 124 | ReadCloser: req.Body, |
| 125 | } |
| 126 | |
| 127 | // publish session_id if needed |
| 128 | if s.sessionId != "" { |
| 129 | rw.Header().Set("X-ClickHouse-Server-Session-Id", s.sessionId) |
| 130 | } |
| 131 | |
| 132 | q, shouldReturnFromCache, err := shouldRespondFromCache(s, origParams, req) |
| 133 | if err != nil { |
| 134 | respondWith(srw, err, http.StatusBadRequest) |
| 135 | return |
| 136 | } |
| 137 | |
| 138 | if shouldReturnFromCache { |
| 139 | rp.serveFromCache(s, srw, req, origParams, q) |
| 140 | } else { |
| 141 | rp.proxyRequest(s, srw, srw, req) |
| 142 | } |
| 143 | |