Skip to content

Deduplicated messages sensors: LOGBROKER-8733 #3147

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 4 commits into from
Apr 1, 2024
Merged

Deduplicated messages sensors: LOGBROKER-8733 #3147

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
6bb22a0
Deduplicated messages sensors: LOGBROKER-8733
FloatingCrowbar Mar 25, 2024
a7e1aa5
cleanup
FloatingCrowbar Mar 26, 2024
741f19a
Fix bad merge
FloatingCrowbar Mar 26, 2024
01c07fc
Fix tests
FloatingCrowbar Mar 26, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions ydb/core/persqueue/partition.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -762,6 +762,9 @@ class TPartition : public TActorBootstrapped<TPartition> {
TPartitionCounterWrapper MsgsWrittenTotal;
TPartitionCounterWrapper MsgsWrittenGrpc;

NKikimr::NPQ::TMultiCounter MsgsDiscarded;
NKikimr::NPQ::TMultiCounter BytesDiscarded;

// Writing blob with topic quota variables
ui64 TopicQuotaRequestCookie = 0;
ui64 NextTopicWriteQuotaRequestCookie = 1;
Expand Down
14 changes: 14 additions & 0 deletions ydb/core/persqueue/partition_init.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -802,6 +802,10 @@ void TPartition::SetupTopicCounters(const TActorContext& ctx) {
MsgsWrittenTotal.Setup(
IsSupportive(), true,
NKikimr::NPQ::TMultiCounter(subGroup, labels, {}, {"MessagesWritten" + txSuffix}, true));
if (IsLocalDC) {
MsgsDiscarded = NKikimr::NPQ::TMultiCounter(subGroup, labels, {}, {"DiscardedMessages"}, true);
BytesDiscarded = NKikimr::NPQ::TMultiCounter(subGroup, labels, {}, {"DiscardedBytes"}, true);
}

TVector<NPersQueue::TPQLabelsInfo> aggr = {{{{"Account", TopicConverter->GetAccount()}}, {"total"}}};
ui32 border = AppData(ctx)->PQConfig.GetWriteLatencyBigMs();
Expand Down Expand Up @@ -904,6 +908,16 @@ void TPartition::SetupStreamCounters(const TActorContext& ctx) {
NPersQueue::GetCountersForTopic(counters, IsServerless), {}, subgroups,
{"topic.write." + messagesSuffix}, true, "name"));

MsgsDiscarded = NKikimr::NPQ::TMultiCounter(
NPersQueue::GetCountersForTopic(counters, IsServerless), {}, subgroups,
{"topic.write.discarded_messages"}, true, "name");
BytesDiscarded = NKikimr::NPQ::TMultiCounter(
NPersQueue::GetCountersForTopic(counters, IsServerless), {}, subgroups,
{"topic.write.discarded_bytes"} , true, "name");


BytesWrittenUncompressed = NKikimr::NPQ::TMultiCounter(

BytesWrittenUncompressed.Setup(
IsSupportive(), false,
NKikimr::NPQ::TMultiCounter(
Expand Down
2 changes: 2 additions & 0 deletions ydb/core/persqueue/partition_write.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -939,7 +939,9 @@ TPartition::ProcessResult TPartition::ProcessRequest(TWriteMsg& p, ProcessParame
);

TabletCounters.Cumulative()[COUNTER_PQ_WRITE_ALREADY].Increment(1);
MsgsDiscarded.Inc();
TabletCounters.Cumulative()[COUNTER_PQ_WRITE_BYTES_ALREADY].Increment(p.Msg.Data.size());
BytesDiscarded.Inc(p.Msg.Data.size());
} else {
TabletCounters.Cumulative()[COUNTER_PQ_WRITE_SMALL_OFFSET].Increment(1);
TabletCounters.Cumulative()[COUNTER_PQ_WRITE_BYTES_SMALL_OFFSET].Increment(p.Msg.Data.size());
Expand Down
7 changes: 5 additions & 2 deletions ydb/core/persqueue/ut/counters_ut.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,6 +84,8 @@ Y_UNIT_TEST(Partition) {
CmdWrite(0, "sourceid0", TestData(), tc, false, {}, true);
CmdWrite(0, "sourceid1", TestData(), tc, false);
CmdWrite(0, "sourceid2", TestData(), tc, false);
CmdWrite(0, "sourceid1", TestData(), tc, false);
CmdWrite(0, "sourceid2", TestData(), tc, false);
PQGetPartInfo(0, 30, tc);


Expand All @@ -94,15 +96,15 @@ Y_UNIT_TEST(Partition) {
dbGroup->OutputHtml(countersStr);
TString referenceCounters = NResource::Find(TStringBuf("counters_pqproxy.html"));

UNIT_ASSERT_EQUAL(countersStr.Str() + "\n", referenceCounters);
UNIT_ASSERT_VALUES_EQUAL(countersStr.Str() + "\n", referenceCounters);
}

{
auto counters = tc.Runtime->GetAppData(0).Counters;
auto dbGroup = GetServiceCounters(counters, "datastreams");
TStringStream countersStr;
dbGroup->OutputHtml(countersStr);
UNIT_ASSERT_EQUAL(countersStr.Str(), "<pre></pre>");
UNIT_ASSERT_VALUES_EQUAL(countersStr.Str(), "<pre></pre>");
}
}

Expand Down Expand Up @@ -173,6 +175,7 @@ Y_UNIT_TEST(PartitionFirstClass) {
CmdWrite(0, "sourceid0", TestData(), tc, false, {}, true);
CmdWrite(0, "sourceid1", TestData(), tc, false);
CmdWrite(0, "sourceid2", TestData(), tc, false);
CmdWrite(0, "sourceid0", TestData(), tc, false);
PQGetPartInfo(0, 30, tc);

{
Expand Down
4 changes: 3 additions & 1 deletion ydb/core/persqueue/ut/resources/counters_datastreams.html
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,8 @@
name=api.grpc.topic.stream_write.bytes: 540
name=api.grpc.topic.stream_write.messages: 30
name=topic.write.bytes: 540
name=topic.write.discarded_bytes: 90
name=topic.write.discarded_messages: 10
name=topic.write.messages: 30
name=topic.write.uncompressed_bytes: 270

Expand Down Expand Up @@ -55,7 +57,7 @@
bin=99999999: 0

name=topic.write.partition_throttled_milliseconds:
bin=0: 30
bin=0: 40
bin=1: 0
bin=10: 0
bin=100: 0
Expand Down
28 changes: 19 additions & 9 deletions ydb/core/persqueue/ut/resources/counters_pqproxy.html
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,7 @@
OriginDC=Dc1:

sensor=PartitionWriteQuotaWaitOriginal:
Interval=0ms: 30
Interval=0ms: 50
Interval=10000ms: 0
Interval=1000ms: 0
Interval=100ms: 0
Expand All @@ -67,7 +67,7 @@
OriginDC=cluster:

sensor=PartitionWriteQuotaWaitOriginal:
Interval=0ms: 30
Interval=0ms: 50
Interval=10000ms: 0
Interval=1000ms: 0
Interval=100ms: 0
Expand All @@ -88,7 +88,7 @@
OriginDC=cluster:

sensor=PartitionWriteQuotaWaitOriginal:
Interval=0ms: 30
Interval=0ms: 50
Interval=10000ms: 0
Interval=1000ms: 0
Interval=100ms: 0
Expand All @@ -111,7 +111,7 @@
OriginDC=cluster:

sensor=PartitionWriteQuotaWaitOriginal:
Interval=0ms: 30
Interval=0ms: 50
Interval=10000ms: 0
Interval=1000ms: 0
Interval=100ms: 0
Expand All @@ -136,7 +136,7 @@
OriginDC=cluster:

sensor=PartitionWriteQuotaWaitOriginal:
Interval=0ms: 30
Interval=0ms: 50
Interval=10000ms: 0
Interval=1000ms: 0
Interval=100ms: 0
Expand Down Expand Up @@ -470,17 +470,21 @@
TopicPath=asdfgs/topic:

ClientDC=Unknown:
sensor=BytesWrittenFromDC: 1560
sensor=BytesWrittenFromDC: 2600

OriginDC=Dc1:
sensor=BytesWrittenOriginal: 540
sensor=CompactedBytesWrittenOriginal: 747
sensor=DiscardedBytes: 180
sensor=DiscardedMessages: 20
sensor=MessagesWrittenOriginal: 30
sensor=UncompressedBytesWrittenOriginal: 270

OriginDC=cluster:
sensor=BytesWrittenOriginal: 540
sensor=CompactedBytesWrittenOriginal: 747
sensor=DiscardedBytes: 180
sensor=DiscardedMessages: 20
sensor=MessagesWrittenOriginal: 30
sensor=UncompressedBytesWrittenOriginal: 270

Expand All @@ -489,11 +493,13 @@
TopicPath=total:

ClientDC=Unknown:
sensor=BytesWrittenFromDC: 1560
sensor=BytesWrittenFromDC: 2600

OriginDC=cluster:
sensor=BytesWrittenOriginal: 540
sensor=CompactedBytesWrittenOriginal: 747
sensor=DiscardedBytes: 180
sensor=DiscardedMessages: 20
sensor=MessagesWrittenOriginal: 30
sensor=UncompressedBytesWrittenOriginal: 270

Expand All @@ -504,11 +510,13 @@
TopicPath=total:

ClientDC=Unknown:
sensor=BytesWrittenFromDC: 1560
sensor=BytesWrittenFromDC: 2600

OriginDC=cluster:
sensor=BytesWrittenOriginal: 540
sensor=CompactedBytesWrittenOriginal: 747
sensor=DiscardedBytes: 180
sensor=DiscardedMessages: 20
sensor=MessagesWrittenOriginal: 30
sensor=UncompressedBytesWrittenOriginal: 270

Expand All @@ -521,11 +529,13 @@
TopicPath=total:

ClientDC=Unknown:
sensor=BytesWrittenFromDC: 1560
sensor=BytesWrittenFromDC: 2600

OriginDC=cluster:
sensor=BytesWrittenOriginal: 540
sensor=CompactedBytesWrittenOriginal: 747
sensor=DiscardedBytes: 180
sensor=DiscardedMessages: 20
sensor=MessagesWrittenOriginal: 30
sensor=UncompressedBytesWrittenOriginal: 270

Expand Down