Skip to content

Commit d717e30

Browse files
niksavelievniksaveliev
authored
Merge 2659681 into 0a916d0 
2 parents 0a916d0 + 2659681 commit d717e30

File tree

2 files changed

+151
-11
lines changed

2 files changed

+151
-11
lines changed

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

+125-7
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,18 @@ namespace NYdb::NConsoleClient {
3838
std::pair<NTopic::EMeteringMode, TString>(NTopic::EMeteringMode::RequestUnits, "Read/write operations valued in request units, storage usage on hourly basis."),
3939
};
4040

41+
THashMap<TString, NTopic::EAutoscalingStrategy> AutoscaleStrategies = {
42+
std::pair<TString, NTopic::EAutoscalingStrategy>("disabled", NTopic::EAutoscalingStrategy::Disabled),
43+
std::pair<TString, NTopic::EAutoscalingStrategy>("up", NTopic::EAutoscalingStrategy::ScaleUp),
44+
std::pair<TString, NTopic::EAutoscalingStrategy>("up-and-down", NTopic::EAutoscalingStrategy::ScaleUpAndDown),
45+
};
46+
47+
THashMap<NTopic::EAutoscalingStrategy, TString> AutoscaleStrategiesDescriptions = {
48+
std::pair<NTopic::EAutoscalingStrategy, TString>(NTopic::EAutoscalingStrategy::Disabled, "Automatic scaling of the number of partitions is disabled"),
49+
std::pair<NTopic::EAutoscalingStrategy, TString>(NTopic::EAutoscalingStrategy::ScaleUp, "The number of partitions can increase under high load, but cannot decrease"),
50+
std::pair<NTopic::EAutoscalingStrategy, TString>(NTopic::EAutoscalingStrategy::ScaleUpAndDown, "The number of partitions can increase under high load and decrease under low load"),
51+
};
52+
4153
THashMap<ETopicMetadataField, TString> TopicMetadataFieldsDescriptions = {
4254
{ETopicMetadataField::Body, "Message data"},
4355
{ETopicMetadataField::WriteTime, "Message write time, a UNIX timestamp the message was written to server."},
@@ -172,6 +184,72 @@ namespace {
172184
return MeteringMode_;
173185
}
174186

187+
void TCommandWithAutoscaling::AddAutoscaling(TClientCommand::TConfig& config, bool withDefault) {
188+
TStringStream description;
189+
description << "A strategy to automatically change the number of partitions depending on the load. Available strategies: ";
190+
NColorizer::TColors colors = NColorizer::AutoColors(Cout);
191+
for (const auto& strategy: AutoscaleStrategies) {
192+
auto findResult = AutoscaleStrategiesDescriptions.find(strategy.second);
193+
Y_ABORT_UNLESS(findResult != AutoscaleStrategiesDescriptions.end(),
194+
"Couldn't find description for %s autoscale strategy", (TStringBuilder() << strategy.second).c_str());
195+
description << "\n " << colors.BoldColor() << strategy.first << colors.OldColor()
196+
<< "\n " << findResult->second;
197+
if (strategy.second == NTopic::EAutoscalingStrategy::Disabled) {
198+
description << colors.CyanColor() << " (default)" << colors.OldColor();
199+
}
200+
}
201+
202+
auto straregy = config.Opts->AddLongOption("autoscale-strategy", description.Str())
203+
.Optional()
204+
.StoreResult(&AutoscaleStrategy_);
205+
auto thresholdTime = config.Opts->AddLongOption("autoscale-threshold-time", "Duration in seconds of high or low load before automatically scale the number of partitions")
206+
.Optional()
207+
.StoreResult(&ScaleThresholdTime_);
208+
auto upThresholdPercent = config.Opts->AddLongOption("autoscale-scale-up-threshold-percent", "The load percentage at which the number of partitions will increase.")
209+
.Optional()
210+
.StoreResult(&ScaleUpThresholdPercent_);
211+
auto downThresholdPercent = config.Opts->AddLongOption("autoscale-scale-down-threshold-percent", "The load percentage at which the number of partitions will decrease.")
212+
.Optional()
213+
.StoreResult(&ScaleDownThresholdPercent_);
214+
215+
if (withDefault) {
216+
straregy.DefaultValue("disabled");
217+
thresholdTime.DefaultValue(300);
218+
upThresholdPercent.DefaultValue(90);
219+
downThresholdPercent.DefaultValue(30);
220+
}
221+
}
222+
223+
void TCommandWithAutoscaling::ParseAutoscalingStrategy() {
224+
if (AutoscalingStrategyStr_.empty()) {
225+
return;
226+
}
227+
228+
TString toLowerStrategy = to_lower(AutoscalingStrategyStr_);
229+
auto strategyIt = AutoscaleStrategies.find(toLowerStrategy);
230+
if (strategyIt.IsEnd()) {
231+
throw TMisuseException() << "Autoscaling strategy " << AutoscalingStrategyStr_ << " is not available for this command";
232+
} else {
233+
AutoscaleStrategy_ = strategyIt->second;
234+
}
235+
}
236+
237+
TMaybe<NTopic::EAutoscalingStrategy> TCommandWithAutoscaling::GetAutoscalingStrategy() const {
238+
return AutoscaleStrategy_;
239+
}
240+
241+
TMaybe<ui32> TCommandWithAutoscaling::GetScaleThresholdTime() const {
242+
return ScaleThresholdTime_;
243+
}
244+
245+
TMaybe<ui32> TCommandWithAutoscaling::GetScaleUpThresholdPercent() const {
246+
return ScaleUpThresholdPercent_;
247+
}
248+
249+
TMaybe<ui32> TCommandWithAutoscaling::GetScaleDownThresholdPercent() const {
250+
return ScaleDownThresholdPercent_;
251+
}
252+
175253
TCommandTopic::TCommandTopic()
176254
: TClientCommandTree("topic", {}, "TopicService operations") {
177255
AddCommand(std::make_unique<TCommandTopicCreate>());
@@ -188,9 +266,13 @@ namespace {
188266

189267
void TCommandTopicCreate::Config(TConfig& config) {
190268
TYdbCommand::Config(config);
191-
config.Opts->AddLongOption("partitions-count", "Total partitions count for topic")
269+
config.Opts->AddLongOption("partitions-count", "Initial number of partitions for topic")
270+
.DefaultValue(1)
271+
.StoreResult(&MinActivePartitions_);
272+
config.Opts->AddLongOption("max-partitions-count", "Maximum number of partitions for topic")
192273
.DefaultValue(1)
193-
.StoreResult(&PartitionsCount_);
274+
.Optional()
275+
.StoreResult(&MaxActivePartitions_);
194276
config.Opts->AddLongOption("retention-period-hours", "Duration in hours for which data in topic is stored")
195277
.DefaultValue(24)
196278
.Optional()
@@ -207,21 +289,30 @@ namespace {
207289
SetFreeArgTitle(0, "<topic-path>", "Topic path");
208290
AddAllowedCodecs(config, AllowedCodecs);
209291
AddAllowedMeteringModes(config);
292+
AddAutoscaling(config, true);
210293
}
211294

212295
void TCommandTopicCreate::Parse(TConfig& config) {
213296
TYdbCommand::Parse(config);
214297
ParseTopicName(config, 0);
215298
ParseCodecs();
216299
ParseMeteringMode();
300+
ParseAutoscalingStrategy();
217301
}
218302

219303
int TCommandTopicCreate::Run(TConfig& config) {
220304
TDriver driver = CreateDriver(config);
221305
NYdb::NTopic::TTopicClient topicClient(driver);
222306

223307
auto settings = NYdb::NTopic::TCreateTopicSettings();
224-
settings.PartitioningSettings(PartitionsCount_, PartitionsCount_);
308+
309+
auto autoscaleSettings = NTopic::TAutoscalingSettings(
310+
GetAutoscalingStrategy() ? *GetAutoscalingStrategy() : NTopic::EAutoscalingStrategy::Disabled,
311+
GetScaleThresholdTime() ? TDuration::Seconds(*GetScaleThresholdTime()) : TDuration::Seconds(0),
312+
GetScaleUpThresholdPercent() ? *GetScaleUpThresholdPercent() : 0,
313+
GetScaleDownThresholdPercent() ? *GetScaleDownThresholdPercent() : 0);
314+
315+
settings.PartitioningSettings(MinActivePartitions_, MaxActivePartitions_, autoscaleSettings);
225316
settings.PartitionWriteBurstBytes(PartitionWriteSpeedKbps_ * 1_KB);
226317
settings.PartitionWriteSpeedBytesPerSecond(PartitionWriteSpeedKbps_ * 1_KB);
227318

@@ -249,8 +340,11 @@ namespace {
249340

250341
void TCommandTopicAlter::Config(TConfig& config) {
251342
TYdbCommand::Config(config);
252-
config.Opts->AddLongOption("partitions-count", "Total partitions count for topic")
253-
.StoreResult(&PartitionsCount_);
343+
config.Opts->AddLongOption("partitions-count", "Initial number of partitions for topic")
344+
.StoreResult(&MinActivePartitions_);
345+
config.Opts->AddLongOption("max-partitions-count", "Maximum number of partitions for topic")
346+
.Optional()
347+
.StoreResult(&MaxActivePartitions_);
254348
config.Opts->AddLongOption("retention-period-hours", "Duration for which data in topic is stored")
255349
.Optional()
256350
.StoreResult(&RetentionPeriodHours_);
@@ -264,6 +358,7 @@ namespace {
264358
SetFreeArgTitle(0, "<topic-path>", "Topic path");
265359
AddAllowedCodecs(config, AllowedCodecs);
266360
AddAllowedMeteringModes(config);
361+
AddAutoscaling(config, false);
267362
}
268363

269364
void TCommandTopicAlter::Parse(TConfig& config) {
@@ -276,9 +371,32 @@ namespace {
276371
NYdb::NTopic::TAlterTopicSettings TCommandTopicAlter::PrepareAlterSettings(
277372
NYdb::NTopic::TDescribeTopicResult& describeResult) {
278373
auto settings = NYdb::NTopic::TAlterTopicSettings();
374+
auto partitioningSettings = settings.BeginAlterPartitioningSettings();
375+
376+
if (MinActivePartitions_.Defined() && (*MinActivePartitions_ != describeResult.GetTopicDescription().GetPartitioningSettings().GetMinActivePartitions())) {
377+
partitioningSettings.MinActivePartitions(*MinActivePartitions_);
378+
}
379+
380+
if (MaxActivePartitions_.Defined() && (*MaxActivePartitions_ != describeResult.GetTopicDescription().GetPartitioningSettings().GetMaxActivePartitions())) {
381+
partitioningSettings.MaxActivePartitions(*MaxActivePartitions_);
382+
}
383+
384+
auto autoscalingSettings = partitioningSettings.BeginAlterAutoscalingSettings();
385+
386+
if (GetScaleThresholdTime().Defined() && *GetScaleThresholdTime() != describeResult.GetTopicDescription().GetPartitioningSettings().GetAutoscalingSettings().GetThresholdTime().Seconds()) {
387+
autoscalingSettings.ThresholdTime(TDuration::Seconds(*GetScaleThresholdTime()));
388+
}
389+
390+
if (GetAutoscalingStrategy().Defined() && *GetAutoscalingStrategy() != describeResult.GetTopicDescription().GetPartitioningSettings().GetAutoscalingSettings().GetStrategy()) {
391+
autoscalingSettings.Strategy(*GetAutoscalingStrategy());
392+
}
393+
394+
if (GetScaleDownThresholdPercent().Defined() && *GetScaleDownThresholdPercent() != describeResult.GetTopicDescription().GetPartitioningSettings().GetAutoscalingSettings().GetScaleDownThresholdPercent()) {
395+
autoscalingSettings.ScaleDownThresholdPercent(*GetScaleDownThresholdPercent());
396+
}
279397

280-
if (PartitionsCount_.Defined() && (*PartitionsCount_ != describeResult.GetTopicDescription().GetTotalPartitionsCount())) {
281-
settings.AlterPartitioningSettings(*PartitionsCount_, *PartitionsCount_);
398+
if (GetScaleUpThresholdPercent().Defined() && *GetScaleUpThresholdPercent() != describeResult.GetTopicDescription().GetPartitioningSettings().GetAutoscalingSettings().GetScaleUpThresholdPercent()) {
399+
autoscalingSettings.ScaleUpThresholdPercent(*GetScaleUpThresholdPercent());
282400
}
283401

284402
auto codecs = GetCodecs();

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

+26-4
Original file line numberDiff line numberDiff line change
@@ -37,12 +37,30 @@ namespace NYdb::NConsoleClient {
3737
NTopic::EMeteringMode MeteringMode_ = NTopic::EMeteringMode::Unspecified;
3838
};
3939

40+
class TCommandWithAutoscaling {
41+
protected:
42+
void AddAutoscaling(TClientCommand::TConfig& config, bool withDefault);
43+
void ParseAutoscalingStrategy();
44+
TMaybe<NTopic::EAutoscalingStrategy> GetAutoscalingStrategy() const;
45+
TMaybe<ui32> GetScaleThresholdTime() const;
46+
TMaybe<ui32> GetScaleUpThresholdPercent() const;
47+
TMaybe<ui32> GetScaleDownThresholdPercent() const;
48+
49+
private:
50+
TMaybe<ui32> ScaleThresholdTime_;
51+
TMaybe<ui32> ScaleUpThresholdPercent_;
52+
TMaybe<ui32> ScaleDownThresholdPercent_;
53+
54+
TString AutoscalingStrategyStr_;
55+
TMaybe<NTopic::EAutoscalingStrategy> AutoscaleStrategy_;
56+
};
57+
4058
class TCommandTopic: public TClientCommandTree {
4159
public:
4260
TCommandTopic();
4361
};
4462

45-
class TCommandTopicCreate: public TYdbCommand, public TCommandWithTopicName, public TCommandWithSupportedCodecs, public TCommandWithMeteringMode {
63+
class TCommandTopicCreate: public TYdbCommand, public TCommandWithTopicName, public TCommandWithSupportedCodecs, public TCommandWithMeteringMode, public TCommandWithAutoscaling {
4664
public:
4765
TCommandTopicCreate();
4866
void Config(TConfig& config) override;
@@ -52,11 +70,13 @@ namespace NYdb::NConsoleClient {
5270
private:
5371
ui64 RetentionPeriodHours_;
5472
ui64 RetentionStorageMb_;
55-
ui32 PartitionsCount_;
73+
ui32 MinActivePartitions_;
74+
ui32 MaxActivePartitions_;
75+
5676
ui32 PartitionWriteSpeedKbps_;
5777
};
5878

59-
class TCommandTopicAlter: public TYdbCommand, public TCommandWithTopicName, public TCommandWithSupportedCodecs, public TCommandWithMeteringMode {
79+
class TCommandTopicAlter: public TYdbCommand, public TCommandWithTopicName, public TCommandWithSupportedCodecs, public TCommandWithMeteringMode, public TCommandWithAutoscaling {
6080
public:
6181
TCommandTopicAlter();
6282
void Config(TConfig& config) override;
@@ -66,7 +86,9 @@ namespace NYdb::NConsoleClient {
6686
private:
6787
TMaybe<ui64> RetentionPeriodHours_;
6888
TMaybe<ui64> RetentionStorageMb_;
69-
TMaybe<ui32> PartitionsCount_;
89+
TMaybe<ui32> MinActivePartitions_;
90+
TMaybe<ui32> MaxActivePartitions_;
91+
7092
TMaybe<ui32> PartitionWriteSpeedKbps_;
7193

7294
NYdb::NTopic::TAlterTopicSettings PrepareAlterSettings(NYdb::NTopic::TDescribeTopicResult& describeResult);

0 commit comments

Comments
 (0)