LOGBROKER-7166 Extend simulate_datastream method for new YDS billing plan

Introduce simulate method for new YDS billing plan
%%
mokhotskii@lbk-dev-02:~$ curl -v -X POST --header 'Accept: application/json' 'localhost:8788/ydbc/cloud-preprod/v1/simulate_datastream?databaseId=cc89j6tm0uq9q2ojq6fh&shards=1&writeQuotaKbps=512&storageLimitMb=1024'
* Hostname was NOT found in DNS cache
*   Trying ::1...
* Connected to localhost (::1) port 8788 (#0)
> POST /ydbc/cloud-preprod/v1/simulate_datastream?databaseId=cc89j6tm0uq9q2ojq6fh&shards=1&writeQuotaKbps=512&storageLimitMb=1024 HTTP/1.1
> User-Agent: curl/7.35.0
> Host: localhost:8788
> Authorization: Bearer t1.9eudmZ2UjpCamZ2Pi5edkZWLlJySm-3rnZmdxs3MnpuXxo3Oic2cicmax5bl8_cVX3dq-e8mE3p-_d3z91UNdWr57yYTen79._BHOALBYkhQyDiMps4TWNpALDpuQPbphUbZRVlpBKnRji7-1E7l2ztA0aI5T2G-PsXN_uFJsk7zpqcTTT8tqBw
> Accept: application/json
>
 POST /ydbc/cloud-preprod/simulate_datastream?databaseId=cc89j6tm0uq9q2ojq6fh&shards=1&writeQuota=512000&retentionHours=24 HTTP/1.1
> User-Agent: curl/7.35.0
> Host: localhost:8788
> Authorization: Bearer t1.9eudmZ2UjpCamZ2Pi5edkZWLlJySm-3rnZmdxs3MnpuXxo3Oic2cicmax5bl8_cVX3dq-e8mE3p-_d3z91UNdWr57yYTen79._BHOALBYkhQyDiMps4TWNpALDpuQPbphUbZRVlpBKnRji7-1E7l2ztA0aI5T2G-PsXN_uFJsk7zpqcTTT8tqBw
> Accept: application/json
>

ref:1ca63ff88eca28e8dbf8b13061a2eac11fa5e3cb

2 files changed, 7 insertions, 6 deletions

diff --git a/ydb/core/persqueue/metering_sink.cpp b/ydb/core/persqueue/metering_sink.cpp
index 942650760d..0726ff94c7 100644
--- a/ydb/core/persqueue/metering_sink.cpp
+++ b/ydb/core/persqueue/metering_sink.cpp
@@ -151,7 +151,7 @@ void TMeteringSink::Flush(TInstant now, bool force) {
                 {"shard_enhanced_consumers_throughput", Parameters_.ConsumersThroughput},
                 {"reserved_storage_bytes", Parameters_.ReservedSpace}
             };
-            auto interval = TInstant::Hours(LastFlush_[whichOne].Hours() + 1);
+            auto interval = TInstant::Hours(LastFlush_[whichOne].Hours()) + Parameters_.FlushLimit;
             while (interval < now) {
                 const auto metricsJson = GetMeteringJson(
                     name, schema, tags, "second",
@@ -159,7 +159,7 @@ void TMeteringSink::Flush(TInstant now, bool force) {
                     LastFlush_[whichOne], interval, now);
                 LastFlush_[whichOne] = interval;
                 FlushFunction_(metricsJson);
-                interval += TDuration::Hours(1);
+                interval += Parameters_.FlushLimit;
             }
             if (LastFlush_[whichOne] < now) {
                 const auto metricsJson = GetMeteringJson(
@@ -182,7 +182,7 @@ void TMeteringSink::Flush(TInstant now, bool force) {
             const THashMap<TString, ui64> tags = {
                 {"reserved_throughput_bps", Parameters_.WriteQuota},
             };
-            auto interval = TInstant::Hours(LastFlush_[whichOne].Hours() + 1);
+            auto interval = TInstant::Hours(LastFlush_[whichOne].Hours()) + Parameters_.FlushLimit;
             while (interval < now) {
                 const auto metricsJson = GetMeteringJson(
                     name, schema, tags, "second",
@@ -190,7 +190,7 @@ void TMeteringSink::Flush(TInstant now, bool force) {
                     LastFlush_[whichOne], interval, now);
                 LastFlush_[whichOne] = interval;
                 FlushFunction_(metricsJson);
-                interval += TDuration::Hours(1);
+                interval += Parameters_.FlushLimit;
             }
             if (LastFlush_[whichOne] < now) {
                 const auto metricsJson = GetMeteringJson(
@@ -210,7 +210,7 @@ void TMeteringSink::Flush(TInstant now, bool force) {
             }
             const TString name = "yds.reserved_resources";
             const TString schema = "yds.storage.reserved.v1";
-            auto interval = TInstant::Hours(LastFlush_[whichOne].Hours() + 1);
+            auto interval = TInstant::Hours(LastFlush_[whichOne].Hours()) + Parameters_.FlushLimit;
             while (interval < now) {
                 const auto metricsJson = GetMeteringJson(
                     name, schema, {}, "mbyte*second",
@@ -219,7 +219,7 @@ void TMeteringSink::Flush(TInstant now, bool force) {
                     LastFlush_[whichOne], interval, now);
                 LastFlush_[whichOne] = interval;
                 FlushFunction_(metricsJson);
-                interval += TDuration::Hours(1);
+                interval += Parameters_.FlushLimit;
             }
             if (LastFlush_[whichOne] < now) {
                 const auto metricsJson = GetMeteringJson(
diff --git a/ydb/core/persqueue/metering_sink.h b/ydb/core/persqueue/metering_sink.h
index 6fe0154811..d836a50e8e 100644
--- a/ydb/core/persqueue/metering_sink.h
+++ b/ydb/core/persqueue/metering_sink.h
@@ -15,6 +15,7 @@ enum class EMeteringJson {
 class TMeteringSink {
 public:
     struct TParameters {
+        TDuration FlushLimit{TDuration::Hours(1)};
         TDuration FlushInterval;
         TString TabletId;
         TString YcCloudId;