Topic APIs for metrics level (#25512)

Author: qyryq

.github/last_commit.txt

@@ -1 +1 @@
-0bf38aa928934c9db08c6a173cbe6e80954d2f8f
+337ad331d577c2bf3d4e92fb73a77dc37cf8fc7f

include/ydb-cpp-sdk/client/topic/control_plane.h

@@ -36,6 +36,13 @@ enum class EAutoPartitioningStrategy: uint32_t {
     Paused = 4,
 };
 
+// 0 - unspecified
+// 1 - disabeld
+// 2 - database level metrics
+// 3 - object level metrics
+// 4 - detailed metrics
+using EMetricsLevel = uint32_t;
+
 class TConsumer {
 public:
     TConsumer(const Ydb::Topic::Consumer&);
@@ -309,6 +316,8 @@ class TTopicDescription {
 
     const TTopicStats& GetTopicStats() const;
 
+    std::optional<EMetricsLevel> GetMetricsLevel() const;
+
     void SerializeTo(Ydb::Topic::CreateTopicRequest& request) const;
 private:
 
@@ -332,6 +341,7 @@ class TTopicDescription {
     NScheme::TVirtualTimestamp CreationTimestamp_;
     std::vector<NScheme::TPermissions> Permissions_;
     std::vector<NScheme::TPermissions> EffectivePermissions_;
+    std::optional<EMetricsLevel> MetricsLevel_;
 };
 
 class TConsumerDescription {
@@ -571,6 +581,8 @@ struct TCreateTopicSettings : public TOperationRequestSettings<TCreateTopicSetti
 
     FLUENT_SETTING(TAttributes, Attributes);
 
+    FLUENT_SETTING_OPTIONAL(EMetricsLevel, MetricsLevel);
+
     TCreateTopicSettings& SetSupportedCodecs(std::vector<ECodec>&& codecs) {
         SupportedCodecs_ = std::move(codecs);
         return *this;
@@ -743,7 +755,20 @@ struct TAlterTopicSettings : public TOperationRequestSettings<TAlterTopicSetting
         return *this;
     }
 
+    TAlterTopicSettings& SetMetricsLevel(EMetricsLevel level) {
+        MetricsLevel_ = level;
+        return *this;
+    }
+    TAlterTopicSettings& ResetMetricsLevel() {
+        MetricsLevel_ = true;
+        return *this;
+    }
+
     std::optional<TAlterPartitioningSettings> AlterPartitioningSettings_;
+    std::variant<
+        bool,         // Reset
+        EMetricsLevel // Set
+    > MetricsLevel_ = false;
 };
 
 inline TPartitioningSettingsBuilder TCreateTopicSettings::BeginConfigurePartitioningSettings() {

src/api/protos/ydb_common.proto

@@ -27,9 +27,3 @@ message VirtualTimestamp {
     uint64 plan_step = 1;
     uint64 tx_id = 2;
 }
-
-enum MetricsLevel {
-    Database = 0;
-    Object = 1;
-    Detailed = 2;
-}

src/api/protos/ydb_persqueue_v1.proto

@@ -1187,6 +1187,9 @@ message TopicSettings {
     }
     // remote mirror rule for this topic.
     RemoteMirrorRule remote_mirror_rule = 11;
+
+    // Set or reset metrics level.
+    optional uint32 metrics_level = 16;
 }
 
 message AutoPartitioningSettings {

src/api/protos/ydb_topic.proto

@@ -6,8 +6,8 @@ import "src/api/protos/ydb_issue_message.proto";
 import "src/api/protos/annotations/sensitive.proto";
 import "src/api/protos/annotations/validation.proto";
 
-import "google/protobuf/empty.proto";
 import "google/protobuf/duration.proto";
+import "google/protobuf/empty.proto";
 import "google/protobuf/timestamp.proto";
 
 package Ydb.Topic;
@@ -156,7 +156,7 @@ message StreamWriteMessage {
             // Message sequence number, provided by client for deduplication.
             // Starts at 1
             int64 seq_no = 1;
-            // Creation timestamp
+            // Creation timestamp.
             google.protobuf.Timestamp created_at = 2;
             // Compressed client message body.
             bytes data = 3;
@@ -1002,6 +1002,9 @@ message CreateTopicRequest {
 
     // Metering mode for the topic in a serverless database.
     MeteringMode metering_mode = 12;
+
+    // Metrics level. If the level is unset, use database setting.
+    optional uint32 metrics_level = 13;
 }
 
 // Create topic response sent from server to client.
@@ -1135,6 +1138,9 @@ message DescribeTopicResult {
         // How much bytes were written statistics.
         MultipleWindowsStat bytes_written = 4;
     }
+
+    // Metrics level.
+    optional uint32 metrics_level = 16;
 }
 
 // Describe partition request sent from client to server.
@@ -1262,7 +1268,6 @@ message PartitionStats {
     int32 partition_node_id = 8 [deprecated=true]; //Use PartitionLocation
 }
 
-
 // Update existing topic request sent from client to server.
 message AlterTopicRequest {
     Ydb.Operations.OperationParams operation_params = 1;
@@ -1307,6 +1312,12 @@ message AlterTopicRequest {
 
     // Set metering mode for topic in serverless database.
     MeteringMode set_metering_mode = 14;
+
+    // Set or reset metrics level.
+    oneof metrics_level {
+        uint32 set_metrics_level = 15;
+        google.protobuf.Empty reset_metrics_level = 16;
+    }
 }
 
 // Update topic response sent from server to client.

src/client/topic/impl/topic.cpp

@@ -54,6 +54,7 @@ TTopicDescription::TTopicDescription(Ydb::Topic::DescribeTopicResult&& result)
     , PartitionWriteBurstBytes_(Proto_.partition_write_burst_bytes())
     , MeteringMode_(TProtoAccessor::FromProto(Proto_.metering_mode()))
     , TopicStats_(Proto_.topic_stats())
+    , MetricsLevel_(Proto_.has_metrics_level() ? std::optional(static_cast<EMetricsLevel>(Proto_.metrics_level())) : std::optional<EMetricsLevel>())
 {
     Owner_ = Proto_.self().owner();
     CreationTimestamp_ = NScheme::TVirtualTimestamp(Proto_.self().created_at());
@@ -193,6 +194,9 @@ void TTopicDescription::SerializeTo(Ydb::Topic::CreateTopicRequest& request) con
     *request.mutable_attributes() = Proto_.attributes();
     *request.mutable_consumers() = Proto_.consumers();
     request.set_metering_mode(Proto_.metering_mode());
+    if (Proto_.has_metrics_level()) {
+        request.set_metrics_level(Proto_.metrics_level());
+    }
 }
 
 const Ydb::Topic::DescribeTopicResult& TTopicDescription::GetProto() const {
@@ -219,6 +223,10 @@ const TTopicStats& TTopicDescription::GetTopicStats() const {
     return TopicStats_;
 }
 
+std::optional<EMetricsLevel> TTopicDescription::GetMetricsLevel() const {
+    return MetricsLevel_;
+}
+
 const std::vector<NScheme::TPermissions>& TTopicDescription::GetPermissions() const {
     return Permissions_;
 }
@@ -664,6 +672,7 @@ TCreateTopicSettings::TCreateTopicSettings(const Ydb::Topic::CreateTopicRequest&
     , PartitionWriteSpeedBytesPerSecond_(proto.partition_write_speed_bytes_per_second())
     , PartitionWriteBurstBytes_(proto.partition_write_burst_bytes())
     , Attributes_(DeserializeAttributes(proto.attributes()))
+    , MetricsLevel_(proto.has_metrics_level() ? std::optional(static_cast<EMetricsLevel>(proto.metrics_level())) : std::nullopt)
 {
     Consumers_ = DeserializeConsumers(*this, proto.consumers());
 }
@@ -678,6 +687,9 @@ void TCreateTopicSettings::SerializeTo(Ydb::Topic::CreateTopicRequest& request)
     request.set_partition_write_burst_bytes(PartitionWriteBurstBytes_);
     *request.mutable_consumers() = SerializeConsumers(Consumers_);
     *request.mutable_attributes() = SerializeAttributes(Attributes_);
+    if (MetricsLevel_) {
+        request.set_metrics_level(*MetricsLevel_);
+    }
 }
 
 } // namespace NYdb::NTopic

src/client/topic/impl/topic_impl.h

@@ -150,6 +150,12 @@ class TTopicClient::TImpl : public TClientImplCommon<TTopicClient::TImpl> {
             ConvertAlterConsumerToProto(consumer, consumerProto);
         }
 
+        if (auto level = std::get_if<EMetricsLevel>(&settings.MetricsLevel_)) {
+            request.set_set_metrics_level(*level);
+        } else if (auto reset = std::get_if<bool>(&settings.MetricsLevel_); *reset) {
+            request.mutable_reset_metrics_level();
+        }
+
         return request;
     }
 

tests/integration/topic/describe_topic.cpp

@@ -119,4 +119,78 @@ TEST_F(Describe, TEST_NAME(Location)) {
     }
 }
 
+TEST_F(Describe, TEST_NAME(MetricsLevel)) {
+    TTopicClient client(MakeDriver());
+
+    // const std::uint32_t MetricsLevelDisabled = 0;
+    // const std::uint32_t MetricsLevelDatabase = 1;
+    const std::uint32_t MetricsLevelObject = 2;
+    const std::uint32_t MetricsLevelDetailed = 3;
+
+    auto createTopic = [&](std::string topic, EMetricsLevel metricsLevel) {
+        auto res = client.CreateTopic(topic, TCreateTopicSettings().MetricsLevel(metricsLevel)).GetValueSync();
+        ASSERT_TRUE(res.IsSuccess());
+    };
+
+    auto setMetricsLevel = [&](std::string topic, EMetricsLevel metricsLevel) {
+        auto res = client.AlterTopic(topic, TAlterTopicSettings().SetMetricsLevel(metricsLevel)).GetValueSync();
+        ASSERT_TRUE(res.IsSuccess());
+    };
+
+    auto resetMetricsLevel = [&](std::string topic) {
+        auto res = client.AlterTopic(topic, TAlterTopicSettings().ResetMetricsLevel()).GetValueSync();
+        ASSERT_TRUE(res.IsSuccess());
+    };
+
+    auto checkFlag = [&](std::string topic, std::optional<EMetricsLevel> expectedMetricsLevel) {
+        auto res = client.DescribeTopic(topic, {}).GetValueSync();
+        Y_ENSURE(res.IsSuccess());
+        return res.GetTopicDescription().GetMetricsLevel() == expectedMetricsLevel;
+    };
+
+    {
+        const std::string topic("topic-with-counters");
+        createTopic(topic, MetricsLevelDetailed);
+        checkFlag(topic, MetricsLevelDetailed);
+        setMetricsLevel(topic, MetricsLevelObject);
+        Y_ENSURE(checkFlag(topic, MetricsLevelObject));
+
+        {
+            // Empty alter should change nothing.
+            auto res = client.AlterTopic(topic).GetValueSync();
+            ASSERT_TRUE(res.IsSuccess());
+            Y_ENSURE(checkFlag(topic, MetricsLevelObject));
+        }
+
+        {
+            resetMetricsLevel(topic);
+            Y_ENSURE(checkFlag(topic, {}));
+        }
+    }
+
+    {
+        const std::string topic("topic-without-counters-by-default");
+        auto res = client.CreateTopic(topic).GetValueSync();
+        ASSERT_TRUE(res.IsSuccess());
+        Y_ENSURE(checkFlag(topic, {}));
+        setMetricsLevel(topic, MetricsLevelDetailed);
+        Y_ENSURE(checkFlag(topic, MetricsLevelDetailed));
+
+        {
+            // Empty alter should change nothing.
+            auto res = client.AlterTopic(topic).GetValueSync();
+            ASSERT_TRUE(res.IsSuccess());
+            Y_ENSURE(checkFlag(topic, MetricsLevelDetailed));
+        }
+    }
+
+    {
+        const std::string topic("topic-without-counters");
+        createTopic(topic, MetricsLevelObject);
+        Y_ENSURE(checkFlag(topic, MetricsLevelObject));
+        setMetricsLevel(topic, MetricsLevelDetailed);
+        Y_ENSURE(checkFlag(topic, MetricsLevelDetailed));
+    }
+}
+
 }

tests/integration/topic/utils/describe.cpp

@@ -42,6 +42,8 @@ void DescribeTopicTest(ITopicTestSetup& setup, TTopicClient& client, bool requir
             Y_ENSURE(partitionLocation.GetNodeId() > 0);
             Y_ENSURE(partitionLocation.GetGeneration() >= 0); // greater-or-equal 0
         }
+
+        Y_ENSURE(!description.GetMetricsLevel().has_value());
     }
 }