mirror of https://github.com/grafana/loki
Remove dependency on Bigchunk, which is a leftover of Cortex and is only used in tests (#10354)
**What this PR does / why we need it**: This PR removes the dependency on "Bigchunk", which is a "default" implementation of the `chunk.Data` interface and used only in tests. **Special notes for your reviewer**: The PR replaces the usage of this chunk type with the Loki implementation (`chunkenc.MemChunk` wrapped into `chunkenc.Facade`) where possible. In case where it was not possible (tests in `pkg/storage/chunk` package) due to otherwise circular import, it was replaces by the `chunk.dummyChunk` implementation, which is just a no-op that implements `chunk.Data` interface) The PR also removes the protos for the ingester client (`pkg/ingester/client/ingester.proto`) which were only referenced in the removed `pkg/util/chunkcompat/compat.go` file. Signed-off-by: Christian Haudum <christian.haudum@gmail.com>pull/10394/head
Christian Haudum
2 years ago
committed by
GitHub
parent
92ba055ae5
commit
31981bd4af
File diff suppressed because it is too large
Load Diff
@ -1,167 +0,0 @@ |
||||
syntax = "proto3"; |
||||
|
||||
package ingesterpb; |
||||
|
||||
import "gogoproto/gogo.proto"; |
||||
import "pkg/logproto/logproto.proto"; |
||||
import "pkg/logproto/metrics.proto"; |
||||
|
||||
option go_package = "github.com/grafana/loki/pkg/ingester/client"; |
||||
option (gogoproto.marshaler_all) = true; |
||||
option (gogoproto.unmarshaler_all) = true; |
||||
|
||||
service Ingester { |
||||
rpc Push(logproto.WriteRequest) returns (logproto.WriteResponse) {} |
||||
|
||||
rpc Query(QueryRequest) returns (QueryResponse) {} |
||||
|
||||
rpc QueryStream(QueryRequest) returns (stream QueryStreamResponse) {} |
||||
|
||||
rpc QueryExemplars(ExemplarQueryRequest) returns (ExemplarQueryResponse) {} |
||||
|
||||
rpc LabelValues(LabelValuesRequest) returns (LabelValuesResponse) {} |
||||
|
||||
rpc LabelNames(LabelNamesRequest) returns (LabelNamesResponse) {} |
||||
|
||||
rpc UserStats(UserStatsRequest) returns (UserStatsResponse) {} |
||||
|
||||
rpc AllUserStats(UserStatsRequest) returns (UsersStatsResponse) {} |
||||
|
||||
rpc MetricsForLabelMatchers(MetricsForLabelMatchersRequest) returns (MetricsForLabelMatchersResponse) {} |
||||
|
||||
rpc MetricsMetadata(MetricsMetadataRequest) returns (MetricsMetadataResponse) {} |
||||
|
||||
// TransferChunks allows leaving ingester (client) to stream chunks directly to joining ingesters (server). |
||||
rpc TransferChunks(stream TimeSeriesChunk) returns (TransferChunksResponse) {} |
||||
} |
||||
|
||||
message ReadRequest { |
||||
repeated QueryRequest queries = 1; |
||||
} |
||||
|
||||
message ReadResponse { |
||||
repeated QueryResponse results = 1; |
||||
} |
||||
|
||||
message QueryRequest { |
||||
int64 start_timestamp_ms = 1; |
||||
int64 end_timestamp_ms = 2; |
||||
repeated LabelMatcher matchers = 3; |
||||
} |
||||
|
||||
message ExemplarQueryRequest { |
||||
int64 start_timestamp_ms = 1; |
||||
int64 end_timestamp_ms = 2; |
||||
repeated LabelMatchers matchers = 3; |
||||
} |
||||
|
||||
message QueryResponse { |
||||
repeated logproto.TimeSeries timeseries = 1 [(gogoproto.nullable) = false]; |
||||
} |
||||
|
||||
// QueryStreamResponse contains a batch of timeseries chunks or timeseries. Only one of these series will be populated. |
||||
message QueryStreamResponse { |
||||
repeated TimeSeriesChunk chunkseries = 1 [(gogoproto.nullable) = false]; |
||||
repeated logproto.TimeSeries timeseries = 2 [(gogoproto.nullable) = false]; |
||||
} |
||||
|
||||
message ExemplarQueryResponse { |
||||
repeated logproto.TimeSeries timeseries = 1 [(gogoproto.nullable) = false]; |
||||
} |
||||
|
||||
message LabelValuesRequest { |
||||
string label_name = 1; |
||||
int64 start_timestamp_ms = 2; |
||||
int64 end_timestamp_ms = 3; |
||||
LabelMatchers matchers = 4; |
||||
} |
||||
|
||||
message LabelValuesResponse { |
||||
repeated string label_values = 1; |
||||
} |
||||
|
||||
message LabelNamesRequest { |
||||
int64 start_timestamp_ms = 1; |
||||
int64 end_timestamp_ms = 2; |
||||
} |
||||
|
||||
message LabelNamesResponse { |
||||
repeated string label_names = 1; |
||||
} |
||||
|
||||
message UserStatsRequest {} |
||||
|
||||
message UserStatsResponse { |
||||
double ingestion_rate = 1; |
||||
uint64 num_series = 2; |
||||
double api_ingestion_rate = 3; |
||||
double rule_ingestion_rate = 4; |
||||
} |
||||
|
||||
message UserIDStatsResponse { |
||||
string user_id = 1; |
||||
UserStatsResponse data = 2; |
||||
} |
||||
|
||||
message UsersStatsResponse { |
||||
repeated UserIDStatsResponse stats = 1; |
||||
} |
||||
|
||||
message MetricsForLabelMatchersRequest { |
||||
int64 start_timestamp_ms = 1; |
||||
int64 end_timestamp_ms = 2; |
||||
repeated LabelMatchers matchers_set = 3; |
||||
} |
||||
|
||||
message MetricsForLabelMatchersResponse { |
||||
repeated logproto.Metric metric = 1; |
||||
} |
||||
|
||||
message MetricsMetadataRequest {} |
||||
|
||||
message MetricsMetadataResponse { |
||||
repeated logproto.MetricMetadata metadata = 1; |
||||
} |
||||
|
||||
message TimeSeriesChunk { |
||||
string from_ingester_id = 1; |
||||
string user_id = 2; |
||||
repeated logproto.LegacyLabelPair labels = 3 [ |
||||
(gogoproto.nullable) = false, |
||||
(gogoproto.customtype) = "github.com/grafana/loki/pkg/logproto.LabelAdapter" |
||||
]; |
||||
repeated Chunk chunks = 4 [(gogoproto.nullable) = false]; |
||||
} |
||||
|
||||
message Chunk { |
||||
int64 start_timestamp_ms = 1; |
||||
int64 end_timestamp_ms = 2; |
||||
int32 encoding = 3; |
||||
bytes data = 4; |
||||
} |
||||
|
||||
message TransferChunksResponse {} |
||||
|
||||
message LabelMatchers { |
||||
repeated LabelMatcher matchers = 1; |
||||
} |
||||
|
||||
enum MatchType { |
||||
EQUAL = 0; |
||||
NOT_EQUAL = 1; |
||||
REGEX_MATCH = 2; |
||||
REGEX_NO_MATCH = 3; |
||||
} |
||||
|
||||
message LabelMatcher { |
||||
MatchType type = 1; |
||||
string name = 2; |
||||
string value = 3; |
||||
} |
||||
|
||||
message TimeSeriesFile { |
||||
string from_ingester_id = 1; |
||||
string user_id = 2; |
||||
string filename = 3; |
||||
bytes data = 4; |
||||
} |
||||
@ -1,256 +0,0 @@ |
||||
package chunk |
||||
|
||||
import ( |
||||
"bytes" |
||||
"encoding/binary" |
||||
"errors" |
||||
"io" |
||||
|
||||
"github.com/prometheus/common/model" |
||||
"github.com/prometheus/prometheus/tsdb/chunkenc" |
||||
|
||||
"github.com/grafana/loki/pkg/util/filter" |
||||
) |
||||
|
||||
const samplesPerChunk = 120 |
||||
|
||||
var errOutOfBounds = errors.New("out of bounds") |
||||
|
||||
type smallChunk struct { |
||||
chunkenc.XORChunk |
||||
start int64 |
||||
} |
||||
|
||||
// bigchunk is a set of prometheus/tsdb chunks. It grows over time and has no
|
||||
// upperbound on number of samples it can contain.
|
||||
type bigchunk struct { |
||||
chunks []smallChunk |
||||
|
||||
appender chunkenc.Appender |
||||
remainingSamples int |
||||
} |
||||
|
||||
func newBigchunk() *bigchunk { |
||||
return &bigchunk{} |
||||
} |
||||
|
||||
// TODO(owen-d): remove bigchunk from our code, we don't use it.
|
||||
// Hack an Entries() impl
|
||||
func (b *bigchunk) Entries() int { return 0 } |
||||
|
||||
func (b *bigchunk) Add(sample model.SamplePair) (Data, error) { |
||||
if b.remainingSamples == 0 { |
||||
if bigchunkSizeCapBytes > 0 && b.Size() > bigchunkSizeCapBytes { |
||||
return addToOverflowChunk(sample) |
||||
} |
||||
if err := b.addNextChunk(sample.Timestamp); err != nil { |
||||
return nil, err |
||||
} |
||||
} |
||||
|
||||
b.appender.Append(int64(sample.Timestamp), float64(sample.Value)) |
||||
b.remainingSamples-- |
||||
return nil, nil |
||||
} |
||||
|
||||
// addNextChunk adds a new XOR "subchunk" to the internal list of chunks.
|
||||
func (b *bigchunk) addNextChunk(start model.Time) error { |
||||
// To save memory, we "compact" the previous chunk - the array backing the slice
|
||||
// will be upto 2x too big, and we can save this space.
|
||||
const chunkCapacityExcess = 32 // don't bother copying if it's within this range
|
||||
if l := len(b.chunks); l > 0 { |
||||
oldBuf := b.chunks[l-1].XORChunk.Bytes() |
||||
if cap(oldBuf) > len(oldBuf)+chunkCapacityExcess { |
||||
buf := make([]byte, len(oldBuf)) |
||||
copy(buf, oldBuf) |
||||
compacted, err := chunkenc.FromData(chunkenc.EncXOR, buf) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
b.chunks[l-1].XORChunk = *compacted.(*chunkenc.XORChunk) |
||||
} |
||||
} |
||||
|
||||
// Explicitly reallocate slice to avoid up to 2x overhead if we let append() do it
|
||||
if len(b.chunks)+1 > cap(b.chunks) { |
||||
newChunks := make([]smallChunk, len(b.chunks), len(b.chunks)+1) |
||||
copy(newChunks, b.chunks) |
||||
b.chunks = newChunks |
||||
} |
||||
b.chunks = append(b.chunks, smallChunk{ |
||||
XORChunk: *chunkenc.NewXORChunk(), |
||||
start: int64(start), |
||||
}) |
||||
|
||||
appender, err := b.chunks[len(b.chunks)-1].Appender() |
||||
if err != nil { |
||||
return err |
||||
} |
||||
b.appender = appender |
||||
b.remainingSamples = samplesPerChunk |
||||
return nil |
||||
} |
||||
|
||||
func (b *bigchunk) Rebound(_, _ model.Time, _ filter.Func) (Data, error) { |
||||
return nil, errors.New("not implemented") |
||||
} |
||||
|
||||
func (b *bigchunk) Marshal(wio io.Writer) error { |
||||
w := writer{wio} |
||||
if err := w.WriteVarInt16(uint16(len(b.chunks))); err != nil { |
||||
return err |
||||
} |
||||
for _, chunk := range b.chunks { |
||||
buf := chunk.Bytes() |
||||
if err := w.WriteVarInt16(uint16(len(buf))); err != nil { |
||||
return err |
||||
} |
||||
if _, err := w.Write(buf); err != nil { |
||||
return err |
||||
} |
||||
} |
||||
return nil |
||||
} |
||||
|
||||
func (b *bigchunk) MarshalToBuf(buf []byte) error { |
||||
writer := bytes.NewBuffer(buf) |
||||
return b.Marshal(writer) |
||||
} |
||||
|
||||
func (b *bigchunk) UnmarshalFromBuf(buf []byte) error { |
||||
r := reader{buf: buf} |
||||
numChunks, err := r.ReadUint16() |
||||
if err != nil { |
||||
return err |
||||
} |
||||
|
||||
b.chunks = make([]smallChunk, 0, numChunks+1) // allow one extra space in case we want to add new data
|
||||
var reuseIter chunkenc.Iterator |
||||
for i := uint16(0); i < numChunks; i++ { |
||||
chunkLen, err := r.ReadUint16() |
||||
if err != nil { |
||||
return err |
||||
} |
||||
|
||||
chunkBuf, err := r.ReadBytes(int(chunkLen)) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
|
||||
chunk, err := chunkenc.FromData(chunkenc.EncXOR, chunkBuf) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
|
||||
var start int64 |
||||
start, reuseIter, err = firstTime(chunk, reuseIter) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
|
||||
b.chunks = append(b.chunks, smallChunk{ |
||||
XORChunk: *chunk.(*chunkenc.XORChunk), |
||||
start: start, |
||||
}) |
||||
} |
||||
return nil |
||||
} |
||||
|
||||
func (b *bigchunk) Encoding() Encoding { |
||||
return Bigchunk |
||||
} |
||||
|
||||
func (b *bigchunk) Utilization() float64 { |
||||
return 1.0 |
||||
} |
||||
|
||||
func (b *bigchunk) Len() int { |
||||
sum := 0 |
||||
for _, c := range b.chunks { |
||||
sum += c.NumSamples() |
||||
} |
||||
return sum |
||||
} |
||||
|
||||
// Unused, but for compatibility
|
||||
func (b *bigchunk) UncompressedSize() int { return b.Size() } |
||||
|
||||
func (b *bigchunk) Size() int { |
||||
sum := 2 // For the number of sub chunks.
|
||||
for _, c := range b.chunks { |
||||
sum += 2 // For the length of the sub chunk.
|
||||
sum += len(c.Bytes()) |
||||
} |
||||
return sum |
||||
} |
||||
|
||||
func (b *bigchunk) Slice(start, end model.Time) Data { |
||||
i, j := 0, len(b.chunks) |
||||
for k := 0; k < len(b.chunks); k++ { |
||||
if b.chunks[k].start <= int64(start) { |
||||
i = k |
||||
} |
||||
if b.chunks[k].start > int64(end) { |
||||
j = k |
||||
break |
||||
} |
||||
} |
||||
return &bigchunk{ |
||||
chunks: b.chunks[i:j], |
||||
} |
||||
} |
||||
|
||||
type writer struct { |
||||
io.Writer |
||||
} |
||||
|
||||
func (w writer) WriteVarInt16(i uint16) error { |
||||
var b [2]byte |
||||
binary.LittleEndian.PutUint16(b[:], i) |
||||
_, err := w.Write(b[:]) |
||||
return err |
||||
} |
||||
|
||||
type reader struct { |
||||
i int |
||||
buf []byte |
||||
} |
||||
|
||||
func (r *reader) ReadUint16() (uint16, error) { |
||||
if r.i+2 > len(r.buf) { |
||||
return 0, errOutOfBounds |
||||
} |
||||
result := binary.LittleEndian.Uint16(r.buf[r.i:]) |
||||
r.i += 2 |
||||
return result, nil |
||||
} |
||||
|
||||
func (r *reader) ReadBytes(count int) ([]byte, error) { |
||||
if r.i+count > len(r.buf) { |
||||
return nil, errOutOfBounds |
||||
} |
||||
result := r.buf[r.i : r.i+count] |
||||
r.i += count |
||||
return result, nil |
||||
} |
||||
|
||||
// addToOverflowChunk is a utility function that creates a new chunk as overflow
|
||||
// chunk, adds the provided sample to it, and returns a chunk slice containing
|
||||
// the provided old chunk followed by the new overflow chunk.
|
||||
func addToOverflowChunk(s model.SamplePair) (Data, error) { |
||||
overflowChunk := New() |
||||
_, err := overflowChunk.(*bigchunk).Add(s) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
return overflowChunk, nil |
||||
} |
||||
|
||||
func firstTime(c chunkenc.Chunk, iter chunkenc.Iterator) (int64, chunkenc.Iterator, error) { |
||||
var first int64 |
||||
iter = c.Iterator(iter) |
||||
if iter.Next() != chunkenc.ValNone { |
||||
first, _ = iter.At() |
||||
} |
||||
return first, iter, iter.Err() |
||||
} |
||||
@ -0,0 +1,54 @@ |
||||
package chunk |
||||
|
||||
import ( |
||||
"io" |
||||
|
||||
"github.com/prometheus/common/model" |
||||
|
||||
"github.com/grafana/loki/pkg/util/filter" |
||||
) |
||||
|
||||
func newDummyChunk() *dummyChunk { |
||||
return &dummyChunk{} |
||||
} |
||||
|
||||
// dummyChunk implements chunk.Data
|
||||
// It is a placeholder chunk with Encoding(0)
|
||||
// It can be used in tests where the content of a chunk is irrelevant.
|
||||
type dummyChunk struct{} |
||||
|
||||
func (chk *dummyChunk) Add(sample model.SamplePair) (Data, error) { |
||||
return nil, nil |
||||
} |
||||
|
||||
func (chk *dummyChunk) Marshal(io.Writer) error { |
||||
return nil |
||||
} |
||||
|
||||
func (chk *dummyChunk) UnmarshalFromBuf([]byte) error { |
||||
return nil |
||||
} |
||||
|
||||
func (chk *dummyChunk) Encoding() Encoding { |
||||
return Dummy |
||||
} |
||||
|
||||
func (chk *dummyChunk) Rebound(start, end model.Time, filter filter.Func) (Data, error) { |
||||
return nil, nil |
||||
} |
||||
|
||||
func (chk *dummyChunk) Size() int { |
||||
return 0 |
||||
} |
||||
|
||||
func (chk *dummyChunk) UncompressedSize() int { |
||||
return 0 |
||||
} |
||||
|
||||
func (chk *dummyChunk) Entries() int { |
||||
return 0 |
||||
} |
||||
|
||||
func (chk *dummyChunk) Utilization() float64 { |
||||
return 0 |
||||
} |
||||
@ -1,53 +0,0 @@ |
||||
package chunkcompat |
||||
|
||||
import ( |
||||
"bytes" |
||||
|
||||
"github.com/prometheus/common/model" |
||||
"github.com/prometheus/prometheus/model/labels" |
||||
|
||||
"github.com/grafana/loki/pkg/ingester/client" |
||||
"github.com/grafana/loki/pkg/storage/chunk" |
||||
) |
||||
|
||||
// FromChunks converts []client.Chunk to []chunk.Chunk.
|
||||
func FromChunks(userID string, metric labels.Labels, in []client.Chunk) ([]chunk.Chunk, error) { |
||||
out := make([]chunk.Chunk, 0, len(in)) |
||||
for _, i := range in { |
||||
o, err := chunk.NewForEncoding(chunk.Encoding(byte(i.Encoding))) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
|
||||
if err := o.UnmarshalFromBuf(i.Data); err != nil { |
||||
return nil, err |
||||
} |
||||
|
||||
firstTime, lastTime := model.Time(i.StartTimestampMs), model.Time(i.EndTimestampMs) |
||||
// As the lifetime of this chunk is scopes to this request, we don't need
|
||||
// to supply a fingerprint.
|
||||
out = append(out, chunk.NewChunk(userID, 0, metric, o, firstTime, lastTime)) |
||||
} |
||||
return out, nil |
||||
} |
||||
|
||||
// ToChunks converts []chunk.Chunk to []client.Chunk.
|
||||
func ToChunks(in []chunk.Chunk) ([]client.Chunk, error) { |
||||
out := make([]client.Chunk, 0, len(in)) |
||||
for _, i := range in { |
||||
wireChunk := client.Chunk{ |
||||
StartTimestampMs: int64(i.From), |
||||
EndTimestampMs: int64(i.Through), |
||||
Encoding: int32(i.Data.Encoding()), |
||||
} |
||||
|
||||
buf := bytes.NewBuffer(make([]byte, 0, 1024)) |
||||
if err := i.Data.Marshal(buf); err != nil { |
||||
return nil, err |
||||
} |
||||
|
||||
wireChunk.Data = buf.Bytes() |
||||
out = append(out, wireChunk) |
||||
} |
||||
return out, nil |
||||
} |
||||
Loading…
Reference in new issue