mudhorn/prometheus
1
0
Fork 0
mirror of https://github.com/prometheus/prometheus.git synced 2024-12-24 05:04:05 -08:00
Code Issues Actions 8 Packages Projects Releases Wiki Activity

remove more format types we probably won't use

Browse source 
Signed-off-by: Callum Styan <callumstyan@gmail.com>
Signed-off-by: Nicolás Pazos <npazosmendez@gmail.com>
This commit is contained in:
Callum Styan 2023-11-20 16:19:42 -08:00 committed by Nicolás Pazos
parent 81c0d78ad4
commit 4bdb73780c
8 changed files with 133 additions and 1793 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

583
prompb/remote.pb.go
View file

@ -60,7 +60,7 @@ func (x ReadRequest_ResponseType) String() string {
} }
func (ReadRequest_ResponseType) EnumDescriptor() ([]byte, []int) { func (ReadRequest_ResponseType) EnumDescriptor() ([]byte, []int) {
return fileDescriptor_eefc82927d57d89b, []int{5, 0} return fileDescriptor_eefc82927d57d89b, []int{3, 0}
} }
type WriteRequest struct { type WriteRequest struct {
@ -175,63 +175,6 @@ func (m *MinimizedWriteRequest) GetSymbols() string {
return "" return ""
} }
type MinimizedWriteRequestBytes struct {
Timeseries []MinimizedTimeSeriesBytes `protobuf:"bytes,1,rep,name=timeseries,proto3" json:"timeseries"`
// The symbols table. All symbols are concatenated strings. To read the symbols table, it's required
// to know the offset:length range of the actual symbol to read from this string.
Symbols string `protobuf:"bytes,4,opt,name=symbols,proto3" json:"symbols,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
}
func (m *MinimizedWriteRequestBytes) Reset() { *m = MinimizedWriteRequestBytes{} }
func (m *MinimizedWriteRequestBytes) String() string { return proto.CompactTextString(m) }
func (*MinimizedWriteRequestBytes) ProtoMessage() {}
func (*MinimizedWriteRequestBytes) Descriptor() ([]byte, []int) {
return fileDescriptor_eefc82927d57d89b, []int{2}
}
func (m *MinimizedWriteRequestBytes) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
}
func (m *MinimizedWriteRequestBytes) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
if deterministic {
return xxx_messageInfo_MinimizedWriteRequestBytes.Marshal(b, m, deterministic)
} else {
b = b[:cap(b)]
n, err := m.MarshalToSizedBuffer(b)
if err != nil {
return nil, err
}
return b[:n], nil
}
}
func (m *MinimizedWriteRequestBytes) XXX_Merge(src proto.Message) {
xxx_messageInfo_MinimizedWriteRequestBytes.Merge(m, src)
}
func (m *MinimizedWriteRequestBytes) XXX_Size() int {
return m.Size()
}
func (m *MinimizedWriteRequestBytes) XXX_DiscardUnknown() {
xxx_messageInfo_MinimizedWriteRequestBytes.DiscardUnknown(m)
}
var xxx_messageInfo_MinimizedWriteRequestBytes proto.InternalMessageInfo
func (m *MinimizedWriteRequestBytes) GetTimeseries() []MinimizedTimeSeriesBytes {
if m != nil {
return m.Timeseries
}
return nil
}
func (m *MinimizedWriteRequestBytes) GetSymbols() string {
if m != nil {
return m.Symbols
}
return ""
}
type MinimizedWriteRequestLen struct { type MinimizedWriteRequestLen struct {
Timeseries []MinimizedTimeSeriesLen `protobuf:"bytes,1,rep,name=timeseries,proto3" json:"timeseries"` Timeseries []MinimizedTimeSeriesLen `protobuf:"bytes,1,rep,name=timeseries,proto3" json:"timeseries"`
// The symbols table. All symbols are concatenated strings prepended with a varint of their length. // The symbols table. All symbols are concatenated strings prepended with a varint of their length.
@ -246,7 +189,7 @@ func (m *MinimizedWriteRequestLen) Reset() { *m = MinimizedWriteRequestL
func (m *MinimizedWriteRequestLen) String() string { return proto.CompactTextString(m) } func (m *MinimizedWriteRequestLen) String() string { return proto.CompactTextString(m) }
func (*MinimizedWriteRequestLen) ProtoMessage() {} func (*MinimizedWriteRequestLen) ProtoMessage() {}
func (*MinimizedWriteRequestLen) Descriptor() ([]byte, []int) { func (*MinimizedWriteRequestLen) Descriptor() ([]byte, []int) {
return fileDescriptor_eefc82927d57d89b, []int{3} return fileDescriptor_eefc82927d57d89b, []int{2}
} }
func (m *MinimizedWriteRequestLen) XXX_Unmarshal(b []byte) error { func (m *MinimizedWriteRequestLen) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b) return m.Unmarshal(b)
@ -289,63 +232,6 @@ func (m *MinimizedWriteRequestLen) GetSymbols() []byte {
return nil return nil
} }
type MinimizedWriteRequestLenBytes struct {
Timeseries []MinimizedTimeSeriesLenBytes `protobuf:"bytes,1,rep,name=timeseries,proto3" json:"timeseries"`
// The symbols table. All symbols are concatenated strings prepended with a varint of their length.
// To read the symbols table, it's required to know the offset of the actual symbol to read from this string.
Symbols []byte `protobuf:"bytes,4,opt,name=symbols,proto3" json:"symbols,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
}
func (m *MinimizedWriteRequestLenBytes) Reset() { *m = MinimizedWriteRequestLenBytes{} }
func (m *MinimizedWriteRequestLenBytes) String() string { return proto.CompactTextString(m) }
func (*MinimizedWriteRequestLenBytes) ProtoMessage() {}
func (*MinimizedWriteRequestLenBytes) Descriptor() ([]byte, []int) {
return fileDescriptor_eefc82927d57d89b, []int{4}
}
func (m *MinimizedWriteRequestLenBytes) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
}
func (m *MinimizedWriteRequestLenBytes) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
if deterministic {
return xxx_messageInfo_MinimizedWriteRequestLenBytes.Marshal(b, m, deterministic)
} else {
b = b[:cap(b)]
n, err := m.MarshalToSizedBuffer(b)
if err != nil {
return nil, err
}
return b[:n], nil
}
}
func (m *MinimizedWriteRequestLenBytes) XXX_Merge(src proto.Message) {
xxx_messageInfo_MinimizedWriteRequestLenBytes.Merge(m, src)
}
func (m *MinimizedWriteRequestLenBytes) XXX_Size() int {
return m.Size()
}
func (m *MinimizedWriteRequestLenBytes) XXX_DiscardUnknown() {
xxx_messageInfo_MinimizedWriteRequestLenBytes.DiscardUnknown(m)
}
var xxx_messageInfo_MinimizedWriteRequestLenBytes proto.InternalMessageInfo
func (m *MinimizedWriteRequestLenBytes) GetTimeseries() []MinimizedTimeSeriesLenBytes {
if m != nil {
return m.Timeseries
}
return nil
}
func (m *MinimizedWriteRequestLenBytes) GetSymbols() []byte {
if m != nil {
return m.Symbols
}
return nil
}
// ReadRequest represents a remote read request. // ReadRequest represents a remote read request.
type ReadRequest struct { type ReadRequest struct {
Queries []*Query `protobuf:"bytes,1,rep,name=queries,proto3" json:"queries,omitempty"` Queries []*Query `protobuf:"bytes,1,rep,name=queries,proto3" json:"queries,omitempty"`
@ -364,7 +250,7 @@ func (m *ReadRequest) Reset() { *m = ReadRequest{} }
func (m *ReadRequest) String() string { return proto.CompactTextString(m) } func (m *ReadRequest) String() string { return proto.CompactTextString(m) }
func (*ReadRequest) ProtoMessage() {} func (*ReadRequest) ProtoMessage() {}
func (*ReadRequest) Descriptor() ([]byte, []int) { func (*ReadRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_eefc82927d57d89b, []int{5} return fileDescriptor_eefc82927d57d89b, []int{3}
} }
func (m *ReadRequest) XXX_Unmarshal(b []byte) error { func (m *ReadRequest) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b) return m.Unmarshal(b)
@ -420,7 +306,7 @@ func (m *ReadResponse) Reset() { *m = ReadResponse{} }
func (m *ReadResponse) String() string { return proto.CompactTextString(m) } func (m *ReadResponse) String() string { return proto.CompactTextString(m) }
func (*ReadResponse) ProtoMessage() {} func (*ReadResponse) ProtoMessage() {}
func (*ReadResponse) Descriptor() ([]byte, []int) { func (*ReadResponse) Descriptor() ([]byte, []int) {
return fileDescriptor_eefc82927d57d89b, []int{6} return fileDescriptor_eefc82927d57d89b, []int{4}
} }
func (m *ReadResponse) XXX_Unmarshal(b []byte) error { func (m *ReadResponse) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b) return m.Unmarshal(b)
@ -470,7 +356,7 @@ func (m *Query) Reset() { *m = Query{} }
func (m *Query) String() string { return proto.CompactTextString(m) } func (m *Query) String() string { return proto.CompactTextString(m) }
func (*Query) ProtoMessage() {} func (*Query) ProtoMessage() {}
func (*Query) Descriptor() ([]byte, []int) { func (*Query) Descriptor() ([]byte, []int) {
return fileDescriptor_eefc82927d57d89b, []int{7} return fileDescriptor_eefc82927d57d89b, []int{5}
} }
func (m *Query) XXX_Unmarshal(b []byte) error { func (m *Query) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b) return m.Unmarshal(b)
@ -539,7 +425,7 @@ func (m *QueryResult) Reset() { *m = QueryResult{} }
func (m *QueryResult) String() string { return proto.CompactTextString(m) } func (m *QueryResult) String() string { return proto.CompactTextString(m) }
func (*QueryResult) ProtoMessage() {} func (*QueryResult) ProtoMessage() {}
func (*QueryResult) Descriptor() ([]byte, []int) { func (*QueryResult) Descriptor() ([]byte, []int) {
return fileDescriptor_eefc82927d57d89b, []int{8} return fileDescriptor_eefc82927d57d89b, []int{6}
} }
func (m *QueryResult) XXX_Unmarshal(b []byte) error { func (m *QueryResult) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b) return m.Unmarshal(b)
@ -592,7 +478,7 @@ func (m *ChunkedReadResponse) Reset() { *m = ChunkedReadResponse{} }
func (m *ChunkedReadResponse) String() string { return proto.CompactTextString(m) } func (m *ChunkedReadResponse) String() string { return proto.CompactTextString(m) }
func (*ChunkedReadResponse) ProtoMessage() {} func (*ChunkedReadResponse) ProtoMessage() {}
func (*ChunkedReadResponse) Descriptor() ([]byte, []int) { func (*ChunkedReadResponse) Descriptor() ([]byte, []int) {
return fileDescriptor_eefc82927d57d89b, []int{9} return fileDescriptor_eefc82927d57d89b, []int{7}
} }
func (m *ChunkedReadResponse) XXX_Unmarshal(b []byte) error { func (m *ChunkedReadResponse) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b) return m.Unmarshal(b)
@ -639,9 +525,7 @@ func init() {
proto.RegisterEnum("prometheus.ReadRequest_ResponseType", ReadRequest_ResponseType_name, ReadRequest_ResponseType_value) proto.RegisterEnum("prometheus.ReadRequest_ResponseType", ReadRequest_ResponseType_name, ReadRequest_ResponseType_value)
proto.RegisterType((*WriteRequest)(nil), "prometheus.WriteRequest") proto.RegisterType((*WriteRequest)(nil), "prometheus.WriteRequest")
proto.RegisterType((*MinimizedWriteRequest)(nil), "prometheus.MinimizedWriteRequest") proto.RegisterType((*MinimizedWriteRequest)(nil), "prometheus.MinimizedWriteRequest")
proto.RegisterType((*MinimizedWriteRequestBytes)(nil), "prometheus.MinimizedWriteRequestBytes")
proto.RegisterType((*MinimizedWriteRequestLen)(nil), "prometheus.MinimizedWriteRequestLen") proto.RegisterType((*MinimizedWriteRequestLen)(nil), "prometheus.MinimizedWriteRequestLen")
proto.RegisterType((*MinimizedWriteRequestLenBytes)(nil), "prometheus.MinimizedWriteRequestLenBytes")
proto.RegisterType((*ReadRequest)(nil), "prometheus.ReadRequest") proto.RegisterType((*ReadRequest)(nil), "prometheus.ReadRequest")
proto.RegisterType((*ReadResponse)(nil), "prometheus.ReadResponse") proto.RegisterType((*ReadResponse)(nil), "prometheus.ReadResponse")
proto.RegisterType((*Query)(nil), "prometheus.Query") proto.RegisterType((*Query)(nil), "prometheus.Query")
@ -652,46 +536,43 @@ func init() {
func init() { proto.RegisterFile("remote.proto", fileDescriptor_eefc82927d57d89b) } func init() { proto.RegisterFile("remote.proto", fileDescriptor_eefc82927d57d89b) }
var fileDescriptor_eefc82927d57d89b = []byte{ var fileDescriptor_eefc82927d57d89b = []byte{
// 612 bytes of a gzipped FileDescriptorProto // 568 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x9c, 0x94, 0xcb, 0x6e, 0xd3, 0x40, 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x8c, 0x54, 0x4b, 0x6f, 0xd3, 0x40,
0x14, 0x86, 0x3b, 0x4d, 0xda, 0x84, 0xe3, 0x50, 0x99, 0x69, 0x4b, 0x4d, 0x24, 0xda, 0xca, 0x42, 0x10, 0xee, 0xd6, 0x69, 0x13, 0xc6, 0xa1, 0x32, 0xdb, 0x96, 0x9a, 0x1e, 0x9a, 0xc8, 0xe2, 0x10,
0x22, 0x52, 0x51, 0x10, 0xa5, 0x62, 0xd5, 0x05, 0x6d, 0x89, 0x54, 0x4a, 0xcc, 0x65, 0x12, 0x04, 0xa9, 0x28, 0x88, 0x50, 0x71, 0xea, 0x81, 0xb4, 0x44, 0x0a, 0x25, 0xe6, 0xb1, 0x09, 0x02, 0x21,
0x42, 0x48, 0x96, 0x63, 0x1f, 0x35, 0x16, 0xf5, 0xa5, 0x9e, 0xb1, 0x54, 0xb3, 0x66, 0xc5, 0x02, 0x24, 0xcb, 0xb1, 0x47, 0x8d, 0x45, 0xfc, 0xa8, 0x77, 0x2d, 0x35, 0x9c, 0x39, 0x71, 0xe2, 0x37,
0xb1, 0xe0, 0x89, 0x58, 0x75, 0x85, 0x78, 0x02, 0x84, 0xfa, 0x24, 0xc8, 0xb7, 0x32, 0x01, 0x97, 0x71, 0xea, 0x09, 0xf1, 0x0b, 0x10, 0xca, 0x2f, 0x41, 0x7e, 0x85, 0x0d, 0x44, 0x94, 0xdb, 0xee,
0x46, 0xec, 0xec, 0x73, 0xfe, 0xff, 0xf7, 0x37, 0x67, 0xc6, 0x03, 0xad, 0x08, 0xbd, 0x40, 0x60, 0x7c, 0x8f, 0xfd, 0x76, 0x76, 0x6c, 0xa8, 0xc7, 0xe8, 0x87, 0x02, 0xdb, 0x51, 0x1c, 0x8a, 0x90,
0x37, 0x8c, 0x02, 0x11, 0x50, 0x08, 0xa3, 0xc0, 0x43, 0x31, 0xc6, 0x98, 0xb7, 0x15, 0x91, 0x84, 0x42, 0x14, 0x87, 0x3e, 0x8a, 0x09, 0x26, 0x7c, 0x5f, 0x15, 0xb3, 0x08, 0x79, 0x0e, 0xec, 0xef,
0xc8, 0xf3, 0x46, 0x7b, 0xe9, 0x30, 0x38, 0x0c, 0xb2, 0xc7, 0xbb, 0xe9, 0x53, 0x5e, 0xd5, 0x3f, 0x9c, 0x87, 0xe7, 0x61, 0xb6, 0xbc, 0x9f, 0xae, 0xf2, 0xaa, 0xf1, 0x85, 0x40, 0xfd, 0x4d, 0xec,
0x13, 0x68, 0xbd, 0x8a, 0x5c, 0x81, 0x0c, 0x8f, 0x63, 0xe4, 0x82, 0x6e, 0x03, 0x08, 0xd7, 0x43, 0x09, 0x64, 0x78, 0x91, 0x20, 0x17, 0xf4, 0x18, 0x40, 0x78, 0x3e, 0x72, 0x8c, 0x3d, 0xe4, 0x3a,
0x8e, 0x91, 0x8b, 0x5c, 0x23, 0xeb, 0xb5, 0x8e, 0xb2, 0x79, 0xbd, 0xfb, 0x3b, 0xb4, 0x3b, 0x74, 0x69, 0x2a, 0x2d, 0xb5, 0x73, 0xbb, 0xfd, 0xdb, 0xb4, 0x3d, 0xf2, 0x7c, 0x1c, 0x66, 0xe8, 0x49,
0x3d, 0x1c, 0x64, 0xdd, 0xdd, 0xfa, 0xe9, 0x8f, 0xb5, 0x19, 0x26, 0xe9, 0xe9, 0x36, 0x34, 0x3d, 0xe5, 0xea, 0x47, 0x63, 0x8d, 0x49, 0x7c, 0x7a, 0x0c, 0x35, 0x1f, 0x85, 0xed, 0xda, 0xc2, 0xd6,
0x14, 0x96, 0x63, 0x09, 0x4b, 0xab, 0x65, 0xde, 0xb6, 0xec, 0x35, 0x50, 0x44, 0xae, 0x6d, 0x14, 0x95, 0x4c, 0xbb, 0x2f, 0x6b, 0x4d, 0x14, 0xb1, 0xe7, 0x98, 0x05, 0xa3, 0xd0, 0x2f, 0x14, 0x67,
0x8a, 0xc2, 0x7f, 0xee, 0x38, 0xa8, 0x37, 0x67, 0xd5, 0x9a, 0xfe, 0x81, 0xc0, 0xb2, 0xe1, 0xfa, 0x95, 0xda, 0xba, 0xa6, 0x18, 0x9f, 0x08, 0xec, 0x9a, 0x5e, 0xe0, 0xf9, 0xde, 0x47, 0x74, 0x97,
0xae, 0xe7, 0xbe, 0x47, 0x67, 0x82, 0xad, 0x57, 0xc1, 0xb6, 0x36, 0x91, 0x5f, 0xda, 0xfe, 0x09, 0xb2, 0xf5, 0x56, 0x64, 0x6b, 0x2c, 0xf9, 0x97, 0xb2, 0x7f, 0x86, 0xd4, 0xa1, 0xca, 0x67, 0xfe,
0xa9, 0x41, 0x83, 0x27, 0xde, 0x28, 0x38, 0xe2, 0x5a, 0x7d, 0x9d, 0x74, 0xae, 0xb0, 0xf2, 0x35, 0x38, 0x9c, 0x72, 0xbd, 0xd2, 0x24, 0xad, 0x1b, 0xac, 0xdc, 0xe6, 0x01, 0xce, 0x2a, 0x35, 0x45,
0x07, 0x38, 0xa8, 0x37, 0x6b, 0x6a, 0x5d, 0xff, 0x44, 0xa0, 0x5d, 0x89, 0xb1, 0x9b, 0x08, 0xe4, 0xab, 0x18, 0x9f, 0x09, 0xe8, 0x2b, 0x63, 0x0c, 0x30, 0xa0, 0xfd, 0x15, 0x49, 0x8c, 0x6b, 0x92,
0xf4, 0xa0, 0x82, 0xe5, 0xd6, 0x65, 0x2c, 0xa9, 0xf3, 0xbf, 0x81, 0x3e, 0x12, 0xd0, 0x2a, 0x81, 0x0c, 0x30, 0xb8, 0x3e, 0x4c, 0x7d, 0x75, 0x98, 0x6f, 0x04, 0x54, 0x86, 0xb6, 0x5b, 0x76, 0xe2,
0xfa, 0xe8, 0xd3, 0xfd, 0x0a, 0x1c, 0xfd, 0x12, 0x9c, 0x3e, 0xfa, 0x97, 0xc3, 0xb4, 0xaa, 0x61, 0x10, 0xaa, 0x17, 0x89, 0x7c, 0xf8, 0x2d, 0xf9, 0xf0, 0x57, 0x09, 0xc6, 0x33, 0x56, 0x32, 0xe8,
0xbe, 0x10, 0xb8, 0x79, 0x11, 0x4c, 0x3e, 0x20, 0xa3, 0x82, 0xe8, 0xf6, 0x14, 0x44, 0xd3, 0xcd, 0x7b, 0xd8, 0xb3, 0x1d, 0x07, 0x23, 0x81, 0xae, 0x15, 0x23, 0x8f, 0xc2, 0x80, 0xa3, 0x95, 0x8d,
0xe8, 0x02, 0xac, 0x6f, 0x04, 0x14, 0x86, 0x96, 0x53, 0x9e, 0x98, 0x0d, 0x68, 0x1c, 0xc7, 0x32, 0x86, 0xbe, 0xde, 0x54, 0x5a, 0x5b, 0x9d, 0xbb, 0xb2, 0x58, 0x3a, 0xa6, 0xcd, 0x0a, 0xf6, 0x68,
0xc1, 0x35, 0x99, 0xe0, 0x45, 0x8c, 0x51, 0xc2, 0x4a, 0x05, 0x7d, 0x0b, 0x2b, 0x96, 0x6d, 0x63, 0x16, 0x21, 0xdb, 0x2d, 0x4d, 0xe4, 0x2a, 0x37, 0x8e, 0xa0, 0x2e, 0x17, 0xa8, 0x0a, 0xd5, 0x61,
0x28, 0xd0, 0x31, 0x23, 0xe4, 0x61, 0xe0, 0x73, 0x34, 0xb3, 0x5f, 0x48, 0x9b, 0x5d, 0xaf, 0x75, 0xd7, 0x7c, 0x39, 0xe8, 0x0d, 0xb5, 0x35, 0xba, 0x07, 0xdb, 0xc3, 0x11, 0xeb, 0x75, 0xcd, 0xde,
0x16, 0x26, 0xf7, 0x57, 0xfa, 0x4c, 0x97, 0x15, 0xea, 0x61, 0x12, 0x22, 0x5b, 0x2e, 0x43, 0xe4, 0x13, 0xeb, 0xed, 0x0b, 0x66, 0x9d, 0xf6, 0x5f, 0x3f, 0x7f, 0x36, 0xd4, 0x88, 0xd1, 0x4d, 0x55,
0x2a, 0xd7, 0xb7, 0xa0, 0x25, 0x17, 0xa8, 0x02, 0x8d, 0xc1, 0x8e, 0xf1, 0xbc, 0xdf, 0x1b, 0xa8, 0xf6, 0xc2, 0x8a, 0x3e, 0x80, 0x6a, 0x8c, 0x3c, 0x99, 0x8a, 0xf2, 0x42, 0x7b, 0x7f, 0x5f, 0x28,
0x33, 0x74, 0x05, 0x16, 0x07, 0x43, 0xd6, 0xdb, 0x31, 0x7a, 0x8f, 0xcc, 0xd7, 0xcf, 0x98, 0xb9, 0xc3, 0x59, 0xc9, 0x33, 0xbe, 0x12, 0xd8, 0xc8, 0x00, 0x7a, 0x0f, 0x28, 0x17, 0x76, 0x2c, 0xac,
0xb7, 0xff, 0xf2, 0xe9, 0x93, 0x81, 0x4a, 0xf4, 0x9d, 0xd4, 0x65, 0x9d, 0x47, 0xd1, 0x7b, 0xd0, 0xac, 0xaf, 0xc2, 0xf6, 0x23, 0xcb, 0x4f, 0x7d, 0x48, 0x4b, 0x61, 0x5a, 0x86, 0x8c, 0x4a, 0xc0,
0x88, 0x90, 0xc7, 0x47, 0xa2, 0x5c, 0xd0, 0xca, 0xdf, 0x0b, 0xca, 0xfa, 0xac, 0xd4, 0xe9, 0x5f, 0xe4, 0xb4, 0x05, 0x1a, 0x06, 0xee, 0x32, 0x77, 0x3d, 0xe3, 0x6e, 0x61, 0xe0, 0xca, 0xcc, 0x23,
0x09, 0xcc, 0x65, 0x0d, 0x7a, 0x07, 0x28, 0x17, 0x56, 0x24, 0xcc, 0x6c, 0xae, 0xc2, 0xf2, 0x42, 0xa8, 0xf9, 0xb6, 0x70, 0x26, 0x18, 0xf3, 0x62, 0x9a, 0x75, 0x39, 0xd5, 0xc0, 0x1e, 0xe3, 0xd4,
0xd3, 0x4b, 0x73, 0x48, 0xa7, 0xc6, 0xd4, 0xac, 0x33, 0x2c, 0x1b, 0x06, 0xa7, 0x1d, 0x50, 0xd1, 0xcc, 0x09, 0x6c, 0xc1, 0xa4, 0x87, 0xb0, 0x31, 0xf1, 0x02, 0x91, 0xbf, 0xa7, 0xda, 0xd9, 0xfd,
0x77, 0x26, 0xb5, 0xb3, 0x99, 0x76, 0x01, 0x7d, 0x47, 0x56, 0x6e, 0x41, 0xd3, 0xb3, 0x84, 0x3d, 0xb3, 0xb9, 0xfd, 0x14, 0x64, 0x39, 0xc7, 0xe8, 0x81, 0x2a, 0x5d, 0x8e, 0x3e, 0xfa, 0xff, 0xaf,
0xc6, 0x88, 0x17, 0x7f, 0xbd, 0x26, 0x53, 0xf5, 0xad, 0x11, 0x1e, 0x19, 0xb9, 0x80, 0x9d, 0x2b, 0x4f, 0x9e, 0x22, 0xe3, 0x12, 0xb6, 0x4f, 0x27, 0x49, 0xf0, 0x21, 0x7d, 0x1c, 0xa9, 0xab, 0x8f,
0xe9, 0x06, 0xcc, 0x8d, 0x5d, 0x5f, 0xe4, 0xfb, 0xa9, 0x6c, 0x2e, 0xff, 0x39, 0xdc, 0xfd, 0xb4, 0x61, 0xcb, 0xc9, 0xcb, 0xd6, 0x92, 0xe5, 0x1d, 0xd9, 0xb2, 0x10, 0x16, 0xae, 0x37, 0x1d, 0x79,
0xc9, 0x72, 0x8d, 0xde, 0x03, 0x45, 0x5a, 0x1c, 0x7d, 0x30, 0xfd, 0x2d, 0x25, 0x9f, 0x22, 0xfd, 0x4b, 0x1b, 0xa0, 0xa6, 0x63, 0x34, 0xb3, 0xbc, 0xc0, 0xc5, 0xcb, 0xa2, 0x4f, 0x90, 0x95, 0x9e,
0x04, 0x16, 0xf7, 0xc6, 0xb1, 0xff, 0x2e, 0xdd, 0x1c, 0x69, 0xaa, 0x0f, 0x61, 0xc1, 0xce, 0xcb, 0xa6, 0x95, 0x93, 0x9d, 0xab, 0xf9, 0x01, 0xf9, 0x3e, 0x3f, 0x20, 0x3f, 0xe7, 0x07, 0xe4, 0xdd,
0xe6, 0x44, 0xe4, 0x0d, 0x39, 0xb2, 0x30, 0x16, 0xa9, 0x57, 0x6d, 0xf9, 0x95, 0xae, 0x81, 0x92, 0x66, 0xea, 0x1b, 0x8d, 0xc7, 0x9b, 0xd9, 0xdf, 0xe5, 0xe1, 0xaf, 0x00, 0x00, 0x00, 0xff, 0xff,
0x1e, 0xa3, 0xc4, 0x74, 0x7d, 0x07, 0x4f, 0x8a, 0x39, 0x41, 0x56, 0x7a, 0x9c, 0x56, 0x76, 0x97, 0xf1, 0x65, 0x72, 0x0c, 0x9c, 0x04, 0x00, 0x00,
0x4e, 0xcf, 0x56, 0xc9, 0xf7, 0xb3, 0x55, 0xf2, 0xf3, 0x6c, 0x95, 0xbc, 0x99, 0x4f, 0x73, 0xc3,
0xd1, 0x68, 0x3e, 0xbb, 0x85, 0xef, 0xff, 0x0a, 0x00, 0x00, 0xff, 0xff, 0xc9, 0xc8, 0x2e, 0x1e,
0xc4, 0x05, 0x00, 0x00,
} }
func (m *WriteRequest) Marshal() (dAtA []byte, err error) { func (m *WriteRequest) Marshal() (dAtA []byte, err error) {
@ -797,54 +678,6 @@ func (m *MinimizedWriteRequest) MarshalToSizedBuffer(dAtA []byte) (int, error) {
return len(dAtA) - i, nil return len(dAtA) - i, nil
} }
func (m *MinimizedWriteRequestBytes) Marshal() (dAtA []byte, err error) {
size := m.Size()
dAtA = make([]byte, size)
n, err := m.MarshalToSizedBuffer(dAtA[:size])
if err != nil {
return nil, err
}
return dAtA[:n], nil
}
func (m *MinimizedWriteRequestBytes) MarshalTo(dAtA []byte) (int, error) {
size := m.Size()
return m.MarshalToSizedBuffer(dAtA[:size])
}
func (m *MinimizedWriteRequestBytes) MarshalToSizedBuffer(dAtA []byte) (int, error) {
i := len(dAtA)
_ = i
var l int
_ = l
if m.XXX_unrecognized != nil {
i -= len(m.XXX_unrecognized)
copy(dAtA[i:], m.XXX_unrecognized)
}
if len(m.Symbols) > 0 {
i -= len(m.Symbols)
copy(dAtA[i:], m.Symbols)
i = encodeVarintRemote(dAtA, i, uint64(len(m.Symbols)))
i--
dAtA[i] = 0x22
}
if len(m.Timeseries) > 0 {
for iNdEx := len(m.Timeseries) - 1; iNdEx >= 0; iNdEx-- {
{
size, err := m.Timeseries[iNdEx].MarshalToSizedBuffer(dAtA[:i])
if err != nil {
return 0, err
}
i -= size
i = encodeVarintRemote(dAtA, i, uint64(size))
}
i--
dAtA[i] = 0xa
}
}
return len(dAtA) - i, nil
}
func (m *MinimizedWriteRequestLen) Marshal() (dAtA []byte, err error) { func (m *MinimizedWriteRequestLen) Marshal() (dAtA []byte, err error) {
size := m.Size() size := m.Size()
dAtA = make([]byte, size) dAtA = make([]byte, size)
@ -893,54 +726,6 @@ func (m *MinimizedWriteRequestLen) MarshalToSizedBuffer(dAtA []byte) (int, error
return len(dAtA) - i, nil return len(dAtA) - i, nil
} }
func (m *MinimizedWriteRequestLenBytes) Marshal() (dAtA []byte, err error) {
size := m.Size()
dAtA = make([]byte, size)
n, err := m.MarshalToSizedBuffer(dAtA[:size])
if err != nil {
return nil, err
}
return dAtA[:n], nil
}
func (m *MinimizedWriteRequestLenBytes) MarshalTo(dAtA []byte) (int, error) {
size := m.Size()
return m.MarshalToSizedBuffer(dAtA[:size])
}
func (m *MinimizedWriteRequestLenBytes) MarshalToSizedBuffer(dAtA []byte) (int, error) {
i := len(dAtA)
_ = i
var l int
_ = l
if m.XXX_unrecognized != nil {
i -= len(m.XXX_unrecognized)
copy(dAtA[i:], m.XXX_unrecognized)
}
if len(m.Symbols) > 0 {
i -= len(m.Symbols)
copy(dAtA[i:], m.Symbols)
i = encodeVarintRemote(dAtA, i, uint64(len(m.Symbols)))
i--
dAtA[i] = 0x22
}
if len(m.Timeseries) > 0 {
for iNdEx := len(m.Timeseries) - 1; iNdEx >= 0; iNdEx-- {
{
size, err := m.Timeseries[iNdEx].MarshalToSizedBuffer(dAtA[:i])
if err != nil {
return 0, err
}
i -= size
i = encodeVarintRemote(dAtA, i, uint64(size))
}
i--
dAtA[i] = 0xa
}
}
return len(dAtA) - i, nil
}
func (m *ReadRequest) Marshal() (dAtA []byte, err error) { func (m *ReadRequest) Marshal() (dAtA []byte, err error) {
size := m.Size() size := m.Size()
dAtA = make([]byte, size) dAtA = make([]byte, size)
@ -1248,28 +1033,6 @@ func (m *MinimizedWriteRequest) Size() (n int) {
return n return n
} }
func (m *MinimizedWriteRequestBytes) Size() (n int) {
if m == nil {
return 0
}
var l int
_ = l
if len(m.Timeseries) > 0 {
for _, e := range m.Timeseries {
l = e.Size()
n += 1 + l + sovRemote(uint64(l))
}
}
l = len(m.Symbols)
if l > 0 {
n += 1 + l + sovRemote(uint64(l))
}
if m.XXX_unrecognized != nil {
n += len(m.XXX_unrecognized)
}
return n
}
func (m *MinimizedWriteRequestLen) Size() (n int) { func (m *MinimizedWriteRequestLen) Size() (n int) {
if m == nil { if m == nil {
return 0 return 0
@ -1292,28 +1055,6 @@ func (m *MinimizedWriteRequestLen) Size() (n int) {
return n return n
} }
func (m *MinimizedWriteRequestLenBytes) Size() (n int) {
if m == nil {
return 0
}
var l int
_ = l
if len(m.Timeseries) > 0 {
for _, e := range m.Timeseries {
l = e.Size()
n += 1 + l + sovRemote(uint64(l))
}
}
l = len(m.Symbols)
if l > 0 {
n += 1 + l + sovRemote(uint64(l))
}
if m.XXX_unrecognized != nil {
n += len(m.XXX_unrecognized)
}
return n
}
func (m *ReadRequest) Size() (n int) { func (m *ReadRequest) Size() (n int) {
if m == nil { if m == nil {
return 0 return 0
@ -1666,123 +1407,6 @@ func (m *MinimizedWriteRequest) Unmarshal(dAtA []byte) error {
} }
return nil return nil
} }
func (m *MinimizedWriteRequestBytes) Unmarshal(dAtA []byte) error {
l := len(dAtA)
iNdEx := 0
for iNdEx < l {
preIndex := iNdEx
var wire uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowRemote
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
wire |= uint64(b&0x7F) << shift
if b < 0x80 {
break
}
}
fieldNum := int32(wire >> 3)
wireType := int(wire & 0x7)
if wireType == 4 {
return fmt.Errorf("proto: MinimizedWriteRequestBytes: wiretype end group for non-group")
}
if fieldNum <= 0 {
return fmt.Errorf("proto: MinimizedWriteRequestBytes: illegal tag %d (wire type %d)", fieldNum, wire)
}
switch fieldNum {
case 1:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field Timeseries", wireType)
}
var msglen int
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowRemote
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
msglen |= int(b&0x7F) << shift
if b < 0x80 {
break
}
}
if msglen < 0 {
return ErrInvalidLengthRemote
}
postIndex := iNdEx + msglen
if postIndex < 0 {
return ErrInvalidLengthRemote
}
if postIndex > l {
return io.ErrUnexpectedEOF
}
m.Timeseries = append(m.Timeseries, MinimizedTimeSeriesBytes{})
if err := m.Timeseries[len(m.Timeseries)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
return err
}
iNdEx = postIndex
case 4:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field Symbols", wireType)
}
var stringLen uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowRemote
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
stringLen |= uint64(b&0x7F) << shift
if b < 0x80 {
break
}
}
intStringLen := int(stringLen)
if intStringLen < 0 {
return ErrInvalidLengthRemote
}
postIndex := iNdEx + intStringLen
if postIndex < 0 {
return ErrInvalidLengthRemote
}
if postIndex > l {
return io.ErrUnexpectedEOF
}
m.Symbols = string(dAtA[iNdEx:postIndex])
iNdEx = postIndex
default:
iNdEx = preIndex
skippy, err := skipRemote(dAtA[iNdEx:])
if err != nil {
return err
}
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthRemote
}
if (iNdEx + skippy) > l {
return io.ErrUnexpectedEOF
}
m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
iNdEx += skippy
}
}
if iNdEx > l {
return io.ErrUnexpectedEOF
}
return nil
}
func (m *MinimizedWriteRequestLen) Unmarshal(dAtA []byte) error { func (m *MinimizedWriteRequestLen) Unmarshal(dAtA []byte) error {
l := len(dAtA) l := len(dAtA)
iNdEx := 0 iNdEx := 0
@ -1902,125 +1526,6 @@ func (m *MinimizedWriteRequestLen) Unmarshal(dAtA []byte) error {
} }
return nil return nil
} }
func (m *MinimizedWriteRequestLenBytes) Unmarshal(dAtA []byte) error {
l := len(dAtA)
iNdEx := 0
for iNdEx < l {
preIndex := iNdEx
var wire uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowRemote
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
wire |= uint64(b&0x7F) << shift
if b < 0x80 {
break
}
}
fieldNum := int32(wire >> 3)
wireType := int(wire & 0x7)
if wireType == 4 {
return fmt.Errorf("proto: MinimizedWriteRequestLenBytes: wiretype end group for non-group")
}
if fieldNum <= 0 {
return fmt.Errorf("proto: MinimizedWriteRequestLenBytes: illegal tag %d (wire type %d)", fieldNum, wire)
}
switch fieldNum {
case 1:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field Timeseries", wireType)
}
var msglen int
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowRemote
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
msglen |= int(b&0x7F) << shift
if b < 0x80 {
break
}
}
if msglen < 0 {
return ErrInvalidLengthRemote
}
postIndex := iNdEx + msglen
if postIndex < 0 {
return ErrInvalidLengthRemote
}
if postIndex > l {
return io.ErrUnexpectedEOF
}
m.Timeseries = append(m.Timeseries, MinimizedTimeSeriesLenBytes{})
if err := m.Timeseries[len(m.Timeseries)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
return err
}
iNdEx = postIndex
case 4:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field Symbols", wireType)
}
var byteLen int
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowRemote
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
byteLen |= int(b&0x7F) << shift
if b < 0x80 {
break
}
}
if byteLen < 0 {
return ErrInvalidLengthRemote
}
postIndex := iNdEx + byteLen
if postIndex < 0 {
return ErrInvalidLengthRemote
}
if postIndex > l {
return io.ErrUnexpectedEOF
}
m.Symbols = append(m.Symbols[:0], dAtA[iNdEx:postIndex]...)
if m.Symbols == nil {
m.Symbols = []byte{}
}
iNdEx = postIndex
default:
iNdEx = preIndex
skippy, err := skipRemote(dAtA[iNdEx:])
if err != nil {
return err
}
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthRemote
}
if (iNdEx + skippy) > l {
return io.ErrUnexpectedEOF
}
m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
iNdEx += skippy
}
}
if iNdEx > l {
return io.ErrUnexpectedEOF
}
return nil
}
func (m *ReadRequest) Unmarshal(dAtA []byte) error { func (m *ReadRequest) Unmarshal(dAtA []byte) error {
l := len(dAtA) l := len(dAtA)
iNdEx := 0 iNdEx := 0

24
prompb/remote.proto
View file

@ -39,18 +39,6 @@ message MinimizedWriteRequest {
string symbols = 4; string symbols = 4;
} }
message MinimizedWriteRequestBytes {
repeated MinimizedTimeSeriesBytes timeseries = 1 [(gogoproto.nullable) = false];
// Cortex uses this field to determine the source of the write request.
// We reserve it to avoid any compatibility issues.
reserved 2;
// Metadata (3) has moved to be part of the TimeSeries type
reserved 3;
// The symbols table. All symbols are concatenated strings. To read the symbols table, it's required
// to know the offset:length range of the actual symbol to read from this string.
string symbols = 4;
}
message MinimizedWriteRequestLen { message MinimizedWriteRequestLen {
repeated MinimizedTimeSeriesLen timeseries = 1 [(gogoproto.nullable) = false]; repeated MinimizedTimeSeriesLen timeseries = 1 [(gogoproto.nullable) = false];
// Cortex uses this field to determine the source of the write request. // Cortex uses this field to determine the source of the write request.
@ -63,18 +51,6 @@ message MinimizedWriteRequestLen {
bytes symbols = 4; bytes symbols = 4;
} }
message MinimizedWriteRequestLenBytes {
repeated MinimizedTimeSeriesLenBytes timeseries = 1 [(gogoproto.nullable) = false];
// Cortex uses this field to determine the source of the write request.
// We reserve it to avoid any compatibility issues.
reserved 2;
// Metadata (3) has moved to be part of the TimeSeries type
reserved 3;
// The symbols table. All symbols are concatenated strings prepended with a varint of their length.
// To read the symbols table, it's required to know the offset of the actual symbol to read from this string.
bytes symbols = 4;
}
// ReadRequest represents a remote read request. // ReadRequest represents a remote read request.
message ReadRequest { message ReadRequest {
repeated Query queries = 1; repeated Query queries = 1;

917
prompb/types.pb.go
View file

File diff suppressed because it is too large Load diff

31
prompb/types.proto
View file

@ -134,6 +134,7 @@ message TimeSeries {
message MinimizedTimeSeries { message MinimizedTimeSeries {
// Sorted list of label name-value pair references. This list's len is always multiple of 4, // Sorted list of label name-value pair references. This list's len is always multiple of 4,
// packing tuples of (label name offset, label name length, label value offset, label value length). // packing tuples of (label name offset, label name length, label value offset, label value length).
// Offsets point to the symbol table in the higher level MinimizedWriteRequestLen.
repeated uint32 label_symbols = 1 [(gogoproto.nullable) = false]; repeated uint32 label_symbols = 1 [(gogoproto.nullable) = false];
// Sorted by time, oldest sample first. // Sorted by time, oldest sample first.
@ -144,23 +145,10 @@ message MinimizedTimeSeries {
// TODO: add metadata // TODO: add metadata
} }
message MinimizedTimeSeriesBytes {
// Sorted list of label name-value pair references, encoded as varints. This
// list's real len is always multiple of 4, packing tuples of (label name offset,
// label name length, label value offset, label value length).
bytes label_symbols = 1;
// Sorted by time, oldest sample first.
// TODO: support references for other types
repeated Sample samples = 2 [(gogoproto.nullable) = false];
repeated Exemplar exemplars = 3 [(gogoproto.nullable) = false];
repeated Histogram histograms = 4 [(gogoproto.nullable) = false];
// TODO: add metadata
}
message MinimizedTimeSeriesLen { message MinimizedTimeSeriesLen {
// Sorted list of label name-value pair references, encoded as varints. This // Sorted list of label name-value pair references, encoded as 32bit uint. This
// list's real len is always multiple of 2, label name offset/label value offset. // list's real len is always multiple of 2, label name offset/label value offset.
// Offsets point to the symbol table in the higher level MinimizedWriteRequestLen.
repeated fixed32 label_symbols = 1; repeated fixed32 label_symbols = 1;
// Sorted by time, oldest sample first. // Sorted by time, oldest sample first.
@ -171,19 +159,6 @@ message MinimizedTimeSeriesLen {
// TODO: add metadata // TODO: add metadata
} }
message MinimizedTimeSeriesLenBytes{
// Sorted list of label name-value pair references, encoded as varints. This
// list's real len is always multiple of 2, label name offset/label value offset.
bytes label_symbols = 1;
// Sorted by time, oldest sample first.
// TODO: support references for other types
repeated Sample samples = 2 [(gogoproto.nullable) = false];
repeated Exemplar exemplars = 3 [(gogoproto.nullable) = false];
repeated Histogram histograms = 4 [(gogoproto.nullable) = false];
// TODO: add metadata
}
message Label { message Label {
string name = 1; string name = 1;
string value = 2; string value = 2;

6
scripts/remotewrite11-bench/run.sh
View file

@ -8,13 +8,7 @@ declare -a INSTANCES
# (sender,receiver) pairs to run: (sender_name; sender_flags; receiver_name; receiver_flags) # (sender,receiver) pairs to run: (sender_name; sender_flags; receiver_name; receiver_flags)
INSTANCES+=('sender-v1;;receiver-v1;') INSTANCES+=('sender-v1;;receiver-v1;')
INSTANCES+=('sender-v11-min32-optimized-varint;--remote-write-format 1;receiver-v11-min32-optimized-varint;--remote-write-format 1') INSTANCES+=('sender-v11-min32-optimized-varint;--remote-write-format 1;receiver-v11-min32-optimized-varint;--remote-write-format 1')
INSTANCES+=('sender-v11-min64-fixed;--remote-write-format 2;receiver-v11-min64-fixed;--remote-write-format 2')
INSTANCES+=('sender-v11-min32-fixed;--remote-write-format 3;receiver-v11-min32-fixed;--remote-write-format 3')
INSTANCES+=('sender-v11-min-bytes;--remote-write-format 4;receiver-v11-min-bytes;--remote-write-format 4')
INSTANCES+=('sender-v11-min-len;--remote-write-format 5;receiver-v11-min-len;--remote-write-format 5') INSTANCES+=('sender-v11-min-len;--remote-write-format 5;receiver-v11-min-len;--remote-write-format 5')
INSTANCES+=('sender-v11-min-len-bytes;--remote-write-format 6;receiver-v11-min-len-bytes;--remote-write-format 6')
# ~~~~~~~~~~~~~ # ~~~~~~~~~~~~~

38
storage/remote/codec.go
View file

@ -1091,25 +1091,6 @@ func DecodeMinimizedWriteRequest(r io.Reader) (*prompb.MinimizedWriteRequest, er
return &req, nil return &req, nil
} }
func DecodeMinimizedWriteRequestBytes(r io.Reader) (*prompb.MinimizedWriteRequestBytes, error) {
compressed, err := io.ReadAll(r)
if err != nil {
return nil, err
}
reqBuf, err := snappy.Decode(nil, compressed)
if err != nil {
return nil, err
}
var req prompb.MinimizedWriteRequestBytes
if err := proto.Unmarshal(reqBuf, &req); err != nil {
return nil, err
}
return &req, nil
}
func DecodeMinimizedWriteRequestLen(r io.Reader) (*prompb.MinimizedWriteRequestLen, error) { func DecodeMinimizedWriteRequestLen(r io.Reader) (*prompb.MinimizedWriteRequestLen, error) {
compressed, err := io.ReadAll(r) compressed, err := io.ReadAll(r)
if err != nil { if err != nil {
@ -1129,25 +1110,6 @@ func DecodeMinimizedWriteRequestLen(r io.Reader) (*prompb.MinimizedWriteRequestL
return &req, nil return &req, nil
} }
func DecodeMinimizedWriteRequestLenBytes(r io.Reader) (*prompb.MinimizedWriteRequestLenBytes, error) {
compressed, err := io.ReadAll(r)
if err != nil {
return nil, err
}
reqBuf, err := snappy.Decode(nil, compressed)
if err != nil {
return nil, err
}
var req prompb.MinimizedWriteRequestLenBytes
if err := proto.Unmarshal(reqBuf, &req); err != nil {
return nil, err
}
return &req, nil
}
func MinimizedWriteRequestToWriteRequest(redReq *prompb.MinimizedWriteRequest) (*prompb.WriteRequest, error) { func MinimizedWriteRequestToWriteRequest(redReq *prompb.MinimizedWriteRequest) (*prompb.WriteRequest, error) {
req := &prompb.WriteRequest{ req := &prompb.WriteRequest{
Timeseries: make([]prompb.TimeSeries, len(redReq.Timeseries)), Timeseries: make([]prompb.TimeSeries, len(redReq.Timeseries)),

240
storage/remote/queue_manager.go
View file

@ -395,11 +395,7 @@ type RemoteWriteFormat int64
const ( const (
Base1 RemoteWriteFormat = iota // original map based format Base1 RemoteWriteFormat = iota // original map based format
Min32Optimized // two 32bit varint plus marshalling optimization Min32Optimized // two 32bit varint plus marshalling optimization
Min64Fixed // a single fixed64 bit value, first 32 are offset and 2nd 32 are MinLen // symbols are now just offsets, and we encode lengths as varints in the large symbols string (which is also now a byte slice)
Min32Fixed
MinBytes // two 32bit fixed, similar to optimized but not varints + no manual marshalling optimization
MinLen // symbols are now just offsets, and we encode lengths as varints in the large symbols string (which is also now a byte slice)
MinLenBytes // the previous two combined
) )
// QueueManager manages a queue of samples to be sent to the Storage // QueueManager manages a queue of samples to be sent to the Storage
@ -1390,21 +1386,11 @@ func (s *shards) runShard(ctx context.Context, shardID int, queue *queue) {
pendingMinimizedData[i].Samples = []prompb.Sample{{}} pendingMinimizedData[i].Samples = []prompb.Sample{{}}
} }
pendingMinBytesData := make([]prompb.MinimizedTimeSeriesBytes, max)
for i := range pendingMinBytesData {
pendingMinBytesData[i].Samples = []prompb.Sample{{}}
}
pendingMinLenData := make([]prompb.MinimizedTimeSeriesLen, max) pendingMinLenData := make([]prompb.MinimizedTimeSeriesLen, max)
for i := range pendingMinLenData { for i := range pendingMinLenData {
pendingMinLenData[i].Samples = []prompb.Sample{{}} pendingMinLenData[i].Samples = []prompb.Sample{{}}
} }
pendingMinLenBytesData := make([]prompb.MinimizedTimeSeriesLenBytes, max)
for i := range pendingMinLenData {
pendingMinLenBytesData[i].Samples = []prompb.Sample{{}}
}
timer := time.NewTimer(time.Duration(s.qm.cfg.BatchSendDeadline)) timer := time.NewTimer(time.Duration(s.qm.cfg.BatchSendDeadline))
stop := func() { stop := func() {
if !timer.Stop() { if !timer.Stop() {
@ -1449,21 +1435,11 @@ func (s *shards) runShard(ctx context.Context, shardID int, queue *queue) {
n := nPendingSamples + nPendingExemplars + nPendingHistograms n := nPendingSamples + nPendingExemplars + nPendingHistograms
s.sendMinSamples(ctx, pendingMinimizedData[:n], symbolTable.LabelsString(), nPendingSamples, nPendingExemplars, nPendingHistograms, &pBufRaw, &buf) s.sendMinSamples(ctx, pendingMinimizedData[:n], symbolTable.LabelsString(), nPendingSamples, nPendingExemplars, nPendingHistograms, &pBufRaw, &buf)
symbolTable.clear() symbolTable.clear()
case MinBytes:
nPendingSamples, nPendingExemplars, nPendingHistograms := populateMinimizedTimeSeriesBytes(&symbolTable, batch, pendingMinBytesData, s.qm.sendExemplars, s.qm.sendNativeHistograms)
n := nPendingSamples + nPendingExemplars + nPendingHistograms
s.sendMinBytes(ctx, pendingMinBytesData[:n], symbolTable.LabelsString(), nPendingSamples, nPendingExemplars, nPendingHistograms, pBuf, &buf)
symbolTable.clear()
case MinLen: case MinLen:
nPendingSamples, nPendingExemplars, nPendingHistograms := populateMinimizedTimeSeriesLen(&symbolTable, batch, pendingMinLenData, s.qm.sendExemplars, s.qm.sendNativeHistograms) nPendingSamples, nPendingExemplars, nPendingHistograms := populateMinimizedTimeSeriesLen(&symbolTable, batch, pendingMinLenData, s.qm.sendExemplars, s.qm.sendNativeHistograms)
n := nPendingSamples + nPendingExemplars + nPendingHistograms n := nPendingSamples + nPendingExemplars + nPendingHistograms
s.sendMinLenSamples(ctx, pendingMinLenData[:n], symbolTable.LabelsData(), nPendingSamples, nPendingExemplars, nPendingHistograms, pBuf, &buf) s.sendMinLenSamples(ctx, pendingMinLenData[:n], symbolTable.LabelsData(), nPendingSamples, nPendingExemplars, nPendingHistograms, pBuf, &buf)
symbolTable.clear() symbolTable.clear()
case MinLenBytes:
nPendingSamples, nPendingExemplars, nPendingHistograms := populateMinimizedTimeSeriesLenBytes(&symbolTable, batch, pendingMinLenBytesData, s.qm.sendExemplars, s.qm.sendNativeHistograms)
n := nPendingSamples + nPendingExemplars + nPendingHistograms
s.sendMinLenBytesSamples(ctx, pendingMinLenBytesData[:n], symbolTable.LabelsData(), nPendingSamples, nPendingExemplars, nPendingHistograms, pBuf, &buf)
symbolTable.clear()
} }
queue.ReturnForReuse(batch) queue.ReturnForReuse(batch)
@ -1572,30 +1548,6 @@ func (s *shards) sendMinLenSamples(ctx context.Context, samples []prompb.Minimiz
s.updateMetrics(ctx, err, sampleCount, exemplarCount, histogramCount, time.Since(begin)) s.updateMetrics(ctx, err, sampleCount, exemplarCount, histogramCount, time.Since(begin))
} }
func (s *shards) sendMinLenBytesSamples(ctx context.Context, samples []prompb.MinimizedTimeSeriesLenBytes, labels []byte, sampleCount, exemplarCount, histogramCount int, pBuf *proto.Buffer, buf *[]byte) {
begin := time.Now()
// Build the ReducedWriteRequest with no metadata.
// Failing to build the write request is non-recoverable, since it will
// only error if marshaling the proto to bytes fails.
req, highest, err := buildMinimizedWriteRequestLenBytes(samples, labels, pBuf, buf)
if err == nil {
err = s.sendSamplesWithBackoff(ctx, req, sampleCount, exemplarCount, histogramCount, highest)
}
s.updateMetrics(ctx, err, sampleCount, exemplarCount, histogramCount, time.Since(begin))
}
func (s *shards) sendMinBytes(ctx context.Context, samples []prompb.MinimizedTimeSeriesBytes, labels string, sampleCount, exemplarCount, histogramCount int, pBuf *proto.Buffer, buf *[]byte) {
begin := time.Now()
// Build the ReducedWriteRequest with no metadata.
// Failing to build the write request is non-recoverable, since it will
// only error if marshaling the proto to bytes fails.
req, highest, err := buildMinimizedWriteRequestBytes(samples, labels, pBuf, buf)
if err == nil {
err = s.sendSamplesWithBackoff(ctx, req, sampleCount, exemplarCount, histogramCount, highest)
}
s.updateMetrics(ctx, err, sampleCount, exemplarCount, histogramCount, time.Since(begin))
}
func (s *shards) updateMetrics(ctx context.Context, err error, sampleCount, exemplarCount, histogramCount int, duration time.Duration) { func (s *shards) updateMetrics(ctx context.Context, err error, sampleCount, exemplarCount, histogramCount int, duration time.Duration) {
if err != nil { if err != nil {
level.Error(s.qm.logger).Log("msg", "non-recoverable error", "count", sampleCount, "exemplarCount", exemplarCount, "err", err) level.Error(s.qm.logger).Log("msg", "non-recoverable error", "count", sampleCount, "exemplarCount", exemplarCount, "err", err)
@ -1720,55 +1672,6 @@ func populateMinimizedTimeSeries(symbolTable *rwSymbolTable, batch []timeSeries,
return nPendingSamples, nPendingExemplars, nPendingHistograms return nPendingSamples, nPendingExemplars, nPendingHistograms
} }
func populateMinimizedTimeSeriesBytes(symbolTable *rwSymbolTable, batch []timeSeries, pendingData []prompb.MinimizedTimeSeriesBytes, sendExemplars, sendNativeHistograms bool) (int, int, int) {
var nPendingSamples, nPendingExemplars, nPendingHistograms int
for nPending, d := range batch {
pendingData[nPending].Samples = pendingData[nPending].Samples[:0]
if sendExemplars {
pendingData[nPending].Exemplars = pendingData[nPending].Exemplars[:0]
}
if sendNativeHistograms {
pendingData[nPending].Histograms = pendingData[nPending].Histograms[:0]
}
// Number of pending samples is limited by the fact that sendSamples (via sendSamplesWithBackoff)
// retries endlessly, so once we reach max samples, if we can never send to the endpoint we'll
// stop reading from the queue. This makes it safe to reference pendingSamples by index.
// pendingData[nPending].Labels = labelsToLabelsProto(d.seriesLabels, pendingData[nPending].Labels)
pendingData[nPending].LabelSymbols = labelsToByteSlice(d.seriesLabels, symbolTable, pendingData[nPending].LabelSymbols)
switch d.sType {
case tSample:
pendingData[nPending].Samples = append(pendingData[nPending].Samples, prompb.Sample{
Value: d.value,
Timestamp: d.timestamp,
})
nPendingSamples++
// TODO: handle all types
//case tExemplar:
// l := make([]prompb.LabelRef, 0, d.exemplarLabels.Len())
// d.exemplarLabels.Range(func(el labels.Label) {
// nRef := pool.intern(el.Name)
// vRef := pool.intern(el.Value)
// l = append(l, prompb.LabelRef{NameRef: nRef, ValueRef: vRef})
// })
// pendingData[nPending].Exemplars = append(pendingData[nPending].Exemplars, prompb.ExemplarRef{
// Labels: l,
// Value: d.value,
// Timestamp: d.timestamp,
// })
// nPendingExemplars++
case tHistogram:
pendingData[nPending].Histograms = append(pendingData[nPending].Histograms, HistogramToHistogramProto(d.timestamp, d.histogram))
nPendingHistograms++
case tFloatHistogram:
pendingData[nPending].Histograms = append(pendingData[nPending].Histograms, FloatHistogramToHistogramProto(d.timestamp, d.floatHistogram))
nPendingHistograms++
}
}
return nPendingSamples, nPendingExemplars, nPendingHistograms
}
func populateMinimizedTimeSeriesLen(symbolTable *rwSymbolTable, batch []timeSeries, pendingData []prompb.MinimizedTimeSeriesLen, sendExemplars, sendNativeHistograms bool) (int, int, int) { func populateMinimizedTimeSeriesLen(symbolTable *rwSymbolTable, batch []timeSeries, pendingData []prompb.MinimizedTimeSeriesLen, sendExemplars, sendNativeHistograms bool) (int, int, int) {
var nPendingSamples, nPendingExemplars, nPendingHistograms int var nPendingSamples, nPendingExemplars, nPendingHistograms int
for nPending, d := range batch { for nPending, d := range batch {
@ -1818,55 +1721,6 @@ func populateMinimizedTimeSeriesLen(symbolTable *rwSymbolTable, batch []timeSeri
return nPendingSamples, nPendingExemplars, nPendingHistograms return nPendingSamples, nPendingExemplars, nPendingHistograms
} }
func populateMinimizedTimeSeriesLenBytes(symbolTable *rwSymbolTable, batch []timeSeries, pendingData []prompb.MinimizedTimeSeriesLenBytes, sendExemplars, sendNativeHistograms bool) (int, int, int) {
var nPendingSamples, nPendingExemplars, nPendingHistograms int
for nPending, d := range batch {
pendingData[nPending].Samples = pendingData[nPending].Samples[:0]
if sendExemplars {
pendingData[nPending].Exemplars = pendingData[nPending].Exemplars[:0]
}
if sendNativeHistograms {
pendingData[nPending].Histograms = pendingData[nPending].Histograms[:0]
}
// Number of pending samples is limited by the fact that sendSamples (via sendSamplesWithBackoff)
// retries endlessly, so once we reach max samples, if we can never send to the endpoint we'll
// stop reading from the queue. This makes it safe to reference pendingSamples by index.
// pendingData[nPending].Labels = labelsToLabelsProto(d.seriesLabels, pendingData[nPending].Labels)
pendingData[nPending].LabelSymbols = labelsToByteSlice(d.seriesLabels, symbolTable, pendingData[nPending].LabelSymbols)
switch d.sType {
case tSample:
pendingData[nPending].Samples = append(pendingData[nPending].Samples, prompb.Sample{
Value: d.value,
Timestamp: d.timestamp,
})
nPendingSamples++
// TODO: handle all types
//case tExemplar:
// l := make([]prompb.LabelRef, 0, d.exemplarLabels.Len())
// d.exemplarLabels.Range(func(el labels.Label) {
// nRef := pool.intern(el.Name)
// vRef := pool.intern(el.Value)
// l = append(l, prompb.LabelRef{NameRef: nRef, ValueRef: vRef})
// })
// pendingData[nPending].Exemplars = append(pendingData[nPending].Exemplars, prompb.ExemplarRef{
// Labels: l,
// Value: d.value,
// Timestamp: d.timestamp,
// })
// nPendingExemplars++
case tHistogram:
pendingData[nPending].Histograms = append(pendingData[nPending].Histograms, HistogramToHistogramProto(d.timestamp, d.histogram))
nPendingHistograms++
case tFloatHistogram:
pendingData[nPending].Histograms = append(pendingData[nPending].Histograms, FloatHistogramToHistogramProto(d.timestamp, d.floatHistogram))
nPendingHistograms++
}
}
return nPendingSamples, nPendingExemplars, nPendingHistograms
}
func sendWriteRequestWithBackoff(ctx context.Context, cfg config.QueueConfig, l log.Logger, attempt func(int) error, onRetry func()) error { func sendWriteRequestWithBackoff(ctx context.Context, cfg config.QueueConfig, l log.Logger, attempt func(int) error, onRetry func()) error {
backoff := cfg.MinBackoff backoff := cfg.MinBackoff
sleepDuration := model.Duration(0) sleepDuration := model.Duration(0)
@ -2152,95 +2006,3 @@ func buildMinimizedWriteRequestLen(samples []prompb.MinimizedTimeSeriesLen, labe
} }
return compressed, highest, nil return compressed, highest, nil
} }
func buildMinimizedWriteRequestLenBytes(samples []prompb.MinimizedTimeSeriesLenBytes, labels []byte, pBuf *proto.Buffer, buf *[]byte) ([]byte, int64, error) {
var highest int64
for _, ts := range samples {
// At the moment we only ever append a TimeSeries with a single sample or exemplar in it.
if len(ts.Samples) > 0 && ts.Samples[0].Timestamp > highest {
highest = ts.Samples[0].Timestamp
}
if len(ts.Exemplars) > 0 && ts.Exemplars[0].Timestamp > highest {
highest = ts.Exemplars[0].Timestamp
}
if len(ts.Histograms) > 0 && ts.Histograms[0].Timestamp > highest {
highest = ts.Histograms[0].Timestamp
}
}
req := &prompb.MinimizedWriteRequestLenBytes{
Symbols: labels,
Timeseries: samples,
}
if pBuf == nil {
pBuf = proto.NewBuffer(nil) // For convenience in tests. Not efficient.
} else {
pBuf.Reset()
}
err := pBuf.Marshal(req)
if err != nil {
return nil, 0, err
}
// snappy uses len() to see if it needs to allocate a new slice. Make the
// buffer as long as possible.
if buf != nil {
*buf = (*buf)[0:cap(*buf)]
} else {
buf = &[]byte{}
}
compressed := snappy.Encode(*buf, pBuf.Bytes())
if n := snappy.MaxEncodedLen(len(pBuf.Bytes())); buf != nil && n > len(*buf) {
// grow the buffer for the next time
*buf = make([]byte, n)
}
return compressed, highest, nil
}
func buildMinimizedWriteRequestBytes(samples []prompb.MinimizedTimeSeriesBytes, labels string, pBuf *proto.Buffer, buf *[]byte) ([]byte, int64, error) {
var highest int64
for _, ts := range samples {
// At the moment we only ever append a TimeSeries with a single sample or exemplar in it.
if len(ts.Samples) > 0 && ts.Samples[0].Timestamp > highest {
highest = ts.Samples[0].Timestamp
}
if len(ts.Exemplars) > 0 && ts.Exemplars[0].Timestamp > highest {
highest = ts.Exemplars[0].Timestamp
}
if len(ts.Histograms) > 0 && ts.Histograms[0].Timestamp > highest {
highest = ts.Histograms[0].Timestamp
}
}
req := &prompb.MinimizedWriteRequestBytes{
Symbols: labels,
Timeseries: samples,
}
if pBuf == nil {
pBuf = proto.NewBuffer(nil) // For convenience in tests. Not efficient.
} else {
pBuf.Reset()
}
err := pBuf.Marshal(req)
if err != nil {
return nil, 0, err
}
// snappy uses len() to see if it needs to allocate a new slice. Make the
// buffer as long as possible.
if buf != nil {
*buf = (*buf)[0:cap(*buf)]
} else {
buf = &[]byte{}
}
compressed := snappy.Encode(*buf, pBuf.Bytes())
if n := snappy.MaxEncodedLen(len(pBuf.Bytes())); buf != nil && n > len(*buf) {
// grow the buffer for the next time
*buf = make([]byte, n)
}
return compressed, highest, nil
}

87
storage/remote/write_handler.go
View file

@ -74,9 +74,7 @@ func (h *writeHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
var err error var err error
var req *prompb.WriteRequest var req *prompb.WriteRequest
var reqMin *prompb.MinimizedWriteRequest var reqMin *prompb.MinimizedWriteRequest
var reqMinBytes *prompb.MinimizedWriteRequestBytes
var reqMinLen *prompb.MinimizedWriteRequestLen var reqMinLen *prompb.MinimizedWriteRequestLen
var reqMinLenBytes *prompb.MinimizedWriteRequestLenBytes
// TODO: this should eventually be done via content negotiation/looking at the header // TODO: this should eventually be done via content negotiation/looking at the header
switch h.rwFormat { switch h.rwFormat {
@ -84,12 +82,8 @@ func (h *writeHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
req, err = DecodeWriteRequest(r.Body) req, err = DecodeWriteRequest(r.Body)
case Min32Optimized: case Min32Optimized:
reqMin, err = DecodeMinimizedWriteRequest(r.Body) reqMin, err = DecodeMinimizedWriteRequest(r.Body)
case MinBytes:
reqMinBytes, err = DecodeMinimizedWriteRequestBytes(r.Body)
case MinLen: case MinLen:
reqMinLen, err = DecodeMinimizedWriteRequestLen(r.Body) reqMinLen, err = DecodeMinimizedWriteRequestLen(r.Body)
case MinLenBytes:
reqMinLenBytes, err = DecodeMinimizedWriteRequestLenBytes(r.Body)
} }
if err != nil { if err != nil {
@ -104,12 +98,8 @@ func (h *writeHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
err = h.write(r.Context(), req) err = h.write(r.Context(), req)
case Min32Optimized: case Min32Optimized:
err = h.writeMin(r.Context(), reqMin) err = h.writeMin(r.Context(), reqMin)
case MinBytes:
err = h.writeMinBytes(r.Context(), reqMinBytes)
case MinLen: case MinLen:
err = h.writeMinLen(r.Context(), reqMinLen) err = h.writeMinLen(r.Context(), reqMinLen)
case MinLenBytes:
err = h.writeMinLenBytes(r.Context(), reqMinLenBytes)
} }
switch { switch {
@ -343,44 +333,6 @@ func (h *writeHandler) writeMin(ctx context.Context, req *prompb.MinimizedWriteR
return nil return nil
} }
func (h *writeHandler) writeMinBytes(ctx context.Context, req *prompb.MinimizedWriteRequestBytes) (err error) {
outOfOrderExemplarErrs := 0
app := h.appendable.Appender(ctx)
defer func() {
if err != nil {
_ = app.Rollback()
return
}
err = app.Commit()
}()
for _, ts := range req.Timeseries {
ls := ByteSliceToLabels(req.Symbols, ts.LabelSymbols)
err := h.appendSamples(app, ts.Samples, ls)
if err != nil {
return err
}
for _, ep := range ts.Exemplars {
e := exemplarProtoToExemplar(ep)
//e := exemplarRefProtoToExemplar(req.StringSymbolTable, ep)
h.appendExemplar(app, e, ls, &outOfOrderExemplarErrs)
}
err = h.appendHistograms(app, ts.Histograms, ls)
if err != nil {
return err
}
}
if outOfOrderExemplarErrs > 0 {
_ = level.Warn(h.logger).Log("msg", "Error on ingesting out-of-order exemplars", "num_dropped", outOfOrderExemplarErrs)
}
return nil
}
func (h *writeHandler) writeMinLen(ctx context.Context, req *prompb.MinimizedWriteRequestLen) (err error) { func (h *writeHandler) writeMinLen(ctx context.Context, req *prompb.MinimizedWriteRequestLen) (err error) {
outOfOrderExemplarErrs := 0 outOfOrderExemplarErrs := 0
@ -419,42 +371,3 @@ func (h *writeHandler) writeMinLen(ctx context.Context, req *prompb.MinimizedWri
return nil return nil
} }
func (h *writeHandler) writeMinLenBytes(ctx context.Context, req *prompb.MinimizedWriteRequestLenBytes) (err error) {
outOfOrderExemplarErrs := 0
app := h.appendable.Appender(ctx)
defer func() {
if err != nil {
_ = app.Rollback()
return
}
err = app.Commit()
}()
for _, ts := range req.Timeseries {
ls := ByteSliceToLabelsSymbolsByte(req.Symbols, ts.LabelSymbols)
err := h.appendSamples(app, ts.Samples, ls)
if err != nil {
return err
}
for _, ep := range ts.Exemplars {
e := exemplarProtoToExemplar(ep)
//e := exemplarRefProtoToExemplar(req.StringSymbolTable, ep)
h.appendExemplar(app, e, ls, &outOfOrderExemplarErrs)
}
err = h.appendHistograms(app, ts.Histograms, ls)
if err != nil {
return err
}
}
if outOfOrderExemplarErrs > 0 {
_ = level.Warn(h.logger).Log("msg", "Error on ingesting out-of-order exemplars", "num_dropped", outOfOrderExemplarErrs)
}
return nil
}