Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 72 additions & 0 deletions cmd/buckyd/compress_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,72 @@
package main

import (
"math"
"path/filepath"
"testing"
"time"

whisper "github.com/go-graphite/go-whisper"
)

// go-carbon can create a metric between healMetric's existence check and the
// compress, and CompressTo refuses to overwrite it. The copy must then land as
// a backfill instead of failing the request with "file already exists".
func TestCompressMetricFillsFileThatAppeared(t *testing.T) {
dir := t.TempDir()
srcPath := filepath.Join(dir, "src.wsp")
dstPath := filepath.Join(dir, "dst.wsp")
rets := whisper.MustParseRetentionDefs("1s:2h")
base := int(time.Now().Unix()) - 3600

srcw, err := whisper.Create(srcPath, rets, whisper.Sum, 0)
if err != nil {
t.Fatalf("create source: %s", err)
}
if err := srcw.UpdateMany([]*whisper.TimeSeriesPoint{
{Time: base, Value: 1},
{Time: base + 1, Value: 2},
}); err != nil {
t.Fatalf("update source: %s", err)
}
if err := srcw.Close(); err != nil {
t.Fatalf("close source: %s", err)
}

// go-carbon got there first.
dstw, err := whisper.CreateWithOptions(dstPath, rets, whisper.Sum, 0,
&whisper.Options{Compressed: true, PointsPerBlock: 1200})
if err != nil {
t.Fatalf("create destination: %s", err)
}
if err := dstw.Close(); err != nil {
t.Fatalf("close destination: %s", err)
}

compressTook, fillTook, err := compressMetric(srcPath, dstPath)
if err != nil {
t.Fatalf("compressMetric: %s", err)
}
if compressTook != 0 || fillTook == 0 {
t.Errorf("expected the backfill fallback to run, got compress=%s fill=%s", compressTook, fillTook)
}

dst, err := whisper.Open(dstPath)
if err != nil {
t.Fatalf("open destination: %s", err)
}
defer dst.Close()
ts, err := dst.Fetch(base-1, base+2)
if err != nil {
t.Fatalf("fetch: %s", err)
}
var got []float64
for _, v := range ts.Values() {
if !math.IsNaN(v) {
got = append(got, v)
}
}
if len(got) != 2 || got[0] != 1 || got[1] != 2 {
t.Errorf("source points did not reach the destination: %v", got)
}
}
43 changes: 32 additions & 11 deletions cmd/buckyd/metrics.go
Original file line number Diff line number Diff line change
Expand Up @@ -425,21 +425,12 @@ func healMetric(w http.ResponseWriter, r *http.Request, path string) {
return
}

compressStart := time.Now()
srcw, err := whisper.Open(srcName)
stats.Compress, stats.Fill, err = compressMetric(srcName, path)
if err != nil {
log.Printf("Failed to open source whisper file: %s", err)
log.Printf("Failed to write whisper file %s: %s", path, err)
httpError(w, err.Error(), http.StatusInternalServerError)
return
}
defer srcw.Close()

if err := srcw.CompressTo(path); err != nil {
log.Printf("Failed to compress to whisper file %s: %s", path, err)
httpError(w, err.Error(), http.StatusInternalServerError)
return
}
stats.Compress = time.Since(compressStart)

return
}
Expand Down Expand Up @@ -484,6 +475,36 @@ func healMetric(w http.ResponseWriter, r *http.Request, path string) {
}
}

// compressMetric writes the whisper file dumped at srcName to path in the
// compressed format.
//
// healMetric only gets here when path did not exist, but go-carbon can create
// it in the meantime and go-whisper refuses to overwrite an existing file. The
// dump is exactly what the backfill path wants, so fill into the file that
// appeared rather than failing the copy with "file already exists".
func compressMetric(srcName, path string) (compressTook, fillTook time.Duration, err error) {
srcw, err := whisper.Open(srcName)
if err != nil {
return 0, 0, fmt.Errorf("open source whisper file: %w", err)
}
defer srcw.Close()

start := time.Now()
err = srcw.CompressTo(path)
if err == nil {
return time.Since(start), 0, nil
}
if !errors.Is(err, os.ErrExist) {
return 0, 0, fmt.Errorf("compress to %s: %w", path, err)
}
Comment thread
Copilot marked this conversation as resolved.

start = time.Now()
if err := fill.All(srcName, path); err != nil {
return 0, 0, fmt.Errorf("backfill %s => %s: %w", srcName, path, err)
}
return 0, time.Since(start), nil
}

var httpClient = &http.Client{
Timeout: 600 * time.Second,
Transport: &http.Transport{
Expand Down
Loading