| 515 | if err != nil { |
| 516 | originFetchesTotal.WithLabelValues(originFetchOutcome(err)).Inc() |
| 517 | return fetchResult{}, err |
| 518 | } |
| 519 | originFetchesTotal.WithLabelValues("success").Inc() |
| 520 | cacheBytesServed.WithLabelValues("s3").Add(float64(size)) |
| 521 | return fetchResult{size: size, contentType: ct, source: "miss"}, nil |
| 522 | }) |
| 523 | } |
| 524 | |
| 525 | // fetchFromPeer pulls cacheKey's body from a peer that has it (flight=false) |
| 526 | // or is mid-flight filling it (flight=true; the peer's /cache/get then blocks |
| 527 | // briefly on its fill instead of 404ing, so we read the bytes the peer just |
| 528 | // fetched rather than re-fetching them from the origin ourselves). |
| 529 | func (p *CacheProxy) fetchFromPeer(holder string, flight bool, cacheKey string, r *http.Request) (fetchResult, bool) { |
| 530 | n, ok := p.peers.FetchFromPeer(r.Context(), holder, cacheKey, flight, func(body io.Reader) (int64, error) { |
| 531 | return p.store.PutStream(cacheKey, body) |
| 532 | }) |