-
Notifications
You must be signed in to change notification settings - Fork 606
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
YQ-3561 first version of FQ run tool (#14259)
- Loading branch information
1 parent
c62ddf2
commit d6360ab
Showing
27 changed files
with
1,442 additions
and
441 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,4 @@ | ||
| sync_dir | ||
|
|
||
| *.log | ||
| *.sql |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,141 @@ | ||
| Enabled: true | ||
| EnableDynamicNameservice: true | ||
| EnableTaskCounters: true | ||
|
|
||
| CheckpointCoordinator { | ||
| Enabled: true | ||
|
|
||
| Storage { | ||
| TablePrefix: "yq/checkpoints" | ||
| ClientTimeoutSec: 70 | ||
| OperationTimeoutSec: 60 | ||
| CancelAfterSec: 60 | ||
| } | ||
| } | ||
|
|
||
| Common { | ||
| MdbGateway: "https://mdb.api.cloud.yandex.net:443" | ||
| ObjectStorageEndpoint: "https://storage-internal.cloud.yandex.net" | ||
| IdsPrefix: "kr" | ||
| QueryArtifactsCompressionMethod: "zstd_6" | ||
| MonitoringEndpoint: "monitoring.api.cloud.yandex.net" | ||
| KeepInternalErrors: true | ||
| UseNativeProtocolForClickHouse: true | ||
| DisableSslForGenericDataSources: true | ||
| ShowQueryTimeline: true | ||
| } | ||
|
|
||
| ControlPlaneProxy { | ||
| Enabled: true | ||
| } | ||
|
|
||
| ControlPlaneStorage { | ||
| Enabled: true | ||
| StatsMode: STATS_MODE_PROFILE | ||
|
|
||
| AvailableConnection: "OBJECT_STORAGE" | ||
| AvailableConnection: "DATA_STREAMS" | ||
| AvailableConnection: "MONITORING" | ||
| AvailableConnection: "POSTGRESQL_CLUSTER" | ||
| AvailableConnection: "CLICKHOUSE_CLUSTER" | ||
| AvailableConnection: "YDB_DATABASE" | ||
| AvailableConnection: "GREENPLUM_CLUSTER" | ||
| AvailableConnection: "MYSQL_CLUSTER" | ||
|
|
||
| AvailableStreamingConnection: "OBJECT_STORAGE" | ||
| AvailableStreamingConnection: "DATA_STREAMS" | ||
| AvailableStreamingConnection: "MONITORING" | ||
| AvailableStreamingConnection: "YDB_DATABASE" | ||
|
|
||
| AvailableBinding: "OBJECT_STORAGE" | ||
| AvailableBinding: "DATA_STREAMS" | ||
|
|
||
| Storage { | ||
| TablePrefix: "yq/control_plane" | ||
| ClientTimeoutSec: 70 | ||
| OperationTimeoutSec: 60 | ||
| CancelAfterSec: 60 | ||
| } | ||
| } | ||
|
|
||
| DbPool { | ||
| Enabled: true | ||
|
|
||
| Storage { | ||
| TablePrefix: "yq/db_pool" | ||
| ClientTimeoutSec: 70 | ||
| OperationTimeoutSec: 60 | ||
| CancelAfterSec: 60 | ||
| } | ||
| } | ||
|
|
||
| NodesManager { | ||
| Enabled: true | ||
| } | ||
|
|
||
| PendingFetcher { | ||
| Enabled: true | ||
| } | ||
|
|
||
| PrivateApi { | ||
| Enabled: true | ||
| } | ||
|
|
||
| PrivateProxy { | ||
| Enabled: true | ||
| } | ||
|
|
||
| QuotasManager { | ||
| Enabled: true | ||
| } | ||
|
|
||
| RateLimiter { | ||
| Enabled: true | ||
| ControlPlaneEnabled: true | ||
| DataPlaneEnabled: true | ||
|
|
||
| Database { | ||
| TablePrefix: "yq/rate_limiter" | ||
| ClientTimeoutSec: 70 | ||
| OperationTimeoutSec: 60 | ||
| CancelAfterSec: 60 | ||
| } | ||
|
|
||
| Limiters { | ||
| CoordinationNodePath: "limiter_alpha" | ||
| } | ||
| } | ||
|
|
||
| ResourceManager { | ||
| Enabled: true | ||
| } | ||
|
|
||
| RowDispatcher { | ||
| Enabled: true | ||
| SendStatusPeriodSec: 10 | ||
| TimeoutBeforeStartSessionSec: 10 | ||
|
|
||
| CompileService { | ||
| ParallelCompilationLimit: 20 | ||
| } | ||
|
|
||
| Coordinator { | ||
| CoordinationNodePath: "yq/row_dispatcher" | ||
|
|
||
| Database { | ||
| TablePrefix: "yq/row_dispatcher" | ||
| ClientTimeoutSec: 70 | ||
| OperationTimeoutSec: 60 | ||
| CancelAfterSec: 60 | ||
| } | ||
| } | ||
|
|
||
| JsonParser { | ||
| BatchSizeBytes: 1048576 | ||
| BatchCreationTimeoutMs: 1000 | ||
| } | ||
| } | ||
|
|
||
| TestConnection { | ||
| Enabled: true | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,166 @@ | ||
| #include <library/cpp/colorizer/colors.h> | ||
| #include <library/cpp/getopt/last_getopt.h> | ||
|
|
||
| #include <util/datetime/base.h> | ||
|
|
||
| #include <ydb/tests/tools/fqrun/src/fq_runner.h> | ||
| #include <ydb/tests/tools/kqprun/runlib/application.h> | ||
| #include <ydb/tests/tools/kqprun/runlib/utils.h> | ||
|
|
||
| using namespace NKikimrRun; | ||
|
|
||
| namespace NFqRun { | ||
|
|
||
| namespace { | ||
|
|
||
| struct TExecutionOptions { | ||
| TString Query; | ||
|
|
||
| bool HasResults() const { | ||
| return !Query.empty(); | ||
| } | ||
|
|
||
| TRequestOptions GetQueryOptions() const { | ||
| return { | ||
| .Query = Query | ||
| }; | ||
| } | ||
|
|
||
| void Validate(const TRunnerOptions& runnerOptions) const { | ||
| if (!Query && !runnerOptions.FqSettings.MonitoringEnabled && !runnerOptions.FqSettings.GrpcEnabled) { | ||
| ythrow yexception() << "Nothing to execute and is not running as daemon"; | ||
| } | ||
| } | ||
| }; | ||
|
|
||
| void RunArgumentQueries(const TExecutionOptions& executionOptions, TFqRunner& runner) { | ||
| NColorizer::TColors colors = NColorizer::AutoColors(Cout); | ||
|
|
||
| if (executionOptions.Query) { | ||
| Cout << colors.Yellow() << TInstant::Now().ToIsoStringLocal() << " Executing query..." << colors.Default() << Endl; | ||
| if (!runner.ExecuteStreamQuery(executionOptions.GetQueryOptions())) { | ||
| ythrow yexception() << TInstant::Now().ToIsoStringLocal() << " Query execution failed"; | ||
| } | ||
| Cout << colors.Yellow() << TInstant::Now().ToIsoStringLocal() << " Fetching query results..." << colors.Default() << Endl; | ||
| if (!runner.FetchQueryResults()) { | ||
| ythrow yexception() << TInstant::Now().ToIsoStringLocal() << " Fetch query results failed"; | ||
| } | ||
| } | ||
|
|
||
| if (executionOptions.HasResults()) { | ||
| try { | ||
| runner.PrintQueryResults(); | ||
| } catch (...) { | ||
| ythrow yexception() << "Failed to print script results, reason:\n" << CurrentExceptionMessage(); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| void RunAsDaemon() { | ||
| NColorizer::TColors colors = NColorizer::AutoColors(Cout); | ||
|
|
||
| Cout << colors.Yellow() << TInstant::Now().ToIsoStringLocal() << " Initialization finished" << colors.Default() << Endl; | ||
| while (true) { | ||
| Sleep(TDuration::Seconds(1)); | ||
| } | ||
| } | ||
|
|
||
| void RunScript(const TExecutionOptions& executionOptions, const TRunnerOptions& runnerOptions) { | ||
| NColorizer::TColors colors = NColorizer::AutoColors(Cout); | ||
|
|
||
| Cout << colors.Yellow() << TInstant::Now().ToIsoStringLocal() << " Initialization of fq runner..." << colors.Default() << Endl; | ||
| TFqRunner runner(runnerOptions); | ||
|
|
||
| try { | ||
| RunArgumentQueries(executionOptions, runner); | ||
| } catch (const yexception& exception) { | ||
| if (runnerOptions.FqSettings.MonitoringEnabled) { | ||
| Cerr << colors.Red() << CurrentExceptionMessage() << colors.Default() << Endl; | ||
| } else { | ||
| throw exception; | ||
| } | ||
| } | ||
|
|
||
| if (runnerOptions.FqSettings.MonitoringEnabled || runnerOptions.FqSettings.GrpcEnabled) { | ||
| RunAsDaemon(); | ||
| } | ||
|
|
||
| Cout << colors.Yellow() << TInstant::Now().ToIsoStringLocal() << " Finalization of fq runner..." << colors.Default() << Endl; | ||
| } | ||
|
|
||
| class TMain : public TMainBase { | ||
| protected: | ||
| void RegisterOptions(NLastGetopt::TOpts& options) override { | ||
| options.SetTitle("FqRun -- tool to execute stream queries through FQ proxy"); | ||
| options.AddHelpOption('h'); | ||
| options.SetFreeArgsNum(0); | ||
|
|
||
| // Inputs | ||
|
|
||
| options.AddLongOption('p', "query", "Query to execute") | ||
| .RequiredArgument("file") | ||
| .StoreMappedResult(&ExecutionOptions.Query, &LoadFile); | ||
|
|
||
| options.AddLongOption("fq-cfg", "File with FQ config (NFq::NConfig::TConfig for FQ proxy)") | ||
| .RequiredArgument("file") | ||
| .DefaultValue("./configuration/fq_config.conf") | ||
| .Handler1([this](const NLastGetopt::TOptsParser* option) { | ||
| if (!google::protobuf::TextFormat::ParseFromString(LoadFile(TString(option->CurValOrDef())), &RunnerOptions.FqSettings.FqConfig)) { | ||
| ythrow yexception() << "Bad format of FQ configuration"; | ||
| } | ||
| }); | ||
|
|
||
| // Outputs | ||
|
|
||
| options.AddLongOption("result-file", "File with query results (use '-' to write in stdout)") | ||
| .RequiredArgument("file") | ||
| .DefaultValue("-") | ||
| .StoreMappedResultT<TString>(&RunnerOptions.ResultOutput, &GetDefaultOutput); | ||
|
|
||
| TChoices<EResultOutputFormat> resultFormat({ | ||
| {"rows", EResultOutputFormat::RowsJson}, | ||
| {"full-json", EResultOutputFormat::FullJson}, | ||
| {"full-proto", EResultOutputFormat::FullProto} | ||
| }); | ||
| options.AddLongOption('R', "result-format", "Query result format") | ||
| .RequiredArgument("result-format") | ||
| .DefaultValue("rows") | ||
| .Choices(resultFormat.GetChoices()) | ||
| .StoreMappedResultT<TString>(&RunnerOptions.ResultOutputFormat, resultFormat); | ||
|
|
||
| RegisterKikimrOptions(options, RunnerOptions.FqSettings); | ||
| } | ||
|
|
||
| int DoRun(NLastGetopt::TOptsParseResult&&) override { | ||
| ExecutionOptions.Validate(RunnerOptions); | ||
|
|
||
| auto& logConfig = RunnerOptions.FqSettings.LogConfig; | ||
| logConfig.SetDefaultLevel(NActors::NLog::EPriority::PRI_CRIT); | ||
| FillLogConfig(logConfig); | ||
|
|
||
| RunScript(ExecutionOptions, RunnerOptions); | ||
|
|
||
| return 0; | ||
| } | ||
|
|
||
| private: | ||
| TExecutionOptions ExecutionOptions; | ||
| TRunnerOptions RunnerOptions; | ||
| }; | ||
|
|
||
| } // anonymous namespace | ||
|
|
||
| } // namespace NFqRun | ||
|
|
||
| int main(int argc, const char* argv[]) { | ||
| SetupSignalActions(); | ||
|
|
||
| try { | ||
| NFqRun::TMain().Run(argc, argv); | ||
| } catch (...) { | ||
| NColorizer::TColors colors = NColorizer::AutoColors(Cerr); | ||
|
|
||
| Cerr << colors.Red() << CurrentExceptionMessage() << colors.Default() << Endl; | ||
| return 1; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,29 @@ | ||
| #pragma once | ||
|
|
||
| #include <util/generic/string.h> | ||
|
|
||
| #include <ydb/core/fq/libs/config/protos/fq_config.pb.h> | ||
| #include <ydb/core/protos/config.pb.h> | ||
| #include <ydb/tests/tools/kqprun/runlib/settings.h> | ||
|
|
||
| namespace NFqRun { | ||
|
|
||
| constexpr i64 MAX_RESULT_SET_ROWS = 1000; | ||
|
|
||
| struct TFqSetupSettings : public NKikimrRun::TServerSettings { | ||
| NFq::NConfig::TConfig FqConfig; | ||
| NKikimrConfig::TLogConfig LogConfig; | ||
| }; | ||
|
|
||
| struct TRunnerOptions { | ||
| IOutputStream* ResultOutput = nullptr; | ||
| NKikimrRun::EResultOutputFormat ResultOutputFormat = NKikimrRun::EResultOutputFormat::RowsJson; | ||
|
|
||
| TFqSetupSettings FqSettings; | ||
| }; | ||
|
|
||
| struct TRequestOptions { | ||
| TString Query; | ||
| }; | ||
|
|
||
| } // namespace NFqRun |
Oops, something went wrong.