nolint:cyclop //TODO refactor this method, most likely requires some work.
(s *scope, srw *statResponseWriter, req *http.Request, origParams url.Values, q []byte)
| 376 | |
| 377 | //nolint:cyclop //TODO refactor this method, most likely requires some work. |
| 378 | func (rp *reverseProxy) serveFromCache(s *scope, srw *statResponseWriter, req *http.Request, origParams url.Values, q []byte) { |
| 379 | labels := makeCacheLabels(s) |
| 380 | key := newCacheKey(s, origParams, q, req) |
| 381 | |
| 382 | startTime := time.Now() |
| 383 | userCache := s.user.cache |
| 384 | // Try to serve from cache |
| 385 | cachedData, err := userCache.Get(key) |
| 386 | if err == nil { |
| 387 | // The response has been successfully served from cache. |
| 388 | defer cachedData.Data.Close() |
| 389 | cacheHit.With(labels).Inc() |
| 390 | cachedResponseDuration.With(labels).Observe(time.Since(startTime).Seconds()) |
| 391 | log.Debugf("%s: cache hit", s) |
| 392 | _ = RespondWithData(srw, cachedData.Data, cachedData.ContentMetadata, cachedData.Ttl, XCacheHit, http.StatusOK, labels) |
| 393 | return |
| 394 | } |
| 395 | // Await for potential result from concurrent query |
| 396 | transactionStatus, err := userCache.AwaitForConcurrentTransaction(key) |
| 397 | if err != nil { |
| 398 | // log and continue processing |
| 399 | log.Errorf("failed to await for concurrent transaction due to: %v", err) |
| 400 | } else { |
| 401 | if transactionStatus.State.IsCompleted() { |
| 402 | cachedData, err := userCache.Get(key) |
| 403 | if err == nil { |
| 404 | defer cachedData.Data.Close() |
| 405 | _ = RespondWithData(srw, cachedData.Data, cachedData.ContentMetadata, cachedData.Ttl, XCacheHit, http.StatusOK, labels) |
| 406 | cacheHitFromConcurrentQueries.With(labels).Inc() |
| 407 | log.Debugf("%s: cache hit after awaiting concurrent query", s) |
| 408 | return |
| 409 | } else { |
| 410 | cacheMissFromConcurrentQueries.With(labels).Inc() |
| 411 | log.Debugf("%s: cache miss after awaiting concurrent query", s) |
| 412 | } |
| 413 | } else if transactionStatus.State.IsFailed() { |
| 414 | respondWith(srw, fmt.Errorf("%v", transactionStatus.FailReason), http.StatusInternalServerError) |
| 415 | return |
| 416 | } |
| 417 | } |
| 418 | |
| 419 | // The response wasn't found in the cache. |
| 420 | // Request it from clickhouse. |
| 421 | tmpFileRespWriter, err := cache.NewTmpFileResponseWriter(srw, os.TempDir()) |
| 422 | if err != nil { |
| 423 | err = fmt.Errorf("%s: %w; query: %q", s, err, q) |
| 424 | respondWith(srw, err, http.StatusInternalServerError) |
| 425 | return |
| 426 | } |
| 427 | defer tmpFileRespWriter.Close() |
| 428 | |
| 429 | // Initialise transaction |
| 430 | err = userCache.Create(key) |
| 431 | if err != nil { |
| 432 | log.Errorf("%s: %s; query: %q - failed to register transaction", s, err, q) |
| 433 | } |
| 434 | |
| 435 | // proxy request and capture response along with headers to [[TmpFileResponseWriter]] |
no test coverage detected