chore(ingest-limits): Merge proto stream-metadata size fields (#17606)

pull/17484/head
Periklis Tsirakidis 12 months ago committed by GitHub
parent 6e075af1fd
commit e83dbfd8d3
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
  1. 35
      pkg/distributor/distributor_test.go
  2. 5
      pkg/distributor/ingest_limits.go
  3. 10
      pkg/distributor/ingest_limits_test.go
  4. 30
      pkg/limits/frontend/frontend_test.go
  5. 20
      pkg/limits/frontend/http_test.go
  6. 80
      pkg/limits/frontend/ring_test.go
  7. 10
      pkg/limits/ingest_limits.go
  8. 53
      pkg/limits/ingest_limits_test.go
  9. 131
      pkg/limits/proto/limits.pb.go
  10. 3
      pkg/limits/proto/limits.proto
  11. 48
      pkg/limits/stream_metadata_bench_test.go
  12. 41
      pkg/limits/stream_metadata_test.go
  13. 4
      tools/stream-generator/generator/kafka.go

@ -2428,9 +2428,8 @@ func TestDistributor_PushIngestLimits(t *testing.T) {
expectedLimitsRequest: &limitsproto.ExceedsLimitsRequest{
Tenant: "test",
Streams: []*limitsproto.StreamMetadata{{
StreamHash: 0x90eb45def17f924,
EntriesSize: 0x3,
StructuredMetadataSize: 0x0,
StreamHash: 0x90eb45def17f924,
TotalSize: 0x3,
}},
},
limitsResponse: &limitsproto.ExceedsLimitsResponse{
@ -2453,9 +2452,8 @@ func TestDistributor_PushIngestLimits(t *testing.T) {
expectedLimitsRequest: &limitsproto.ExceedsLimitsRequest{
Tenant: "test",
Streams: []*limitsproto.StreamMetadata{{
StreamHash: 0x90eb45def17f924,
EntriesSize: 0x3,
StructuredMetadataSize: 0x0,
StreamHash: 0x90eb45def17f924,
TotalSize: 0x3,
}},
},
limitsResponse: &limitsproto.ExceedsLimitsResponse{
@ -2482,9 +2480,8 @@ func TestDistributor_PushIngestLimits(t *testing.T) {
expectedLimitsRequest: &limitsproto.ExceedsLimitsRequest{
Tenant: "test",
Streams: []*limitsproto.StreamMetadata{{
StreamHash: 0x90eb45def17f924,
EntriesSize: 0x3,
StructuredMetadataSize: 0x0,
StreamHash: 0x90eb45def17f924,
TotalSize: 0x3,
}},
},
limitsResponse: &limitsproto.ExceedsLimitsResponse{
@ -2517,13 +2514,11 @@ func TestDistributor_PushIngestLimits(t *testing.T) {
expectedLimitsRequest: &limitsproto.ExceedsLimitsRequest{
Tenant: "test",
Streams: []*limitsproto.StreamMetadata{{
StreamHash: 0x90eb45def17f924,
EntriesSize: 0x3,
StructuredMetadataSize: 0x0,
StreamHash: 0x90eb45def17f924,
TotalSize: 0x3,
}, {
StreamHash: 0x11561609feba8cf6,
EntriesSize: 0x3,
StructuredMetadataSize: 0x0,
StreamHash: 0x11561609feba8cf6,
TotalSize: 0x3,
}},
},
limitsResponse: &limitsproto.ExceedsLimitsResponse{
@ -2550,9 +2545,8 @@ func TestDistributor_PushIngestLimits(t *testing.T) {
expectedLimitsRequest: &limitsproto.ExceedsLimitsRequest{
Tenant: "test",
Streams: []*limitsproto.StreamMetadata{{
StreamHash: 0x90eb45def17f924,
EntriesSize: 0x3,
StructuredMetadataSize: 0x0,
StreamHash: 0x90eb45def17f924,
TotalSize: 0x3,
}},
},
limitsResponse: &limitsproto.ExceedsLimitsResponse{
@ -2578,9 +2572,8 @@ func TestDistributor_PushIngestLimits(t *testing.T) {
expectedLimitsRequest: &limitsproto.ExceedsLimitsRequest{
Tenant: "test",
Streams: []*limitsproto.StreamMetadata{{
StreamHash: 0x90eb45def17f924,
EntriesSize: 0x3,
StructuredMetadataSize: 0x0,
StreamHash: 0x90eb45def17f924,
TotalSize: 0x3,
}},
},
limitsResponseErr: errors.New("failed to check limits"),

@ -147,9 +147,8 @@ func newExceedsLimitsRequest(tenant string, streams []KeyedStream) (*proto.Excee
for _, stream := range streams {
entriesSize, structuredMetadataSize := calculateStreamSizes(stream.Stream)
streamMetadata = append(streamMetadata, &proto.StreamMetadata{
StreamHash: stream.HashKeyNoShard,
EntriesSize: entriesSize,
StructuredMetadataSize: structuredMetadataSize,
StreamHash: stream.HashKeyNoShard,
TotalSize: entriesSize + structuredMetadataSize,
})
}
return &proto.ExceedsLimitsRequest{

@ -87,13 +87,11 @@ func TestIngestLimits_EnforceLimits(t *testing.T) {
expectedRequest: &proto.ExceedsLimitsRequest{
Tenant: "test",
Streams: []*proto.StreamMetadata{{
StreamHash: 1,
EntriesSize: 0x3,
StructuredMetadataSize: 0x6,
StreamHash: 1,
TotalSize: 9,
}, {
StreamHash: 2,
EntriesSize: 0x3,
StructuredMetadataSize: 0x8,
StreamHash: 2,
TotalSize: 11,
}},
},
responseErr: errors.New("failed to check limits"),

@ -31,9 +31,8 @@ func TestFrontend_ExceedsLimits(t *testing.T) {
exceedsLimitsRequest: &proto.ExceedsLimitsRequest{
Tenant: "test",
Streams: []*proto.StreamMetadata{{
StreamHash: 0x1,
EntriesSize: 0x2,
StructuredMetadataSize: 0x3,
StreamHash: 0x1,
TotalSize: 0x5,
}},
},
exceedsLimitsResponses: []*proto.ExceedsLimitsResponse{{
@ -53,9 +52,8 @@ func TestFrontend_ExceedsLimits(t *testing.T) {
exceedsLimitsRequest: &proto.ExceedsLimitsRequest{
Tenant: "test",
Streams: []*proto.StreamMetadata{{
StreamHash: 0x1,
EntriesSize: 0x2,
StructuredMetadataSize: 0x3,
StreamHash: 0x1,
TotalSize: 0x5,
}},
},
exceedsLimitsResponses: []*proto.ExceedsLimitsResponse{{
@ -69,13 +67,11 @@ func TestFrontend_ExceedsLimits(t *testing.T) {
exceedsLimitsRequest: &proto.ExceedsLimitsRequest{
Tenant: "test",
Streams: []*proto.StreamMetadata{{
StreamHash: 0x1,
EntriesSize: 0x2,
StructuredMetadataSize: 0x3,
StreamHash: 0x1,
TotalSize: 0x5,
}, {
StreamHash: 0x4,
EntriesSize: 0x5,
StructuredMetadataSize: 0x6,
StreamHash: 0x4,
TotalSize: 0x9,
}},
},
exceedsLimitsResponses: []*proto.ExceedsLimitsResponse{{
@ -101,13 +97,11 @@ func TestFrontend_ExceedsLimits(t *testing.T) {
exceedsLimitsRequest: &proto.ExceedsLimitsRequest{
Tenant: "test",
Streams: []*proto.StreamMetadata{{
StreamHash: 0x1,
EntriesSize: 0x2,
StructuredMetadataSize: 0x3,
StreamHash: 0x1,
TotalSize: 0x5,
}, {
StreamHash: 0x4,
EntriesSize: 0x5,
StructuredMetadataSize: 0x6,
StreamHash: 0x4,
TotalSize: 0x9,
}},
},
exceedsLimitsResponses: []*proto.ExceedsLimitsResponse{{

@ -26,18 +26,16 @@ func TestFrontend_ServeHTTP(t *testing.T) {
expectedExceedsLimitsRequest: &proto.ExceedsLimitsRequest{
Tenant: "test",
Streams: []*proto.StreamMetadata{{
StreamHash: 0x1,
EntriesSize: 0x2,
StructuredMetadataSize: 0x3,
StreamHash: 0x1,
TotalSize: 0x5,
}},
},
exceedsLimitsResponses: []*proto.ExceedsLimitsResponse{{}},
request: httpExceedsLimitsRequest{
Tenant: "test",
Streams: []*proto.StreamMetadata{{
StreamHash: 0x1,
EntriesSize: 0x2,
StructuredMetadataSize: 0x3,
StreamHash: 0x1,
TotalSize: 0x5,
}},
},
// expected should be default value.
@ -46,9 +44,8 @@ func TestFrontend_ServeHTTP(t *testing.T) {
expectedExceedsLimitsRequest: &proto.ExceedsLimitsRequest{
Tenant: "test",
Streams: []*proto.StreamMetadata{{
StreamHash: 0x1,
EntriesSize: 0x2,
StructuredMetadataSize: 0x3,
StreamHash: 0x1,
TotalSize: 0x5,
}},
},
exceedsLimitsResponses: []*proto.ExceedsLimitsResponse{{
@ -60,9 +57,8 @@ func TestFrontend_ServeHTTP(t *testing.T) {
request: httpExceedsLimitsRequest{
Tenant: "test",
Streams: []*proto.StreamMetadata{{
StreamHash: 0x1,
EntriesSize: 0x2,
StructuredMetadataSize: 0x3,
StreamHash: 0x1,
TotalSize: 0x5,
}},
},
expected: httpExceedsLimitsResponse{

@ -55,9 +55,8 @@ func TestRingGatherer_ExceedsLimits(t *testing.T) {
request: &proto.ExceedsLimitsRequest{
Tenant: "test",
Streams: []*proto.StreamMetadata{{
StreamHash: 0x1, // 0x1 is assigned to partition 0.
EntriesSize: 0x2,
StructuredMetadataSize: 0x3,
StreamHash: 0x1, // 0x1 is assigned to partition 0.
TotalSize: 0x5,
}},
},
instances: []ring.InstanceDesc{{
@ -73,9 +72,8 @@ func TestRingGatherer_ExceedsLimits(t *testing.T) {
expectedExceedsLimitsRequests: []*proto.ExceedsLimitsRequest{{
Tenant: "test",
Streams: []*proto.StreamMetadata{{
StreamHash: 0x1,
EntriesSize: 0x2,
StructuredMetadataSize: 0x3,
StreamHash: 0x1,
TotalSize: 0x5,
}},
}},
exceedsLimitsResponses: []*proto.ExceedsLimitsResponse{{
@ -100,9 +98,8 @@ func TestRingGatherer_ExceedsLimits(t *testing.T) {
request: &proto.ExceedsLimitsRequest{
Tenant: "test",
Streams: []*proto.StreamMetadata{{
StreamHash: 0x1, // 0x1 is assigned to partition 1.
EntriesSize: 0x2,
StructuredMetadataSize: 0x3,
StreamHash: 0x1, // 0x1 is assigned to partition 1.
TotalSize: 0x5,
}},
},
instances: []ring.InstanceDesc{{
@ -124,9 +121,8 @@ func TestRingGatherer_ExceedsLimits(t *testing.T) {
expectedExceedsLimitsRequests: []*proto.ExceedsLimitsRequest{nil, {
Tenant: "test",
Streams: []*proto.StreamMetadata{{
StreamHash: 0x1,
EntriesSize: 0x2,
StructuredMetadataSize: 0x3,
StreamHash: 0x1,
TotalSize: 0x5,
}},
}},
exceedsLimitsResponses: []*proto.ExceedsLimitsResponse{nil, {
@ -150,13 +146,11 @@ func TestRingGatherer_ExceedsLimits(t *testing.T) {
request: &proto.ExceedsLimitsRequest{
Tenant: "test",
Streams: []*proto.StreamMetadata{{
StreamHash: 0x1, // 0x1 is assigned to partition 1.
EntriesSize: 0x2,
StructuredMetadataSize: 0x3,
StreamHash: 0x1, // 0x1 is assigned to partition 1.
TotalSize: 0x5,
}, {
StreamHash: 0x3, // 0x3 is also assigned to partition 1.
EntriesSize: 0x4,
StructuredMetadataSize: 0x5,
StreamHash: 0x3, // 0x3 is also assigned to partition 1.
TotalSize: 0x9,
}},
},
instances: []ring.InstanceDesc{{
@ -178,13 +172,11 @@ func TestRingGatherer_ExceedsLimits(t *testing.T) {
expectedExceedsLimitsRequests: []*proto.ExceedsLimitsRequest{nil, {
Tenant: "test",
Streams: []*proto.StreamMetadata{{
StreamHash: 0x1,
EntriesSize: 0x2,
StructuredMetadataSize: 0x3,
StreamHash: 0x1,
TotalSize: 0x5,
}, {
StreamHash: 0x3,
EntriesSize: 0x4,
StructuredMetadataSize: 0x5,
StreamHash: 0x3,
TotalSize: 0x9,
}},
}},
exceedsLimitsResponses: []*proto.ExceedsLimitsResponse{nil, {
@ -208,13 +200,11 @@ func TestRingGatherer_ExceedsLimits(t *testing.T) {
request: &proto.ExceedsLimitsRequest{
Tenant: "test",
Streams: []*proto.StreamMetadata{{
StreamHash: 0x1, // 0x1 is assigned to partition 1.
EntriesSize: 0x2,
StructuredMetadataSize: 0x3,
StreamHash: 0x1, // 0x1 is assigned to partition 1.
TotalSize: 0x5,
}, {
StreamHash: 0x2, // 0x2 is also assigned to partition 0.
EntriesSize: 0x4,
StructuredMetadataSize: 0x5,
StreamHash: 0x2, // 0x2 is also assigned to partition 0.
TotalSize: 0x9,
}},
},
instances: []ring.InstanceDesc{{
@ -236,16 +226,14 @@ func TestRingGatherer_ExceedsLimits(t *testing.T) {
expectedExceedsLimitsRequests: []*proto.ExceedsLimitsRequest{{
Tenant: "test",
Streams: []*proto.StreamMetadata{{
StreamHash: 0x2,
EntriesSize: 0x4,
StructuredMetadataSize: 0x5,
StreamHash: 0x2,
TotalSize: 0x9,
}},
}, {
Tenant: "test",
Streams: []*proto.StreamMetadata{{
StreamHash: 0x1,
EntriesSize: 0x2,
StructuredMetadataSize: 0x3,
StreamHash: 0x1,
TotalSize: 0x5,
}},
}},
exceedsLimitsResponses: []*proto.ExceedsLimitsResponse{{
@ -277,13 +265,11 @@ func TestRingGatherer_ExceedsLimits(t *testing.T) {
request: &proto.ExceedsLimitsRequest{
Tenant: "test",
Streams: []*proto.StreamMetadata{{
StreamHash: 0x1, // 0x1 is assigned to partition 1.
EntriesSize: 0x2,
StructuredMetadataSize: 0x3,
StreamHash: 0x1, // 0x1 is assigned to partition 1.
TotalSize: 0x5,
}, {
StreamHash: 0x2, // 0x2 is also assigned to partition 0.
EntriesSize: 0x4,
StructuredMetadataSize: 0x5,
StreamHash: 0x2, // 0x2 is also assigned to partition 0.
TotalSize: 0x9,
}},
},
instances: []ring.InstanceDesc{{
@ -305,16 +291,14 @@ func TestRingGatherer_ExceedsLimits(t *testing.T) {
expectedExceedsLimitsRequests: []*proto.ExceedsLimitsRequest{{
Tenant: "test",
Streams: []*proto.StreamMetadata{{
StreamHash: 0x2,
EntriesSize: 0x4,
StructuredMetadataSize: 0x5,
StreamHash: 0x2,
TotalSize: 0x9,
}},
}, {
Tenant: "test",
Streams: []*proto.StreamMetadata{{
StreamHash: 0x1,
EntriesSize: 0x2,
StructuredMetadataSize: 0x3,
StreamHash: 0x1,
TotalSize: 0x5,
}},
}},
exceedsLimitsResponses: []*proto.ExceedsLimitsResponse{{

@ -309,17 +309,15 @@ func (s *IngestLimits) updateMetadata(rec *proto.StreamMetadata, tenant string,
bucketStart = lastSeenAt.Truncate(s.cfg.BucketDuration).UnixNano()
// Calculate the rate window cutoff for cleaning up old buckets
rateWindowCutoff = lastSeenAt.Add(-s.cfg.RateWindow).UnixNano()
// Calculate the total size of the stream
totalSize = rec.EntriesSize + rec.StructuredMetadataSize
)
if assigned := s.partitionManager.Has(partition); !assigned {
return
}
s.metadata.Store(tenant, partition, rec.StreamHash, totalSize, recordTime, bucketStart, rateWindowCutoff)
s.metadata.Store(tenant, partition, rec.StreamHash, rec.TotalSize, recordTime, bucketStart, rateWindowCutoff)
s.metrics.tenantIngestedBytesTotal.WithLabelValues(tenant).Add(float64(totalSize))
s.metrics.tenantIngestedBytesTotal.WithLabelValues(tenant).Add(float64(rec.TotalSize))
}
// stopping implements the Service interface's stopping method.
@ -381,7 +379,7 @@ func (s *IngestLimits) ExceedsLimits(_ context.Context, req *proto.ExceedsLimits
streams[partitionID] = append(streams[partitionID], Stream{
Hash: stream.StreamHash,
LastSeenAt: recordTime,
TotalSize: stream.EntriesSize + stream.StructuredMetadataSize,
TotalSize: stream.TotalSize,
})
}
@ -402,5 +400,5 @@ func (s *IngestLimits) ExceedsLimits(_ context.Context, req *proto.ExceedsLimits
s.metrics.tenantIngestedBytesTotal.WithLabelValues(req.Tenant).Add(float64(ingestedBytes))
return &proto.ExceedsLimitsResponse{results}, nil
return &proto.ExceedsLimitsResponse{Results: results}, nil
}

@ -68,9 +68,8 @@ func TestIngestLimits_ExceedsLimits(t *testing.T) {
tenantID: "tenant2",
streams: []*proto.StreamMetadata{
{
StreamHash: 0x2,
EntriesSize: 1000,
StructuredMetadataSize: 10,
StreamHash: 0x2,
TotalSize: 1010,
},
},
// expect data
@ -103,10 +102,10 @@ func TestIngestLimits_ExceedsLimits(t *testing.T) {
tenantID: "tenant1",
maxActiveStreams: 10,
streams: []*proto.StreamMetadata{
{StreamHash: 0x1, EntriesSize: 1000, StructuredMetadataSize: 10},
{StreamHash: 0x2, EntriesSize: 1000, StructuredMetadataSize: 10},
{StreamHash: 0x3, EntriesSize: 1000, StructuredMetadataSize: 10},
{StreamHash: 0x4, EntriesSize: 1000, StructuredMetadataSize: 10},
{StreamHash: 0x1, TotalSize: 1010},
{StreamHash: 0x2, TotalSize: 1010},
{StreamHash: 0x3, TotalSize: 1010},
{StreamHash: 0x4, TotalSize: 1010},
},
// expect data
expectedIngestedBytes: 4040,
@ -137,8 +136,8 @@ func TestIngestLimits_ExceedsLimits(t *testing.T) {
// request data
tenantID: "tenant1",
streams: []*proto.StreamMetadata{
{StreamHash: 0x2, EntriesSize: 1000, StructuredMetadataSize: 10},
{StreamHash: 0x4, EntriesSize: 1000, StructuredMetadataSize: 10},
{StreamHash: 0x2, TotalSize: 1010},
{StreamHash: 0x4, TotalSize: 1010},
},
// expect data
expectedIngestedBytes: 0,
@ -173,11 +172,11 @@ func TestIngestLimits_ExceedsLimits(t *testing.T) {
// request data
tenantID: "tenant1",
streams: []*proto.StreamMetadata{
{StreamHash: 0x1, EntriesSize: 1000, StructuredMetadataSize: 10},
{StreamHash: 0x2, EntriesSize: 1000, StructuredMetadataSize: 10},
{StreamHash: 0x3, EntriesSize: 1000, StructuredMetadataSize: 10},
{StreamHash: 0x4, EntriesSize: 1000, StructuredMetadataSize: 10},
{StreamHash: 0x5, EntriesSize: 1000, StructuredMetadataSize: 10},
{StreamHash: 0x1, TotalSize: 1010},
{StreamHash: 0x2, TotalSize: 1010},
{StreamHash: 0x3, TotalSize: 1010},
{StreamHash: 0x4, TotalSize: 1010},
{StreamHash: 0x5, TotalSize: 1010},
},
// expect data
expectedIngestedBytes: 3030,
@ -214,11 +213,11 @@ func TestIngestLimits_ExceedsLimits(t *testing.T) {
// request data
tenantID: "tenant1",
streams: []*proto.StreamMetadata{
{StreamHash: 0x1, EntriesSize: 1000, StructuredMetadataSize: 10},
{StreamHash: 0x2, EntriesSize: 1000, StructuredMetadataSize: 10},
{StreamHash: 0x3, EntriesSize: 1000, StructuredMetadataSize: 10},
{StreamHash: 0x4, EntriesSize: 1000, StructuredMetadataSize: 10},
{StreamHash: 0x5, EntriesSize: 1000, StructuredMetadataSize: 10},
{StreamHash: 0x1, TotalSize: 1010},
{StreamHash: 0x2, TotalSize: 1010},
{StreamHash: 0x3, TotalSize: 1010},
{StreamHash: 0x4, TotalSize: 1010},
{StreamHash: 0x5, TotalSize: 1010},
},
// expect data
expectedIngestedBytes: 5050,
@ -242,10 +241,10 @@ func TestIngestLimits_ExceedsLimits(t *testing.T) {
// request data
tenantID: "tenant1",
streams: []*proto.StreamMetadata{
{StreamHash: 0x1, EntriesSize: 1000, StructuredMetadataSize: 10},
{StreamHash: 0x2, EntriesSize: 1000, StructuredMetadataSize: 10},
{StreamHash: 0x3, EntriesSize: 1000, StructuredMetadataSize: 10},
{StreamHash: 0x4, EntriesSize: 1000, StructuredMetadataSize: 10},
{StreamHash: 0x1, TotalSize: 1010},
{StreamHash: 0x2, TotalSize: 1010},
{StreamHash: 0x3, TotalSize: 1010},
{StreamHash: 0x4, TotalSize: 1010},
},
// expect data
expectedIngestedBytes: 2020,
@ -273,10 +272,10 @@ func TestIngestLimits_ExceedsLimits(t *testing.T) {
// request data
tenantID: "tenant1",
streams: []*proto.StreamMetadata{
{StreamHash: 0x1, EntriesSize: 1000, StructuredMetadataSize: 10}, // Unassigned
{StreamHash: 0x2, EntriesSize: 1000, StructuredMetadataSize: 10}, // Assigned
{StreamHash: 0x3, EntriesSize: 1000, StructuredMetadataSize: 10}, // Unassigned
{StreamHash: 0x4, EntriesSize: 1000, StructuredMetadataSize: 10}, // Assigned but exceeds stream limit
{StreamHash: 0x1, TotalSize: 1010}, // Unassigned
{StreamHash: 0x2, TotalSize: 1010}, // Assigned
{StreamHash: 0x3, TotalSize: 1010}, // Unassigned
{StreamHash: 0x4, TotalSize: 1010}, // Assigned but exceeds stream limit
},
// expect data
expectedIngestedBytes: 1010,

@ -253,9 +253,8 @@ func (m *GetAssignedPartitionsResponse) GetAssignedPartitions() map[int32]int64
}
type StreamMetadata struct {
StreamHash uint64 `protobuf:"varint,1,opt,name=streamHash,proto3" json:"streamHash,omitempty"`
EntriesSize uint64 `protobuf:"varint,2,opt,name=entriesSize,proto3" json:"entriesSize,omitempty"`
StructuredMetadataSize uint64 `protobuf:"varint,3,opt,name=structuredMetadataSize,proto3" json:"structuredMetadataSize,omitempty"`
StreamHash uint64 `protobuf:"varint,1,opt,name=streamHash,proto3" json:"streamHash,omitempty"`
TotalSize uint64 `protobuf:"varint,2,opt,name=totalSize,proto3" json:"totalSize,omitempty"`
}
func (m *StreamMetadata) Reset() { *m = StreamMetadata{} }
@ -297,16 +296,9 @@ func (m *StreamMetadata) GetStreamHash() uint64 {
return 0
}
func (m *StreamMetadata) GetEntriesSize() uint64 {
func (m *StreamMetadata) GetTotalSize() uint64 {
if m != nil {
return m.EntriesSize
}
return 0
}
func (m *StreamMetadata) GetStructuredMetadataSize() uint64 {
if m != nil {
return m.StructuredMetadataSize
return m.TotalSize
}
return 0
}
@ -324,37 +316,36 @@ func init() {
func init() { proto.RegisterFile("pkg/limits/proto/limits.proto", fileDescriptor_aaed9e7d5298ac0f) }
var fileDescriptor_aaed9e7d5298ac0f = []byte{
// 470 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x92, 0xbf, 0x6e, 0x13, 0x41,
0x10, 0xc6, 0x6f, 0xed, 0x38, 0x11, 0x13, 0x82, 0xd0, 0x62, 0x07, 0xcb, 0x24, 0x2b, 0x6b, 0xa1,
0x70, 0x65, 0x4b, 0x26, 0x42, 0x08, 0xd1, 0x80, 0x64, 0xfe, 0x48, 0xb1, 0x84, 0x36, 0x0f, 0x80,
0x96, 0xdc, 0xc8, 0x9c, 0x72, 0xd9, 0x33, 0xbb, 0x73, 0x88, 0x50, 0x51, 0x53, 0xf1, 0x18, 0x3c,
0x07, 0x15, 0xa5, 0x0b, 0x8a, 0x94, 0xf8, 0xdc, 0x50, 0xe6, 0x11, 0x50, 0xf6, 0xce, 0x92, 0x93,
0xf8, 0x0c, 0x45, 0xaa, 0xbb, 0x99, 0xf9, 0xf4, 0x9b, 0xfd, 0x66, 0x06, 0x76, 0xc7, 0x47, 0xa3,
0x5e, 0x1c, 0x1d, 0x47, 0xe4, 0x7a, 0x63, 0x9b, 0x50, 0x52, 0x04, 0x5d, 0x1f, 0xf0, 0x9a, 0xff,
0xc8, 0xb7, 0x50, 0x1f, 0x7c, 0x3a, 0x44, 0x0c, 0xdd, 0xbe, 0xaf, 0x2a, 0xfc, 0x90, 0xa2, 0x23,
0xbe, 0x0d, 0xeb, 0x84, 0x46, 0x1b, 0x6a, 0xb2, 0x36, 0xeb, 0xdc, 0x50, 0x45, 0xc4, 0x7b, 0xb0,
0xe1, 0xc8, 0xa2, 0x3e, 0x76, 0xcd, 0x4a, 0xbb, 0xda, 0xd9, 0xec, 0x37, 0x72, 0x5e, 0xf7, 0xc0,
0x67, 0x87, 0x48, 0x3a, 0xd4, 0xa4, 0xd5, 0x5c, 0x25, 0x87, 0xd0, 0xb8, 0xd4, 0xc0, 0x8d, 0x13,
0xe3, 0x90, 0xef, 0xc1, 0x86, 0x45, 0x97, 0xc6, 0xe4, 0x9a, 0xcc, 0x93, 0x5a, 0x05, 0xe9, 0xb2,
0x3c, 0x8d, 0x49, 0xcd, 0xa5, 0x72, 0x08, 0x77, 0x96, 0xd4, 0xb9, 0x00, 0xc8, 0x1b, 0xbe, 0xd2,
0xee, 0xbd, 0x7f, 0xf2, 0x9a, 0x5a, 0xc8, 0x9c, 0xdb, 0xb1, 0xa8, 0x5d, 0x62, 0x9a, 0x95, 0x36,
0xeb, 0x6c, 0xa9, 0x22, 0x92, 0x02, 0x76, 0x5e, 0x22, 0x3d, 0x73, 0x2e, 0x1a, 0x19, 0x0c, 0xdf,
0x68, 0x4b, 0x11, 0x45, 0x89, 0x99, 0x8f, 0x41, 0xfe, 0x62, 0xb0, 0x5b, 0x22, 0x28, 0x6c, 0xc4,
0xc0, 0xf5, 0x95, 0x6a, 0xe1, 0xe8, 0x69, 0xe1, 0x68, 0x25, 0xa1, 0x7b, 0xb5, 0x34, 0x30, 0x64,
0x4f, 0xd4, 0x12, 0x6e, 0x6b, 0x00, 0x77, 0x4b, 0xe4, 0xfc, 0x36, 0x54, 0x8f, 0xf0, 0xc4, 0x7b,
0xaf, 0xa9, 0xf3, 0x5f, 0x5e, 0x87, 0xda, 0x47, 0x1d, 0xa7, 0xe8, 0x3d, 0x57, 0x55, 0x1e, 0x3c,
0xa9, 0x3c, 0x66, 0xf2, 0x2b, 0x83, 0x5b, 0x17, 0x17, 0xf6, 0xcf, 0x09, 0xb6, 0x61, 0x13, 0x0d,
0xd9, 0x08, 0xdd, 0x41, 0xf4, 0x39, 0x47, 0xae, 0xa9, 0xc5, 0x14, 0x7f, 0x04, 0xdb, 0x8e, 0x6c,
0x7a, 0x48, 0xa9, 0xc5, 0x70, 0xce, 0xf5, 0xe2, 0xaa, 0x17, 0x97, 0x54, 0xfb, 0x21, 0xd4, 0x5f,
0x9b, 0x11, 0x3a, 0xca, 0x37, 0xfa, 0xc2, 0x26, 0x86, 0xd0, 0x84, 0x7c, 0x1f, 0xb6, 0x2e, 0xac,
0x9a, 0xdf, 0x5b, 0x7e, 0x20, 0x7e, 0x53, 0xad, 0x9d, 0x92, 0xeb, 0xf1, 0x33, 0x96, 0x41, 0xff,
0x07, 0x83, 0x9b, 0x8b, 0x6d, 0xae, 0x17, 0xcf, 0x43, 0x68, 0x2c, 0xdd, 0x32, 0xbf, 0xbf, 0xfa,
0x06, 0x72, 0xfa, 0x83, 0xff, 0x39, 0x14, 0x19, 0x3c, 0xdf, 0x9b, 0x4c, 0x45, 0x70, 0x3a, 0x15,
0xc1, 0xd9, 0x54, 0xb0, 0x2f, 0x99, 0x60, 0xdf, 0x33, 0xc1, 0x7e, 0x66, 0x82, 0x4d, 0x32, 0xc1,
0x7e, 0x67, 0x82, 0xfd, 0xc9, 0x44, 0x70, 0x96, 0x09, 0xf6, 0x6d, 0x26, 0x82, 0xc9, 0x4c, 0x04,
0xa7, 0x33, 0x11, 0xbc, 0x5b, 0xf7, 0xf0, 0x87, 0x7f, 0x03, 0x00, 0x00, 0xff, 0xff, 0x50, 0x83,
0x40, 0x9b, 0x12, 0x04, 0x00, 0x00,
// 449 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x92, 0xcf, 0x6e, 0xd3, 0x40,
0x10, 0xc6, 0x77, 0x93, 0xa6, 0x55, 0x07, 0x8a, 0xd0, 0x92, 0x80, 0x15, 0xd2, 0x55, 0xb4, 0x70,
0xc8, 0x29, 0x91, 0x42, 0x0f, 0x08, 0x71, 0x01, 0x29, 0xfc, 0x91, 0x1a, 0x84, 0xb6, 0x0f, 0x80,
0x16, 0x3c, 0x0a, 0x56, 0xdd, 0x75, 0xf0, 0x8e, 0x11, 0xe5, 0xc4, 0x23, 0xf0, 0x18, 0x3c, 0x07,
0x27, 0x8e, 0x39, 0x70, 0xe8, 0x91, 0x38, 0x17, 0x8e, 0x7d, 0x04, 0xd4, 0xb5, 0x23, 0xfa, 0xc7,
0x29, 0x1c, 0x7a, 0xb2, 0x67, 0xe6, 0xd3, 0x6f, 0xfc, 0xcd, 0x67, 0xd8, 0x9e, 0xee, 0x4f, 0x06,
0x71, 0x74, 0x10, 0x91, 0x1b, 0x4c, 0xd3, 0x84, 0x92, 0xb2, 0xe8, 0xfb, 0x42, 0x34, 0xfc, 0x43,
0xbd, 0x81, 0xe6, 0xe8, 0xd3, 0x3b, 0xc4, 0xd0, 0xed, 0xfa, 0xa9, 0xc6, 0x0f, 0x19, 0x3a, 0x12,
0xb7, 0x61, 0x9d, 0xd0, 0x1a, 0x4b, 0x01, 0xef, 0xf2, 0xde, 0xa6, 0x2e, 0x2b, 0x31, 0x80, 0x0d,
0x47, 0x29, 0x9a, 0x03, 0x17, 0xd4, 0xba, 0xf5, 0xde, 0xb5, 0x61, 0xab, 0xe0, 0xf5, 0xf7, 0x7c,
0x77, 0x8c, 0x64, 0x42, 0x43, 0x46, 0x2f, 0x55, 0x6a, 0x0c, 0xad, 0x73, 0x0b, 0xdc, 0x34, 0xb1,
0x0e, 0xc5, 0x0e, 0x6c, 0xa4, 0xe8, 0xb2, 0x98, 0x5c, 0xc0, 0x3d, 0xa9, 0x5d, 0x92, 0xce, 0xcb,
0xb3, 0x98, 0xf4, 0x52, 0xaa, 0xc6, 0x70, 0xab, 0x62, 0x2e, 0x24, 0x40, 0xb1, 0xf0, 0x85, 0x71,
0xef, 0xfd, 0x27, 0xaf, 0xe9, 0x53, 0x9d, 0x13, 0x3b, 0x29, 0x1a, 0x97, 0xd8, 0xa0, 0xd6, 0xe5,
0xbd, 0x2d, 0x5d, 0x56, 0x4a, 0x42, 0xe7, 0x39, 0xd2, 0x13, 0xe7, 0xa2, 0x89, 0xc5, 0xf0, 0xb5,
0x49, 0x29, 0xa2, 0x28, 0xb1, 0xcb, 0x33, 0xa8, 0x9f, 0x1c, 0xb6, 0x57, 0x08, 0x4a, 0x1b, 0x31,
0x08, 0x73, 0x61, 0x5a, 0x3a, 0x7a, 0x5c, 0x3a, 0xba, 0x94, 0xd0, 0xbf, 0x38, 0x1a, 0x59, 0x4a,
0x0f, 0x75, 0x05, 0xb7, 0x3d, 0x82, 0x3b, 0x2b, 0xe4, 0xe2, 0x26, 0xd4, 0xf7, 0xf1, 0xd0, 0x7b,
0x6f, 0xe8, 0x93, 0x57, 0xd1, 0x84, 0xc6, 0x47, 0x13, 0x67, 0xe8, 0x3d, 0xd7, 0x75, 0x51, 0x3c,
0xaa, 0x3d, 0xe4, 0xea, 0x15, 0xdc, 0x38, 0x9b, 0xd7, 0x3f, 0x0f, 0xd8, 0x81, 0x4d, 0x4a, 0xc8,
0xc4, 0x7b, 0xd1, 0xe7, 0x82, 0xb7, 0xa6, 0xff, 0x36, 0x86, 0x21, 0x34, 0x5f, 0xda, 0x09, 0x3a,
0x2a, 0x42, 0x79, 0x96, 0x26, 0x96, 0xd0, 0x86, 0x62, 0x17, 0xb6, 0xce, 0xa4, 0x25, 0xee, 0x56,
0x67, 0xec, 0x8f, 0xdd, 0xee, 0xac, 0xf8, 0x01, 0xfc, 0x99, 0x14, 0x1b, 0x7e, 0xe7, 0x70, 0xfd,
0xf4, 0x9a, 0xab, 0xc5, 0x8b, 0x10, 0x5a, 0x95, 0x41, 0x89, 0x7b, 0x97, 0xc7, 0x58, 0xd0, 0xef,
0xff, 0x4f, 0xd6, 0x8a, 0x3d, 0xdd, 0x99, 0xcd, 0x25, 0x3b, 0x9a, 0x4b, 0x76, 0x3c, 0x97, 0xfc,
0x4b, 0x2e, 0xf9, 0xb7, 0x5c, 0xf2, 0x1f, 0xb9, 0xe4, 0xb3, 0x5c, 0xf2, 0x5f, 0xb9, 0xe4, 0xbf,
0x73, 0xc9, 0x8e, 0x73, 0xc9, 0xbf, 0x2e, 0x24, 0x9b, 0x2d, 0x24, 0x3b, 0x5a, 0x48, 0xf6, 0x76,
0xdd, 0xc3, 0x1f, 0xfc, 0x09, 0x00, 0x00, 0xff, 0xff, 0x14, 0x3f, 0x94, 0x86, 0xd5, 0x03, 0x00,
0x00,
}
func (this *ExceedsLimitsRequest) Equal(that interface{}) bool {
@ -517,10 +508,7 @@ func (this *StreamMetadata) Equal(that interface{}) bool {
if this.StreamHash != that1.StreamHash {
return false
}
if this.EntriesSize != that1.EntriesSize {
return false
}
if this.StructuredMetadataSize != that1.StructuredMetadataSize {
if this.TotalSize != that1.TotalSize {
return false
}
return true
@ -596,11 +584,10 @@ func (this *StreamMetadata) GoString() string {
if this == nil {
return "nil"
}
s := make([]string, 0, 7)
s := make([]string, 0, 6)
s = append(s, "&proto.StreamMetadata{")
s = append(s, "StreamHash: "+fmt.Sprintf("%#v", this.StreamHash)+",\n")
s = append(s, "EntriesSize: "+fmt.Sprintf("%#v", this.EntriesSize)+",\n")
s = append(s, "StructuredMetadataSize: "+fmt.Sprintf("%#v", this.StructuredMetadataSize)+",\n")
s = append(s, "TotalSize: "+fmt.Sprintf("%#v", this.TotalSize)+",\n")
s = append(s, "}")
return strings.Join(s, "")
}
@ -996,13 +983,8 @@ func (m *StreamMetadata) MarshalToSizedBuffer(dAtA []byte) (int, error) {
_ = i
var l int
_ = l
if m.StructuredMetadataSize != 0 {
i = encodeVarintLimits(dAtA, i, uint64(m.StructuredMetadataSize))
i--
dAtA[i] = 0x18
}
if m.EntriesSize != 0 {
i = encodeVarintLimits(dAtA, i, uint64(m.EntriesSize))
if m.TotalSize != 0 {
i = encodeVarintLimits(dAtA, i, uint64(m.TotalSize))
i--
dAtA[i] = 0x10
}
@ -1109,11 +1091,8 @@ func (m *StreamMetadata) Size() (n int) {
if m.StreamHash != 0 {
n += 1 + sovLimits(uint64(m.StreamHash))
}
if m.EntriesSize != 0 {
n += 1 + sovLimits(uint64(m.EntriesSize))
}
if m.StructuredMetadataSize != 0 {
n += 1 + sovLimits(uint64(m.StructuredMetadataSize))
if m.TotalSize != 0 {
n += 1 + sovLimits(uint64(m.TotalSize))
}
return n
}
@ -1201,8 +1180,7 @@ func (this *StreamMetadata) String() string {
}
s := strings.Join([]string{`&StreamMetadata{`,
`StreamHash:` + fmt.Sprintf("%v", this.StreamHash) + `,`,
`EntriesSize:` + fmt.Sprintf("%v", this.EntriesSize) + `,`,
`StructuredMetadataSize:` + fmt.Sprintf("%v", this.StructuredMetadataSize) + `,`,
`TotalSize:` + fmt.Sprintf("%v", this.TotalSize) + `,`,
`}`,
}, "")
return s
@ -1767,28 +1745,9 @@ func (m *StreamMetadata) Unmarshal(dAtA []byte) error {
}
case 2:
if wireType != 0 {
return fmt.Errorf("proto: wrong wireType = %d for field EntriesSize", wireType)
}
m.EntriesSize = 0
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowLimits
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
m.EntriesSize |= uint64(b&0x7F) << shift
if b < 0x80 {
break
}
}
case 3:
if wireType != 0 {
return fmt.Errorf("proto: wrong wireType = %d for field StructuredMetadataSize", wireType)
return fmt.Errorf("proto: wrong wireType = %d for field TotalSize", wireType)
}
m.StructuredMetadataSize = 0
m.TotalSize = 0
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowLimits
@ -1798,7 +1757,7 @@ func (m *StreamMetadata) Unmarshal(dAtA []byte) error {
}
b := dAtA[iNdEx]
iNdEx++
m.StructuredMetadataSize |= uint64(b&0x7F) << shift
m.TotalSize |= uint64(b&0x7F) << shift
if b < 0x80 {
break
}

@ -33,6 +33,5 @@ message GetAssignedPartitionsResponse {
message StreamMetadata {
uint64 streamHash = 1;
uint64 entriesSize = 2;
uint64 structuredMetadataSize = 3;
uint64 totalSize = 2;
}

@ -67,16 +67,14 @@ func BenchmarkStreamMetadata_Store(b *testing.B) {
updateTime := now.Add(time.Duration(i) * time.Second)
metadata := &proto.StreamMetadata{
StreamHash: uint64(streamIdx),
EntriesSize: 1000,
StructuredMetadataSize: 500,
StreamHash: uint64(streamIdx),
TotalSize: 1500,
}
bucketStart := updateTime.Truncate(bucketDuration).UnixNano()
bucketCutOff := updateTime.Add(-rateWindow).UnixNano()
totalSize := metadata.EntriesSize + metadata.StructuredMetadataSize
s.Store(tenant, partition, metadata.StreamHash, totalSize, updateTime.UnixNano(), bucketStart, bucketCutOff)
s.Store(tenant, partition, metadata.StreamHash, metadata.TotalSize, updateTime.UnixNano(), bucketStart, bucketCutOff)
}
})
@ -92,16 +90,14 @@ func BenchmarkStreamMetadata_Store(b *testing.B) {
updateTime := now.Add(time.Duration(i) * time.Second)
metadata := &proto.StreamMetadata{
StreamHash: uint64(streamIdx),
EntriesSize: 1000,
StructuredMetadataSize: 500,
StreamHash: uint64(streamIdx),
TotalSize: 1500,
}
bucketStart := updateTime.Truncate(bucketDuration).UnixNano()
bucketCutOff := updateTime.Add(-rateWindow).UnixNano()
totalSize := metadata.EntriesSize + metadata.StructuredMetadataSize
s.Store(tenant, partition, metadata.StreamHash, totalSize, updateTime.UnixNano(), bucketStart, bucketCutOff)
s.Store(tenant, partition, metadata.StreamHash, metadata.TotalSize, updateTime.UnixNano(), bucketStart, bucketCutOff)
}
})
@ -120,16 +116,14 @@ func BenchmarkStreamMetadata_Store(b *testing.B) {
updateTime := now.Add(time.Duration(i) * time.Second)
metadata := &proto.StreamMetadata{
StreamHash: uint64(streamIdx),
EntriesSize: 1000,
StructuredMetadataSize: 500,
StreamHash: uint64(streamIdx),
TotalSize: 1500,
}
bucketStart := updateTime.Truncate(bucketDuration).UnixNano()
bucketCutOff := updateTime.Add(-rateWindow).UnixNano()
totalSize := metadata.EntriesSize + metadata.StructuredMetadataSize
s.Store(tenant, partition, metadata.StreamHash, totalSize, updateTime.UnixNano(), bucketStart, bucketCutOff)
s.Store(tenant, partition, metadata.StreamHash, metadata.TotalSize, updateTime.UnixNano(), bucketStart, bucketCutOff)
i++
}
})
@ -147,16 +141,14 @@ func BenchmarkStreamMetadata_Store(b *testing.B) {
updateTime := now.Add(time.Duration(i) * time.Second)
metadata := &proto.StreamMetadata{
StreamHash: uint64(streamIdx),
EntriesSize: 1000,
StructuredMetadataSize: 500,
StreamHash: uint64(streamIdx),
TotalSize: 1500,
}
bucketStart := updateTime.Truncate(bucketDuration).UnixNano()
bucketCutOff := updateTime.Add(-rateWindow).UnixNano()
totalSize := metadata.EntriesSize + metadata.StructuredMetadataSize
s.Store(tenant, partition, metadata.StreamHash, totalSize, updateTime.UnixNano(), bucketStart, bucketCutOff)
s.Store(tenant, partition, metadata.StreamHash, metadata.TotalSize, updateTime.UnixNano(), bucketStart, bucketCutOff)
i++
}
})
@ -228,16 +220,14 @@ func BenchmarkStreamMetadata_UsageAndStore(b *testing.B) {
updateTime := now.Add(time.Duration(partition) * time.Second)
metadata := &proto.StreamMetadata{
StreamHash: uint64(streamIdx),
EntriesSize: 1000,
StructuredMetadataSize: 500,
StreamHash: uint64(streamIdx),
TotalSize: 1500,
}
bucketStart := updateTime.Truncate(bucketDuration).UnixNano()
bucketCutOff := updateTime.Add(-rateWindow).UnixNano()
totalSize := metadata.EntriesSize + metadata.StructuredMetadataSize
s.Store(tenant, int32(partition), metadata.StreamHash, totalSize, updateTime.UnixNano(), bucketStart, bucketCutOff)
s.Store(tenant, int32(partition), metadata.StreamHash, metadata.TotalSize, updateTime.UnixNano(), bucketStart, bucketCutOff)
}
b.Run(fmt.Sprintf("%s_create", bm.name), func(b *testing.B) {
@ -262,16 +252,14 @@ func BenchmarkStreamMetadata_UsageAndStore(b *testing.B) {
updateTime := now.Add(time.Duration(i) * time.Second)
metadata := &proto.StreamMetadata{
StreamHash: uint64(streamIdx),
EntriesSize: 1000,
StructuredMetadataSize: 500,
StreamHash: uint64(streamIdx),
TotalSize: 1500,
}
bucketStart := updateTime.Truncate(bucketDuration).UnixNano()
bucketCutOff := updateTime.Add(-rateWindow).UnixNano()
totalSize := metadata.EntriesSize + metadata.StructuredMetadataSize
s.Store(tenant, partition, metadata.StreamHash, totalSize, updateTime.UnixNano(), bucketStart, bucketCutOff)
s.Store(tenant, partition, metadata.StreamHash, metadata.TotalSize, updateTime.UnixNano(), bucketStart, bucketCutOff)
}
}(i)
}

@ -142,9 +142,8 @@ func TestStreamMetadata_Store(t *testing.T) {
partitionID: 0,
lastSeenAt: time.Unix(100, 0),
record: &proto.StreamMetadata{
StreamHash: 123,
EntriesSize: 1000,
StructuredMetadataSize: 500,
StreamHash: 123,
TotalSize: 1500,
},
expected: map[string]map[int32]map[uint64]Stream{
"tenant1": {
@ -186,9 +185,8 @@ func TestStreamMetadata_Store(t *testing.T) {
tenantID: "tenant1",
partitionID: 1,
record: &proto.StreamMetadata{
StreamHash: 456,
EntriesSize: 2000,
StructuredMetadataSize: 1000,
StreamHash: 456,
TotalSize: 3000,
},
lastSeenAt: time.Unix(200, 0),
expected: map[string]map[int32]map[uint64]Stream{
@ -240,9 +238,8 @@ func TestStreamMetadata_Store(t *testing.T) {
tenantID: "tenant1",
partitionID: 0,
record: &proto.StreamMetadata{
StreamHash: 123,
EntriesSize: 3000,
StructuredMetadataSize: 1500,
StreamHash: 123,
TotalSize: 4500,
},
lastSeenAt: time.Unix(300, 0),
expected: map[string]map[int32]map[uint64]Stream{
@ -265,9 +262,8 @@ func TestStreamMetadata_Store(t *testing.T) {
name: "update existing bucket",
tenantID: "tenant1",
record: &proto.StreamMetadata{
StreamHash: 888,
EntriesSize: 1000,
StructuredMetadataSize: 500,
StreamHash: 888,
TotalSize: 1500,
},
lastSeenAt: time.Unix(852, 0),
metadata: &streamMetadata{
@ -308,9 +304,8 @@ func TestStreamMetadata_Store(t *testing.T) {
name: "clean up buckets outside rate window",
tenantID: "tenant1",
record: &proto.StreamMetadata{
StreamHash: 999,
EntriesSize: 2000,
StructuredMetadataSize: 1000,
StreamHash: 999,
TotalSize: 3000,
},
lastSeenAt: time.Unix(1000, 0), // Current time reference
metadata: &streamMetadata{
@ -354,9 +349,8 @@ func TestStreamMetadata_Store(t *testing.T) {
name: "update same minute bucket",
tenantID: "tenant1",
record: &proto.StreamMetadata{
StreamHash: 555,
EntriesSize: 1000,
StructuredMetadataSize: 500,
StreamHash: 555,
TotalSize: 1500,
},
lastSeenAt: time.Unix(1100, 0),
metadata: &streamMetadata{
@ -398,11 +392,10 @@ func TestStreamMetadata_Store(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
totalSize := tt.record.EntriesSize + tt.record.StructuredMetadataSize
bucketStart := tt.lastSeenAt.Truncate(bucketDuration).UnixNano()
bucketCutOff := tt.lastSeenAt.Add(-rateWindow).UnixNano()
tt.metadata.Store(tt.tenantID, tt.partitionID, tt.record.StreamHash, totalSize, tt.lastSeenAt.UnixNano(), bucketStart, bucketCutOff)
tt.metadata.Store(tt.tenantID, tt.partitionID, tt.record.StreamHash, tt.record.TotalSize, tt.lastSeenAt.UnixNano(), bucketStart, bucketCutOff)
tt.metadata.All(func(tenant string, partitionID int32, stream Stream) {
require.Contains(t, tt.expected, tenant)
@ -440,13 +433,11 @@ func TestStreamMetadata_Store_Concurrent(t *testing.T) {
}
record := &proto.StreamMetadata{
StreamHash: uint64(i),
EntriesSize: 1000,
StructuredMetadataSize: 500,
StreamHash: uint64(i),
TotalSize: 1500,
}
totalSize := record.EntriesSize + record.StructuredMetadataSize
m.Store(tenantID, partitionID, record.StreamHash, totalSize, lastSeenAt.UnixNano(), bucketStart, bucketCutOff)
m.Store(tenantID, partitionID, record.StreamHash, record.TotalSize, lastSeenAt.UnixNano(), bucketStart, bucketCutOff)
}(i)
}
wg.Wait()

@ -94,8 +94,8 @@ func (s *Generator) sendStreamsToKafka(ctx context.Context, streams []distributo
}
metadata := proto.StreamMetadata{
StreamHash: stream.HashKeyNoShard,
EntriesSize: logSize,
StreamHash: stream.HashKeyNoShard,
TotalSize: logSize,
}
b, err := metadata.Marshal()
if err != nil {

Loading…
Cancel
Save