merge to stable-25-3: Skip metering report with ydb_size=0 and topic category (#26080)

LOGBROKER-9983

Author: ubyte

ydb/core/persqueue/pqtablet/metering_sink.cpp

@@ -167,16 +167,19 @@ const TMeteringSink::FlushParameters TMeteringSink::GetFlushParameters(const EMe
     case EMeteringJson::UsedStorageV1: {
         ui64 duration = (now - lastFlush).MilliSeconds();
         ui64 avgUsage = duration > 0 ? CurrentUsedStorage_ * 1_MB * 1000 / duration : 0;
+        ui64 quantity = (avgUsage > 0) ? 1 : 0;
 
         CurrentUsedStorage_ = 0;
 
         return TMeteringSink::FlushParameters(
             "used_storage",
             "ydb.serverless.v1",
-            "byte*second"
+            "byte*second",
+            quantity
         ).withTags({
                 {"ydb_size", avgUsage}
             })
+        .withImplicitZero()
         .withVersion("1.0.0");
     }
 
@@ -195,7 +198,7 @@ void TMeteringSink::Flush(TInstant now, bool force) {
             continue;
         }
 
-        auto parameters = GetFlushParameters(whichOne, now, lastFlush);
+        const auto parameters = GetFlushParameters(whichOne, now, lastFlush);
 
         if (parameters.OneFlush) {
             const auto isTimeToFlushUnits = now.Hours() > lastFlush.Hours();
@@ -218,13 +221,15 @@ void TMeteringSink::Flush(TInstant now, bool force) {
             auto interval = TInstant::Hours(lastFlush.Hours()) + Parameters_.FlushLimit;
 
             auto tryFlush = [&](TInstant start, TInstant finish) {
-                const auto metricsJson = GetMeteringJson(
-                    parameters,
-                    parameters.Quantity * (finish.Seconds() - start.Seconds()),
-                    start,
-                    finish,
-                    now);
-                FlushFunction_(metricsJson);
+                if (parameters.Quantity > 0 || !parameters.ImplicitZero) {
+                    const auto metricsJson = GetMeteringJson(
+                        parameters,
+                        parameters.Quantity * (finish.Seconds() - start.Seconds()),
+                        start,
+                        finish,
+                        now);
+                    FlushFunction_(metricsJson);
+                }
 
                 lastFlush = finish;
             };

ydb/core/persqueue/pqtablet/metering_sink.h

@@ -79,12 +79,19 @@ class TMeteringSink {
             return *this;
         }
 
+        // skip flush, if quantity is zero
+        FlushParameters& withImplicitZero() {
+            ImplicitZero = true;
+            return *this;
+        }
+
         TString Name;
         TString Schema;
         TString Units;
         ui64 Quantity;
         THashMap<TString, ui64> Tags;
         bool OneFlush = false;
+        bool ImplicitZero = false;
         TString Version = "v1";
     };
 

ydb/core/persqueue/ut/metering_sink_ut.cpp

@@ -202,6 +202,34 @@ Y_UNIT_TEST(UsedStorageV1) {
     UNIT_ASSERT_VALUES_EQUAL(fullMetering, referenceStorageJson);
 }
 
+Y_UNIT_TEST(UnusedStorageV1) {
+    const ui64 creationTs = 1651752943168786;
+    const ui64 flushTs    = 1651754943168786;
+    const ui32 partitions = 7;
+    const ui64 reservedSpace = 42_GB;
+
+    TMeteringSink meteringSink;
+    meteringSink.Create(TInstant::FromValue(creationTs), {
+            .FlushInterval = TDuration::Seconds(10),
+            .TabletId = "tabletId",
+            .YcCloudId = "cloudId",
+            .YcFolderId = "folderId",
+            .YdbDatabaseId = "databaseId",
+            .StreamName = "streamName",
+            .ResourceId = "streamPath",
+            .PartitionsSize = partitions,
+            .ReservedSpace = reservedSpace,
+        }, {EMeteringJson::UsedStorageV1}, [&](TString json) {
+            UNIT_FAIL("Flush should not be called");
+            Y_UNUSED(json);
+        });
+
+    const ui32 quantity = 0;
+
+    meteringSink.IncreaseQuantity(EMeteringJson::UsedStorageV1, quantity);
+    meteringSink.MayFlushForcibly(TInstant::FromValue(flushTs));
+}
+
 } // Y_UNIT_TEST_SUITE(MeteringSink)
 
 } // namespace NKikimr::NPQ

ydb/core/tx/schemeshard/ut_cdc_stream/ut_cdc_stream.cpp

@@ -1134,6 +1134,7 @@ Y_UNIT_TEST_SUITE(TCdcStreamTests) {
         TTestBasicRuntime runtime;
         TTestEnv env(runtime, TTestEnvOptions()
             .EnableProtoSourceIdInfo(true)
+            .EnableChangefeedInitialScan(true)
             .EnablePqBilling(serverless));
         ui64 txId = 100;
 
@@ -1207,20 +1208,28 @@ Y_UNIT_TEST_SUITE(TCdcStreamTests) {
 
         TestCreateTable(runtime, schemeShard, ++txId, dbName, R"(
             Name: "Table"
-            Columns { Name: "key" Type: "Uint64" }
-            Columns { Name: "value" Type: "Uint64" }
+            Columns { Name: "key" Type: "Uint32" }
+            Columns { Name: "value" Type: "Utf8" }
             KeyColumnNames: ["key"]
             UniformPartitionsCount: 2
         )");
         env.TestWaitNotification(runtime, txId, schemeShard);
 
         runtime.SetLogPriority(NKikimrServices::PERSQUEUE, NLog::PRI_NOTICE);
+        {
+            // should write at least 16MiB to exceed the non-metered limit of topic's partition
+            const TString value = TString(500_KB, 'x');
+            for (size_t i = 0; i < 100; ++i) {
+                WriteRow(runtime, schemeShard, ++txId, dbName + "/Table", 0, i, value);
+            }
+        }
+
         TVector<TString> meteringRecords;
         runtime.SetObserverFunc([&meteringRecords](TAutoPtr<IEventHandle>& ev) {
             if (ev->GetTypeRewrite() != NMetering::TEvMetering::EvWriteMeteringJson) {
                 return TTestActorRuntime::EEventAction::PROCESS;
             }
-
+            Cerr << "GOT METERING RECORD: " << ev->Get<NMetering::TEvMetering::TEvWriteMeteringJson>()->MeteringJson << Endl;
             meteringRecords.push_back(ev->Get<NMetering::TEvMetering::TEvWriteMeteringJson>()->MeteringJson);
             return TTestActorRuntime::EEventAction::PROCESS;
         });
@@ -1229,8 +1238,9 @@ Y_UNIT_TEST_SUITE(TCdcStreamTests) {
             TableName: "Table"
             StreamDescription {
               Name: "Stream"
-              Mode: ECdcStreamModeKeysOnly
+              Mode: ECdcStreamModeNewImage
               Format: ECdcStreamFormatProto
+              State: ECdcStreamStateScan
             }
         )");
         env.TestWaitNotification(runtime, txId, schemeShard);
@@ -1239,26 +1249,22 @@ Y_UNIT_TEST_SUITE(TCdcStreamTests) {
             env.SimulateSleep(runtime, TDuration::Seconds(10));
         }
 
-        for (const auto& rec : meteringRecords) {
-            Cerr << "GOT METERING: " << rec << "\n";
-        }
-
         UNIT_ASSERT_VALUES_EQUAL(meteringRecords.size(), (serverless ? 3 : 0));
 
         if (!meteringRecords) {
             return;
         }
 
         NJson::TJsonValue json;
-        NJson::ReadJsonTree(meteringRecords[0], &json, true);
+        NJson::ReadJsonTree(meteringRecords.back(), &json, true);
         auto& map = json.GetMap();
         UNIT_ASSERT(map.contains("schema"));
         UNIT_ASSERT(map.contains("resource_id"));
         UNIT_ASSERT(map.contains("tags"));
         UNIT_ASSERT(map.find("tags")->second.GetMap().contains("ydb_size"));
         UNIT_ASSERT_VALUES_EQUAL(map.find("schema")->second.GetString(), "ydb.serverless.v1");
         UNIT_ASSERT_VALUES_EQUAL(map.find("resource_id")->second.GetString(), Sprintf("%s/Table/Stream/streamImpl", dbName.c_str()));
-        UNIT_ASSERT_VALUES_EQUAL(map.find("tags")->second.GetMap().find("ydb_size")->second.GetInteger(), 0);
+        UNIT_ASSERT_GT(map.find("tags")->second.GetMap().find("ydb_size")->second.GetInteger(), 0);
     }
 
     Y_UNIT_TEST(MeteringServerless) {