support consumer's availability_period in the ydb_cli (#26351)

KIKIMR-24054

Author: ubyte

ydb/apps/ydb/CHANGELOG.md

@@ -1,4 +1,6 @@
 * Fixed a bug where the `ydb debug ping` command crashed in case of any error.
+* Added a new `--retention-period` option to the `ydb topic` subcommands. The new option supports various time units, such as seconds, minutes, or days. Usage of the legacy `--retention-period-hours` option is discouraged.
+* The `ydb topic consumer add` subcommand now has a new `--availability-period` option, which overrides the consumer's retention guarantee.
 
 ## 2.26.0 ##
 

ydb/public/lib/ydb_cli/commands/ydb_service_scheme.cpp

@@ -9,6 +9,7 @@
 
 #include <google/protobuf/port_def.inc>
 
+#include <util/stream/format.h>
 #include <util/string/join.h>
 
 namespace NYdb {
@@ -121,6 +122,39 @@ namespace {
             Cout << "none" << Endl;
         }
     }
+
+    struct TPrettyDurationFormatParameters {
+        double EntitiesPerSecond;
+        TStringBuf EntityName;
+        TStringBuf ZeroString;
+        TStringBuf MaxString;
+        int MaxPrecision;
+    };
+
+    constexpr TPrettyDurationFormatParameters PRETTY_HOURS_DEFAULT{
+        .EntitiesPerSecond = 3600,
+        .EntityName = "hours",
+        .ZeroString = "0 hours",
+        .MaxString = "+infinity",
+        .MaxPrecision = 3,
+    };
+
+    constexpr TPrettyDurationFormatParameters PRETTY_HOURS_NON_ZERO{
+        .EntitiesPerSecond = 3600,
+        .EntityName = "hours",
+        .ZeroString = "",
+        .MaxString = "+infinity",
+        .MaxPrecision = 3,
+    };
+
+    TString PrettyDurationString(TDuration duration, const TPrettyDurationFormatParameters& paramerters) {
+        if (duration == TDuration::Zero()) {
+            return ToString(paramerters.ZeroString);
+        } else if (duration == TDuration::Max()) {
+            return ToString(paramerters.MaxString);
+        }
+        return TStringBuilder() << Prec(duration.MillisecondsFloat() / (paramerters.EntitiesPerSecond * 1000.0), PREC_POINT_DIGITS_STRIP_ZEROES, paramerters.MaxPrecision) << " " << paramerters.EntityName;
+    }
 }
 
 void PrintAllPermissions(
@@ -139,6 +173,9 @@ int PrintPrettyDescribeConsumerResult(const NYdb::NTopic::TConsumerDescription&
     const NYdb::NTopic::TConsumer& consumer = description.GetConsumer();
     Cout << "Consumer " << consumer.GetConsumerName() << ": " << Endl;
     Cout << "Important: " << (consumer.GetImportant() ? "Yes" : "No") << Endl;
+    if (const auto availabilityPeriodStr = PrettyDurationString(consumer.GetAvailabilityPeriod(), PRETTY_HOURS_NON_ZERO)) {
+        Cout << "Availability period: " << availabilityPeriodStr << Endl;
+    }
     if (const TInstant& readFrom = consumer.GetReadFrom()) {
         Cout << "Read from: " << readFrom.ToRfc822StringLocal() << Endl;
     } else {
@@ -315,13 +352,14 @@ namespace {
         if (consumers.empty()) {
             return;
         }
-        TPrettyTable table({ "ConsumerName", "SupportedCodecs", "ReadFrom", "Important" });
+        TPrettyTable table({ "ConsumerName", "SupportedCodecs", "ReadFrom", "Important", "Availability period", });
         for (const auto& c: consumers) {
             table.AddRow()
                 .Column(0, c.GetConsumerName())
                 .Column(1, FormatCodecs(c.GetSupportedCodecs()))
                 .Column(2, c.GetReadFrom().ToRfc822StringLocal())
-                .Column(3, c.GetImportant() ? "Yes" : "No");
+                .Column(3, c.GetImportant() ? "Yes" : "No")
+                .Column(4, PrettyDurationString(c.GetAvailabilityPeriod(), PRETTY_HOURS_NON_ZERO));
 //                .Column(4, rule.ServiceType())
 //                .Column(5, rule.Version());
         }
@@ -345,7 +383,7 @@ namespace {
 
     void PrintMain(const NTopic::TTopicDescription& topicDescription) {
         Cout << Endl << "Main:";
-        Cout << Endl << "RetentionPeriod: " << topicDescription.GetRetentionPeriod().Hours() << " hours";
+        Cout << Endl << "RetentionPeriod: " << PrettyDurationString(topicDescription.GetRetentionPeriod(), PRETTY_HOURS_DEFAULT);
         if (topicDescription.GetRetentionStorageMb().has_value()) {
             Cout << Endl << "StorageRetention: " << *topicDescription.GetRetentionStorageMb() << " MB";
         }

ydb/public/lib/ydb_cli/commands/ydb_service_topic.cpp

@@ -11,11 +11,13 @@
 #include <ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/proto/accessor.h>
 
 #include <util/generic/set.h>
+#include <util/stream/format.h>
 #include <util/stream/str.h>
 #include <util/string/cast.h>
 #include <util/string/hex.h>
 #include <util/string/vector.h>
 #include <util/string/join.h>
+#include <util/string/strip.h>
 
 #define TIMESTAMP_FORMAT_OPTION_DESCRIPTION "Timestamp may be specified in unix time format (seconds from 1970.01.01) or in ISO-8601 format (like 2020-07-10T15:00:00Z)"
 
@@ -162,6 +164,28 @@ namespace NYdb::NConsoleClient {
 
             return exists->second;
         }
+
+        TDuration ParseDurationHours(const TStringBuf str) {
+            double hours = 0;
+            if (!TryFromString(str, hours)) {
+                throw TMisuseException() << "Invalid hours duration '" << str << "'";
+            }
+            if (hours < 0) {
+                throw TMisuseException() << "Duration must be non-negative";
+            }
+            if (!std::isfinite(hours)) {
+                throw TMisuseException() << "Duration must be finite";
+            }
+            return TDuration::Seconds(hours * 3600); // using floating-point ctor with saturation
+        }
+
+        TDuration ParseDuration(TStringBuf str) {
+            StripInPlace(str);
+            if (!str.empty() && !IsAsciiAlpha(str.back())) {
+                throw TMisuseException() << "Duration must ends with a unit name (ex. 'h' for hours, 's' for seconds)";
+            }
+            return TDuration::Parse(str);
+        }
     }
 
     void TCommandWithSupportedCodecs::AddAllowedCodecs(TClientCommand::TConfig& config, const TVector<NYdb::NTopic::ECodec>& supportedCodecs) {
@@ -343,10 +367,18 @@ namespace NYdb::NConsoleClient {
             .Optional()
             .StoreResult(&MinActivePartitions_)
             .DefaultValue(1);
-        config.Opts->AddLongOption("retention-period-hours", "Duration in hours for which data in topic is stored")
-            .DefaultValue(24)
+        config.Opts->AddLongOption("retention-period-hours", TStringBuilder()
+                << "Duration in hours for which data in topic is stored "
+                << "(default: " << NColorizer::StdOut().CyanColor() << RetentionPeriod_.Hours() << NColorizer::StdOut().OldColor() << ")")
+            .Hidden()
+            .Optional()
+            .RequiredArgument("HOURS")
+            .StoreMappedResult(&RetentionPeriod_, ParseDurationHours);
+        config.Opts->AddLongOption("retention-period", TStringBuilder()
+                << "Duration for which data in topic is stored (ex. '72h', '1440m') "
+                << "(default: " << NColorizer::StdOut().CyanColor() << HumanReadable(RetentionPeriod_) << NColorizer::StdOut().OldColor() << ")")
             .Optional()
-            .StoreResult(&RetentionPeriodHours_);
+            .StoreMappedResult(&RetentionPeriod_, ParseDuration);
         config.Opts->AddLongOption("partition-write-speed-kbps", "Partition write speed in kilobytes per second")
             .DefaultValue(1024)
             .Optional()
@@ -370,6 +402,8 @@ namespace NYdb::NConsoleClient {
             .Optional()
             .Hidden()
             .StoreResult(&PartitionsPerTablet_);
+
+        config.Opts->MutuallyExclusive("retention-period-hours", "retention-period");
     }
 
     void TCommandTopicCreate::Parse(TConfig& config) {
@@ -409,7 +443,7 @@ namespace NYdb::NConsoleClient {
             settings.MeteringMode(GetMeteringMode());
         }
 
-        settings.RetentionPeriod(TDuration::Hours(RetentionPeriodHours_));
+        settings.RetentionPeriod(RetentionPeriod_);
         settings.RetentionStorageMb(RetentionStorageMb_);
 
         if (PartitionsPerTablet_.Defined()) {
@@ -434,9 +468,14 @@ namespace NYdb::NConsoleClient {
         config.Opts->AddLongOption("partitions-count", "Initial and minimum number of partitions for topic")
             .Optional()
             .StoreResult(&MinActivePartitions_);
-        config.Opts->AddLongOption("retention-period-hours", "Duration for which data in topic is stored")
+        config.Opts->AddLongOption("retention-period-hours", "Duration in hours for which data in topic is stored")
+            .Hidden()
             .Optional()
-            .StoreResult(&RetentionPeriodHours_);
+            .RequiredArgument("HOURS")
+            .StoreMappedResult(&RetentionPeriod_, ParseDurationHours);
+        config.Opts->AddLongOption("retention-period", "Duration for which data in topic is stored (ex. '72h', '1440m')")
+            .Optional()
+            .StoreMappedResult(&RetentionPeriod_, ParseDuration);
         config.Opts->AddLongOption("partition-write-speed-kbps", "Partition write speed in kilobytes per second")
             .Optional()
             .StoreResult(&PartitionWriteSpeedKbps_);
@@ -453,6 +492,8 @@ namespace NYdb::NConsoleClient {
             .Optional()
             .StoreResult(&MaxActivePartitions_);
         AddAutoPartitioning(config, true);
+
+        config.Opts->MutuallyExclusive("retention-period-hours", "retention-period");
     }
 
     void TCommandTopicAlter::Parse(TConfig& config) {
@@ -501,8 +542,8 @@ namespace NYdb::NConsoleClient {
             settings.SetSupportedCodecs(codecs);
         }
 
-        if (RetentionPeriodHours_.Defined() && describeResult.GetTopicDescription().GetRetentionPeriod() != TDuration::Hours(*RetentionPeriodHours_)) {
-            settings.SetRetentionPeriod(TDuration::Hours(*RetentionPeriodHours_));
+        if (RetentionPeriod_.Defined() && describeResult.GetTopicDescription().GetRetentionPeriod() != RetentionPeriod_) {
+            settings.SetRetentionPeriod(*RetentionPeriod_);
         }
 
         if (PartitionWriteSpeedKbps_.Defined() && describeResult.GetTopicDescription().GetPartitionWriteSpeedBytesPerSecond() / 1_KB != *PartitionWriteSpeedKbps_) {
@@ -600,6 +641,9 @@ namespace NYdb::NConsoleClient {
             .Optional()
             .DefaultValue(false)
             .StoreResult(&IsImportant_);
+        config.Opts->AddLongOption("availability-period", "Duration for which uncommited data in topic is retained (ex. '72h', '1440m')")
+            .Optional()
+            .StoreMappedResult(&AvailabilityPeriod_, ParseDuration);
         config.Opts->SetFreeArgsNum(1);
         SetFreeArgTitle(0, "<topic-path>", "Topic path");
         AddAllowedCodecs(config, AllowedCodecs);
@@ -630,6 +674,9 @@ namespace NYdb::NConsoleClient {
             consumerSettings.SetSupportedCodecs(codecs);
         }
         consumerSettings.SetImportant(IsImportant_);
+        if (AvailabilityPeriod_.Defined()) {
+            consumerSettings.AvailabilityPeriod(*AvailabilityPeriod_);
+        }
 
         readRuleSettings.AppendAddConsumers(consumerSettings);
 

ydb/public/lib/ydb_cli/commands/ydb_service_topic.h

@@ -83,7 +83,7 @@ namespace NYdb::NConsoleClient {
         int Run(TConfig& config) override;
 
     private:
-        ui64 RetentionPeriodHours_;
+        TDuration RetentionPeriod_ = TDuration::Hours(24);
         ui64 RetentionStorageMb_;
         ui32 MinActivePartitions_;
         TMaybe<ui32> MaxActivePartitions_;
@@ -105,7 +105,7 @@ namespace NYdb::NConsoleClient {
         int Run(TConfig& config) override;
 
     private:
-        TMaybe<ui64> RetentionPeriodHours_;
+        TMaybe<TDuration> RetentionPeriod_;
         TMaybe<ui64> RetentionStorageMb_;
         TMaybe<ui32> MinActivePartitions_;
         TMaybe<ui32> MaxActivePartitions_;
@@ -145,6 +145,7 @@ namespace NYdb::NConsoleClient {
     private:
         TString ConsumerName_;
         bool IsImportant_;
+        TMaybe<TDuration> AvailabilityPeriod_;
         TMaybe<TInstant> StartingMessageTimestamp_;
     };
 

ydb/public/lib/ydb_cli/dump/restore_impl.cpp

@@ -2160,6 +2160,7 @@ TRestoreResult TRestoreClient::RestoreConsumers(const TString& topicPath, const
                 .BeginAddConsumer()
                     .ConsumerName(consumer.GetConsumerName())
                     .Important(consumer.GetImportant())
+                    .AvailabilityPeriod(consumer.GetAvailabilityPeriod())
                     .Attributes(consumer.GetAttributes())
                 .EndAddConsumer()
         ).GetValueSync();