Skip to content

Commit

Permalink
Merge 99d2a25 into 2b5467c
Browse files Browse the repository at this point in the history
  • Loading branch information
@dorooleg
dorooleg authored Aug 7, 2024
2 parents 2b5467c + 99d2a25 commit 29b9e38
Show file tree
Hide file tree
Showing 4 changed files with 51 additions and 30 deletions.
2 changes: 2 additions & 0 deletions ydb/core/fq/libs/compute/ydb/synchronization_service/synchronization_service.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -436,6 +436,7 @@ class TSynchronizeScopeActor : public NActors::TActorBootstrapped<TSynchronizeSc

request.Get()->Get()->YDBClient = Client;
request.Get()->Get()->ComputeDatabase = ComputeDatabase;
request.Get()->Get()->Scope = Scope;

Register(NFq::NPrivate::MakeCreateConnectionActor(
SelfId(),
Expand Down Expand Up @@ -465,6 +466,7 @@ class TSynchronizeScopeActor : public NActors::TActorBootstrapped<TSynchronizeSc

request.Get()->Get()->YDBClient = Client;
request.Get()->Get()->ComputeDatabase = ComputeDatabase;
request.Get()->Get()->Scope = Scope;

auto it = Connections.find(binding.second.content().connection_id());
if (it == Connections.end()) {
Expand Down
40 changes: 26 additions & 14 deletions ydb/core/fq/libs/control_plane_proxy/actors/query_utils.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
#include <ydb/core/fq/libs/result_formatter/result_formatter.h>
#include <ydb/core/kqp/provider/yql_kikimr_results.h>
#include <ydb/public/api/protos/draft/fq.pb.h>
#include <ydb/public/lib/fq/scope.h>

namespace NFq {
namespace NPrivate {
Expand Down Expand Up @@ -94,7 +95,9 @@ TString SignAccountId(const TString& id, const TSigner::TPtr& signer) {

TMaybe<TString> CreateSecretObjectQuery(const FederatedQuery::ConnectionSetting& setting,
const TString& name,
const TSigner::TPtr& signer) {
const TSigner::TPtr& signer,
const TString& scope) {
const TString folderId = NYdb::NFq::TScope{scope}.ParseFolder();
using namespace fmt::literals;
TString secretObjects;
auto serviceAccountId = ExtractServiceAccountId(setting);
Expand All @@ -103,7 +106,7 @@ TMaybe<TString> CreateSecretObjectQuery(const FederatedQuery::ConnectionSetting&
R"(
UPSERT OBJECT {sa_secret_name} (TYPE SECRET) WITH value={signature};
)",
"sa_secret_name"_a = EncloseAndEscapeString("k1" + name, '`'),
"sa_secret_name"_a = EncloseAndEscapeString(TStringBuilder{} << "f1_" << folderId << name, '`'),
"signature"_a = EncloseSecret(EncloseAndEscapeString(SignAccountId(serviceAccountId, signer), '"'))) : std::string{};
}

Expand All @@ -113,7 +116,7 @@ TMaybe<TString> CreateSecretObjectQuery(const FederatedQuery::ConnectionSetting&
R"(
UPSERT OBJECT {password_secret_name} (TYPE SECRET) WITH value={password};
)",
"password_secret_name"_a = EncloseAndEscapeString("k2" + name, '`'),
"password_secret_name"_a = EncloseAndEscapeString(TStringBuilder{} << "f2_" << folderId << name, '`'),
"password"_a = EncloseSecret(EncloseAndEscapeString(*password, '"')));
}

Expand All @@ -122,8 +125,10 @@ TMaybe<TString> CreateSecretObjectQuery(const FederatedQuery::ConnectionSetting&

TString CreateAuthParamsQuery(const FederatedQuery::ConnectionSetting& setting,
const TString& name,
const TSigner::TPtr& signer) {
const TSigner::TPtr& signer,
const TString& scope) {
using namespace fmt::literals;
const TString folderId = NYdb::NFq::TScope{scope}.ParseFolder();
auto authMethod = GetYdbComputeAuthMethod(setting);
switch (authMethod) {
case EYdbComputeAuth::UNKNOWN:
Expand All @@ -139,7 +144,7 @@ TString CreateAuthParamsQuery(const FederatedQuery::ConnectionSetting& setting,
)",
"auth_method"_a = ToString(authMethod),
"service_account_id"_a = EncloseAndEscapeString(ExtractServiceAccountId(setting), '"'),
"sa_secret_name"_a = EncloseAndEscapeString(signer ? "k1" + name : TString{}, '"'));
"sa_secret_name"_a = EncloseAndEscapeString(signer ? TStringBuilder{} << "f1_" << folderId << name : TString{}, '"'));
case EYdbComputeAuth::BASIC:
return fmt::format(
R"(,
Expand All @@ -149,7 +154,7 @@ TString CreateAuthParamsQuery(const FederatedQuery::ConnectionSetting& setting,
)",
"auth_method"_a = ToString(authMethod),
"login"_a = EncloseAndEscapeString(GetLogin(setting).GetOrElse({}), '"'),
"password_secret_name"_a = EncloseAndEscapeString("k2" + name, '"'));
"password_secret_name"_a = EncloseAndEscapeString(TStringBuilder{} << "f2_" << folderId << name, '"'));
case EYdbComputeAuth::MDB_BASIC:
return fmt::format(
R"(,
Expand All @@ -161,17 +166,18 @@ TString CreateAuthParamsQuery(const FederatedQuery::ConnectionSetting& setting,
)",
"auth_method"_a = ToString(authMethod),
"service_account_id"_a = EncloseAndEscapeString(ExtractServiceAccountId(setting), '"'),
"sa_secret_name"_a = EncloseAndEscapeString(signer ? "k1" + name : TString{}, '"'),
"sa_secret_name"_a = EncloseAndEscapeString(signer ? TStringBuilder{} << "f1_" << folderId << name : TString{}, '"'),
"login"_a = EncloseAndEscapeString(GetLogin(setting).GetOrElse({}), '"'),
"password_secret_name"_a = EncloseAndEscapeString("k2" + name, '"'));
"password_secret_name"_a = EncloseAndEscapeString(TStringBuilder{} << "f2_" << folderId << name, '"'));
}
}

TString MakeCreateExternalDataSourceQuery(
const FederatedQuery::ConnectionContent& connectionContent,
const TSigner::TPtr& signer,
const NConfig::TCommonConfig& common,
bool replaceIfExists) {
bool replaceIfExists,
const TString& scope) {
using namespace fmt::literals;

TString properties;
Expand Down Expand Up @@ -278,20 +284,26 @@ TString MakeCreateExternalDataSourceQuery(
"auth_params"_a =
CreateAuthParamsQuery(connectionContent.setting(),
connectionContent.name(),
signer));
signer,
scope));
}

TMaybe<TString> DropSecretObjectQuery(const TString& name) {
TMaybe<TString> DropSecretObjectQuery(const TString& name, const TString& scope) {
const TString folderId = NYdb::NFq::TScope{scope}.ParseFolder();
using namespace fmt::literals;
return fmt::format(
R"(
DROP OBJECT {secret_name1} (TYPE SECRET);
DROP OBJECT {secret_name2} (TYPE SECRET);
DROP OBJECT {secret_name3} (TYPE SECRET); -- for backward compatibility
DROP OBJECT {secret_name4} (TYPE SECRET); -- for backward compatibility
DROP OBJECT {secret_name5} (TYPE SECRET); -- for backward compatibility
)",
"secret_name1"_a = EncloseAndEscapeString("k1" + name, '`'),
"secret_name2"_a = EncloseAndEscapeString("k2" + name, '`'),
"secret_name3"_a = EncloseAndEscapeString(name, '`'));
"secret_name1"_a = EncloseAndEscapeString(TStringBuilder{} << "f1_" << folderId << name, '`'),
"secret_name2"_a = EncloseAndEscapeString(TStringBuilder{} << "f2_" << folderId << name, '`'),
"secret_name3"_a = EncloseAndEscapeString(TStringBuilder{} << "k1" << name, '`'),
"secret_name4"_a = EncloseAndEscapeString(TStringBuilder{} << "k2" << name, '`'),
"secret_name5"_a = EncloseAndEscapeString(name, '`'));
}

TString MakeDeleteExternalDataTableQuery(const TString& tableName) {
Expand Down
8 changes: 5 additions & 3 deletions ydb/core/fq/libs/control_plane_proxy/actors/query_utils.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -10,15 +10,17 @@ namespace NPrivate {

TMaybe<TString> CreateSecretObjectQuery(const FederatedQuery::ConnectionSetting& setting,
const TString& name,
const TSigner::TPtr& signer);
const TSigner::TPtr& signer,
const TString& scope);

TMaybe<TString> DropSecretObjectQuery(const TString& name);
TMaybe<TString> DropSecretObjectQuery(const TString& name, const TString& scope);

TString MakeCreateExternalDataSourceQuery(
const FederatedQuery::ConnectionContent& connectionContent,
const TSigner::TPtr& signer,
const NConfig::TCommonConfig& common,
bool replaceIfExists);
bool replaceIfExists,
const TString& scope);

TString MakeDeleteExternalDataSourceQuery(const TString& sourceName);

Expand Down
31 changes: 18 additions & 13 deletions ydb/core/fq/libs/control_plane_proxy/actors/ydb_schema_query_actor.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -557,10 +557,12 @@ IActor* MakeCreateConnectionActor(
computeConfig](const TEvControlPlaneProxy::TEvCreateConnectionRequest::TPtr& req)
-> std::vector<TSchemaQueryTask> {
auto& connectionContent = req->Get()->Request.content();
const auto& scope = req->Get()->Scope;

auto createSecretStatement = CreateSecretObjectQuery(connectionContent.setting(),
connectionContent.name(),
signer);
signer,
scope);

std::vector<TSchemaQueryTask> statements;
if (createSecretStatement) {
Expand Down Expand Up @@ -603,7 +605,7 @@ IActor* MakeCreateConnectionActor(
statements.push_back(TSchemaQueryTask{
.SQL = MakeCreateExternalDataSourceQuery(
connectionContent, signer, commonConfig,
computeConfig.IsReplaceIfExistsSyntaxSupported()),
computeConfig.IsReplaceIfExistsSyntaxSupported(), scope),
.ScheduleErrorRecoverySQLGeneration =
withoutRollback
? NoRecoverySQLGeneration()
Expand Down Expand Up @@ -659,21 +661,23 @@ IActor* MakeModifyConnectionActor(
auto& oldConnectionContent = (*request->Get()->OldConnectionContent);
auto& oldBindings = request->Get()->OldBindingContents;
auto& newConnectionContent = request->Get()->Request.content();
const auto& scope = request->Get()->Scope;

auto dropOldSecret =
DropSecretObjectQuery(oldConnectionContent.name());
DropSecretObjectQuery(oldConnectionContent.name(), scope);
auto createNewSecret =
CreateSecretObjectQuery(newConnectionContent.setting(),
newConnectionContent.name(),
signer);
signer,
scope);

bool replaceSupported = computeConfig.IsReplaceIfExistsSyntaxSupported();
if (replaceSupported &&
oldConnectionContent.name() == newConnectionContent.name()) {
// CREATE OR REPLACE
auto createSecretStatement =
CreateSecretObjectQuery(newConnectionContent.setting(),
newConnectionContent.name(), signer);
newConnectionContent.name(), signer, scope);

std::vector<TSchemaQueryTask> statements;
if (createSecretStatement) {
Expand All @@ -683,7 +687,7 @@ IActor* MakeModifyConnectionActor(

statements.push_back(TSchemaQueryTask{
.SQL = MakeCreateExternalDataSourceQuery(
newConnectionContent, signer, commonConfig, replaceSupported)});
newConnectionContent, signer, commonConfig, replaceSupported, scope)});
return statements;
}

Expand Down Expand Up @@ -712,26 +716,26 @@ IActor* MakeModifyConnectionActor(
statements.push_back(TSchemaQueryTask{
.SQL = TString{MakeDeleteExternalDataSourceQuery(oldConnectionContent.name())},
.RollbackSQL = TString{MakeCreateExternalDataSourceQuery(
oldConnectionContent, signer, commonConfig, false)},
oldConnectionContent, signer, commonConfig, false, scope)},
.ShouldSkipStepOnError = IsPathDoesNotExistIssue});

if (dropOldSecret) {
statements.push_back(TSchemaQueryTask{
.SQL = *dropOldSecret,
.RollbackSQL = CreateSecretObjectQuery(oldConnectionContent.setting(),
oldConnectionContent.name(),
signer),
signer, scope),
.ShouldSkipStepOnError = IsPathDoesNotExistIssue});
}
if (createNewSecret) {
statements.push_back(TSchemaQueryTask{.SQL = *createNewSecret,
.RollbackSQL = DropSecretObjectQuery(
newConnectionContent.name())});
newConnectionContent.name(), scope)});
}

statements.push_back(
TSchemaQueryTask{.SQL = TString{MakeCreateExternalDataSourceQuery(
newConnectionContent, signer, commonConfig, false)},
newConnectionContent, signer, commonConfig, false, scope)},
.RollbackSQL = TString{MakeDeleteExternalDataSourceQuery(
newConnectionContent.name())}});

Expand Down Expand Up @@ -787,23 +791,24 @@ IActor* MakeDeleteConnectionActor(
const TEvControlPlaneProxy::TEvDeleteConnectionRequest::TPtr& request)
-> std::vector<TSchemaQueryTask> {
auto& connectionContent = *request->Get()->ConnectionContent;
const auto& scope = request->Get()->Scope;

auto dropSecret =
DropSecretObjectQuery(connectionContent.name());
DropSecretObjectQuery(connectionContent.name(), scope);

std::vector statements = {
TSchemaQueryTask{.SQL = TString{MakeDeleteExternalDataSourceQuery(
connectionContent.name())},
.RollbackSQL = MakeCreateExternalDataSourceQuery(
connectionContent, signer, commonConfig, false),
connectionContent, signer, commonConfig, false, scope),
.ShouldSkipStepOnError = IsPathDoesNotExistIssue}};
if (dropSecret) {
statements.push_back(
TSchemaQueryTask{.SQL = *dropSecret,
.RollbackSQL =
CreateSecretObjectQuery(connectionContent.setting(),
connectionContent.name(),
signer),
signer, scope),
.ShouldSkipStepOnError = IsPathDoesNotExistIssue});
}
return statements;
Expand Down

0 comments on commit 29b9e38

Please sign in to comment.