This repository has been archived by the owner on Aug 23, 2023. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 105
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
move conversion and chunk encoding to importer pkg
also moves all the related tests
- Loading branch information
Showing
5 changed files
with
260 additions
and
245 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,49 @@ | ||
| package importer | ||
|
|
||
| import ( | ||
| "fmt" | ||
|
|
||
| "github.com/grafana/metrictank/mdata/chunk" | ||
| "github.com/kisielk/whisper-go/whisper" | ||
| ) | ||
|
|
||
| func EncodeChunksFromPoints(points []whisper.Point, intervalIn, chunkSpan uint32, writeUnfinishedChunks bool) []*chunk.Chunk { | ||
| var point whisper.Point | ||
| var t0, prevT0 uint32 | ||
| var c *chunk.Chunk | ||
| var encodedChunks []*chunk.Chunk | ||
|
|
||
| for _, point = range points { | ||
| // this shouldn't happen, but if it would we better catch it here because Metrictank wouldn't handle it well: | ||
| // https://github.com/grafana/metrictank/blob/f1868cccfb92fc82cd853914af958f6d187c5f74/mdata/aggmetric.go#L378 | ||
| if point.Timestamp == 0 { | ||
| continue | ||
| } | ||
|
|
||
| t0 = point.Timestamp - (point.Timestamp % chunkSpan) | ||
| if prevT0 == 0 { | ||
| c = chunk.New(t0) | ||
| prevT0 = t0 | ||
| } else if prevT0 != t0 { | ||
| c.Finish() | ||
| encodedChunks = append(encodedChunks, c) | ||
|
|
||
| c = chunk.New(t0) | ||
| prevT0 = t0 | ||
| } | ||
|
|
||
| err := c.Push(point.Timestamp, point.Value) | ||
| if err != nil { | ||
| panic(fmt.Sprintf("ERROR: Failed to push value into chunk at t0 %d: %q", t0, err)) | ||
| } | ||
| } | ||
|
|
||
| // if the last written point was also the last one of the current chunk, | ||
| // or if writeUnfinishedChunks is on, we close the chunk and push it | ||
| if point.Timestamp == t0+chunkSpan-intervalIn || writeUnfinishedChunks { | ||
| c.Finish() | ||
| encodedChunks = append(encodedChunks, c) | ||
| } | ||
|
|
||
| return encodedChunks | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,79 @@ | ||
| package importer | ||
|
|
||
| import ( | ||
| "testing" | ||
|
|
||
| "github.com/grafana/metrictank/mdata/chunk" | ||
| ) | ||
|
|
||
| func TestEncodedChunksFromPointsWithUnfinished(t *testing.T) { | ||
| points := generatePoints(25200, 10, 10, 0, 8640, func(i float64) float64 { return i + 1 }) | ||
| expectedCount := 8640 // count including unfinished chunks | ||
|
|
||
| chunks := EncodeChunksFromPoints(points, 10, 21600, true) | ||
|
|
||
| if len(chunks) != 5 { | ||
| t.Fatalf("Expected to get 5 chunks, but got %d", len(chunks)) | ||
| } | ||
|
|
||
| i := 0 | ||
| for _, c := range chunks { | ||
| iterGen, err := chunk.NewIterGen(c.Series.T0, 10, c.Encode(21600)) | ||
| if err != nil { | ||
| t.Fatalf("Error getting iterator: %s", err) | ||
| } | ||
|
|
||
| iter, err := iterGen.Get() | ||
| if err != nil { | ||
| t.Fatalf("Error getting iterator: %s", err) | ||
| } | ||
|
|
||
| for iter.Next() { | ||
| ts, val := iter.Values() | ||
| if points[i].Timestamp != ts || points[i].Value != val { | ||
| t.Fatalf("Unexpected value at index %d:\nExpected: %d:%f\nGot: %d:%f\n", i, ts, val, points[i].Timestamp, points[i].Value) | ||
| } | ||
| i++ | ||
| } | ||
| } | ||
| if i != expectedCount { | ||
| t.Fatalf("Unexpected number of datapoints in chunks:\nExpected: %d\nGot: %d\n", expectedCount, i) | ||
| } | ||
| } | ||
|
|
||
| func TestEncodedChunksFromPointsWithoutUnfinished(t *testing.T) { | ||
| // the actual data in these points doesn't matter, we just want to be sure | ||
| // that the chunks resulting from these points include the same data | ||
| points := generatePoints(25200, 10, 10, 0, 8640, func(i float64) float64 { return i + 1 }) | ||
| expectedCount := 8640 - (2520 % 2160) // count minus what would end up in an unfinished chunk | ||
|
|
||
| chunks := EncodeChunksFromPoints(points, 10, 21600, false) | ||
|
|
||
| if len(chunks) != 4 { | ||
| t.Fatalf("Expected to get 4 chunks, but got %d", len(chunks)) | ||
| } | ||
|
|
||
| i := 0 | ||
| for _, c := range chunks { | ||
| iterGen, err := chunk.NewIterGen(c.Series.T0, 10, c.Encode(21600)) | ||
| if err != nil { | ||
| t.Fatalf("Error getting iterator: %s", err) | ||
| } | ||
|
|
||
| iter, err := iterGen.Get() | ||
| if err != nil { | ||
| t.Fatalf("Error getting iterator: %s", err) | ||
| } | ||
|
|
||
| for iter.Next() { | ||
| ts, val := iter.Values() | ||
| if points[i].Timestamp != ts || points[i].Value != val { | ||
| t.Fatalf("Unexpected value at index %d:\nExpected: %d:%f\nGot: %d:%f\n", i, ts, val, points[i].Timestamp, points[i].Value) | ||
| } | ||
| i++ | ||
| } | ||
| } | ||
| if i != expectedCount { | ||
| t.Fatalf("Unexpected number of datapoints in chunks:\nExpected: %d\nGot: %d\n", expectedCount, i) | ||
| } | ||
| } |
Oops, something went wrong.