Merge pull request #137 from MichaHoffmann/mhoffmann/timing-reader-ig…

…nores-initial-get-range-call fix: timing reader ignores initial get_range
thanos-io · Sep 13, 2024 · d29b6ed · d29b6ed
2 parents 7adf08b + 1a62b9e
commit d29b6ed
Show file tree

Hide file tree

Showing 2 changed files with 19 additions and 7 deletions.
diff --git a/objstore.go b/objstore.go
@@ -566,14 +566,18 @@ func (b *metricBucket) Get(ctx context.Context, name string) (io.ReadCloser, err
 	const op = OpGet
 	b.metrics.ops.WithLabelValues(op).Inc()
 
+	start := time.Now()
+
 	rc, err := b.bkt.Get(ctx, name)
 	if err != nil {
 		if !b.metrics.isOpFailureExpected(err) && ctx.Err() != context.Canceled {
 			b.metrics.opsFailures.WithLabelValues(op).Inc()
 		}
+		b.metrics.opsDuration.WithLabelValues(op).Observe(float64(time.Since(start)))
 		return nil, err
 	}
 	return newTimingReader(
+		start,
 		rc,
 		true,
 		op,
@@ -589,14 +593,18 @@ func (b *metricBucket) GetRange(ctx context.Context, name string, off, length in
 	const op = OpGetRange
 	b.metrics.ops.WithLabelValues(op).Inc()
 
+	start := time.Now()
+
 	rc, err := b.bkt.GetRange(ctx, name, off, length)
 	if err != nil {
 		if !b.metrics.isOpFailureExpected(err) && ctx.Err() != context.Canceled {
 			b.metrics.opsFailures.WithLabelValues(op).Inc()
 		}
+		b.metrics.opsDuration.WithLabelValues(op).Observe(float64(time.Since(start)))
 		return nil, err
 	}
 	return newTimingReader(
+		start,
 		rc,
 		true,
 		op,
@@ -628,7 +636,10 @@ func (b *metricBucket) Upload(ctx context.Context, name string, r io.Reader) err
 	const op = OpUpload
 	b.metrics.ops.WithLabelValues(op).Inc()
 
+	start := time.Now()
+
 	trc := newTimingReader(
+		start,
 		r,
 		false,
 		op,
@@ -705,7 +716,7 @@ type timingReader struct {
 	transferredBytes  *prometheus.HistogramVec
 }
 
-func newTimingReader(r io.Reader, closeReader bool, op string, dur *prometheus.HistogramVec, failed *prometheus.CounterVec, isFailureExpected IsOpFailureExpectedFunc, fetchedBytes *prometheus.CounterVec, transferredBytes *prometheus.HistogramVec) io.ReadCloser {
+func newTimingReader(start time.Time, r io.Reader, closeReader bool, op string, dur *prometheus.HistogramVec, failed *prometheus.CounterVec, isFailureExpected IsOpFailureExpectedFunc, fetchedBytes *prometheus.CounterVec, transferredBytes *prometheus.HistogramVec) io.ReadCloser {
 	// Initialize the metrics with 0.
 	dur.WithLabelValues(op)
 	failed.WithLabelValues(op)
@@ -716,7 +727,7 @@ func newTimingReader(r io.Reader, closeReader bool, op string, dur *prometheus.H
 		closeReader:       closeReader,
 		objSize:           objSize,
 		objSizeErr:        objSizeErr,
-		start:             time.Now(),
+		start:             start,
 		op:                op,
 		duration:          dur,
 		failed:            failed,

diff --git a/objstore_test.go b/objstore_test.go
@@ -11,6 +11,7 @@ import (
 	"path/filepath"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/efficientgo/core/testutil"
 	"github.com/go-kit/log"
@@ -412,7 +413,7 @@ func TestDownloadUploadDirConcurrency(t *testing.T) {
 func TestTimingReader(t *testing.T) {
 	m := WrapWithMetrics(NewInMemBucket(), nil, "")
 	r := bytes.NewReader([]byte("hello world"))
-	tr := newTimingReader(r, true, OpGet, m.metrics.opsDuration, m.metrics.opsFailures, func(err error) bool {
+	tr := newTimingReader(time.Now(), r, true, OpGet, m.metrics.opsDuration, m.metrics.opsFailures, func(err error) bool {
 		return false
 	}, m.metrics.opsFetchedBytes, m.metrics.opsTransferredBytes)
 
@@ -447,7 +448,7 @@ func TestTimingReader_ExpectedError(t *testing.T) {
 
 	m := WrapWithMetrics(NewInMemBucket(), nil, "")
 	r := dummyReader{readerErr}
-	tr := newTimingReader(r, true, OpGet, m.metrics.opsDuration, m.metrics.opsFailures, func(err error) bool { return errors.Is(err, readerErr) }, m.metrics.opsFetchedBytes, m.metrics.opsTransferredBytes)
+	tr := newTimingReader(time.Now(), r, true, OpGet, m.metrics.opsDuration, m.metrics.opsFailures, func(err error) bool { return errors.Is(err, readerErr) }, m.metrics.opsFetchedBytes, m.metrics.opsTransferredBytes)
 
 	buf := make([]byte, 1)
 	_, err := io.ReadFull(tr, buf)
@@ -461,7 +462,7 @@ func TestTimingReader_UnexpectedError(t *testing.T) {
 
 	m := WrapWithMetrics(NewInMemBucket(), nil, "")
 	r := dummyReader{readerErr}
-	tr := newTimingReader(r, true, OpGet, m.metrics.opsDuration, m.metrics.opsFailures, func(err error) bool { return false }, m.metrics.opsFetchedBytes, m.metrics.opsTransferredBytes)
+	tr := newTimingReader(time.Now(), r, true, OpGet, m.metrics.opsDuration, m.metrics.opsFailures, func(err error) bool { return false }, m.metrics.opsFetchedBytes, m.metrics.opsTransferredBytes)
 
 	buf := make([]byte, 1)
 	_, err := io.ReadFull(tr, buf)
@@ -476,7 +477,7 @@ func TestTimingReader_ContextCancellation(t *testing.T) {
 
 	m := WrapWithMetrics(NewInMemBucket(), nil, "")
 	r := dummyReader{ctx.Err()}
-	tr := newTimingReader(r, true, OpGet, m.metrics.opsDuration, m.metrics.opsFailures, func(err error) bool { return false }, m.metrics.opsFetchedBytes, m.metrics.opsTransferredBytes)
+	tr := newTimingReader(time.Now(), r, true, OpGet, m.metrics.opsDuration, m.metrics.opsFailures, func(err error) bool { return false }, m.metrics.opsFetchedBytes, m.metrics.opsTransferredBytes)
 
 	buf := make([]byte, 1)
 	_, err := io.ReadFull(tr, buf)
@@ -506,7 +507,7 @@ func TestTimingReader_ShouldCorrectlyWrapFile(t *testing.T) {
 	})
 
 	m := WrapWithMetrics(NewInMemBucket(), nil, "")
-	r := newTimingReader(file, true, "", m.metrics.opsDuration, m.metrics.opsFailures, func(err error) bool {
+	r := newTimingReader(time.Now(), file, true, "", m.metrics.opsDuration, m.metrics.opsFailures, func(err error) bool {
 		return false
 	}, m.metrics.opsFetchedBytes, m.metrics.opsTransferredBytes)