-
Notifications
You must be signed in to change notification settings - Fork 599
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Support Console interoperation through distconf
- Loading branch information
Showing
15 changed files
with
495 additions
and
95 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
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,211 @@ | ||
| #include "distconf.h" | ||
|
|
||
| #include <ydb/library/yaml_config/yaml_config.h> | ||
| #include <library/cpp/streams/zstd/zstd.h> | ||
|
|
||
| namespace NKikimr::NStorage { | ||
|
|
||
| void TDistributedConfigKeeper::ConnectToConsole() { | ||
| STLOG(PRI_DEBUG, BS_NODE, NWDC66, "ConnectToConsole"); | ||
| Y_ABORT_UNLESS(!ConsolePipeId); | ||
| ConsolePipeId = Register(NTabletPipe::CreateClient(SelfId(), MakeConsoleID(), | ||
| NTabletPipe::TClientRetryPolicy::WithRetries())); | ||
| } | ||
|
|
||
| void TDistributedConfigKeeper::DisconnectFromConsole() { | ||
| NTabletPipe::CloseAndForgetClient(SelfId(), ConsolePipeId); | ||
| ConsoleConnected = false; | ||
| } | ||
|
|
||
| void TDistributedConfigKeeper::SendConfigProposeRequest() { | ||
| auto& sc = StorageConfig; | ||
|
|
||
| STLOG(PRI_DEBUG, BS_NODE, NWDC67, "SendConfigProposeRequest", | ||
| (ConsoleConnected, ConsoleConnected), | ||
| (ProposeRequestInFlight, ProposeRequestInFlight), | ||
| (HasStorageConfig, sc.has_value()), | ||
| (StorageConfigFetchYamlHash, StorageConfigFetchYamlHash), | ||
| (StorageConfigYamlVersion, StorageConfigYamlVersion), | ||
| (ProposedConfigHashVersion, ProposedConfigHashVersion), | ||
| (ProposeRequestCookie, ProposeRequestCookie), | ||
| (NextProposeRequestCookie, ProposeRequestCookie + 1)); | ||
|
|
||
| if (!ConsoleConnected) { | ||
| return; | ||
| } | ||
|
|
||
| if (ProposeRequestInFlight) { | ||
| return; // still waiting for previous one | ||
| } | ||
|
|
||
| if (!sc || !sc->HasConfigComposite()) { | ||
| return; // no config yet | ||
| } | ||
|
|
||
| Y_DEBUG_ABORT_UNLESS(!ProposedConfigHashVersion || ProposedConfigHashVersion == std::make_tuple( | ||
| StorageConfigFetchYamlHash, StorageConfigYamlVersion)); | ||
| ProposedConfigHashVersion.emplace(StorageConfigFetchYamlHash, StorageConfigYamlVersion); | ||
| NTabletPipe::SendData(SelfId(), ConsolePipeId, new TEvBlobStorage::TEvControllerProposeConfigRequest( | ||
| StorageConfigFetchYamlHash, StorageConfigYamlVersion), ++ProposeRequestCookie); | ||
| ProposeRequestInFlight = true; | ||
| } | ||
|
|
||
| void TDistributedConfigKeeper::Handle(TEvBlobStorage::TEvControllerValidateConfigResponse::TPtr ev) { | ||
| if (!ConsoleConnected) { | ||
| return; | ||
| } | ||
|
|
||
| const auto it = PendingValidation.find(ev->Cookie); | ||
| if (it == PendingValidation.end()) { | ||
| return; | ||
| } | ||
|
|
||
| TActivationContext::Send(ev->Forward(it->second)); | ||
| PendingValidation.erase(it); | ||
| } | ||
|
|
||
| void TDistributedConfigKeeper::Handle(TEvBlobStorage::TEvControllerProposeConfigResponse::TPtr ev) { | ||
| STLOG(PRI_DEBUG, BS_NODE, NWDC68, "received TEvControllerProposeConfigResponse", | ||
| (ConsoleConnected, ConsoleConnected), | ||
| (ProposeRequestInFlight, ProposeRequestInFlight), | ||
| (Cookie, ev->Cookie), | ||
| (ProposeRequestCookie, ProposeRequestCookie), | ||
| (ProposedConfigHashVersion, ProposedConfigHashVersion), | ||
| (Record, ev->Get()->Record)); | ||
|
|
||
| if (!ConsoleConnected || !ProposeRequestInFlight || ev->Cookie != ProposeRequestCookie) { | ||
| return; | ||
| } | ||
| ProposeRequestInFlight = false; | ||
|
|
||
| const auto& record = ev->Get()->Record; | ||
| switch (record.GetStatus()) { | ||
| case NKikimrBlobStorage::TEvControllerProposeConfigResponse::HashMismatch: | ||
| case NKikimrBlobStorage::TEvControllerProposeConfigResponse::UnexpectedConfig: | ||
| // TODO: error condition; restart? | ||
| ProposedConfigHashVersion.reset(); | ||
| break; | ||
|
|
||
| case NKikimrBlobStorage::TEvControllerProposeConfigResponse::CommitIsNeeded: { | ||
| if (!StorageConfig || !StorageConfig->HasConfigComposite() || ProposedConfigHashVersion != | ||
| std::make_tuple(StorageConfigFetchYamlHash, StorageConfigYamlVersion)) { | ||
| const char *err = "proposed config, but something has gone awfully wrong"; | ||
| STLOG(PRI_CRIT, BS_NODE, NWDC69, err, (StorageConfig, StorageConfig), | ||
| (ProposedConfigHashVersion, ProposedConfigHashVersion), | ||
| (StorageConfigFetchYamlHash, StorageConfigFetchYamlHash), | ||
| (StorageConfigYamlVersion, StorageConfigYamlVersion)); | ||
| Y_DEBUG_ABORT("%s", err); | ||
| return; | ||
| } | ||
|
|
||
| NTabletPipe::SendData(SelfId(), ConsolePipeId, new TEvBlobStorage::TEvControllerConsoleCommitRequest( | ||
| StorageConfigYaml), ++CommitRequestCookie); | ||
| break; | ||
| } | ||
|
|
||
| case NKikimrBlobStorage::TEvControllerProposeConfigResponse::CommitIsNotNeeded: | ||
| // it's okay, just wait for another configuration change or something like that | ||
| ConfigCommittedToConsole = true; | ||
| break; | ||
| } | ||
| } | ||
|
|
||
| void TDistributedConfigKeeper::Handle(TEvBlobStorage::TEvControllerConsoleCommitResponse::TPtr ev) { | ||
| STLOG(PRI_DEBUG, BS_NODE, NWDC70, "received TEvControllerConsoleCommitResponse", | ||
| (ConsoleConnected, ConsoleConnected), | ||
| (Cookie, ev->Cookie), | ||
| (CommitRequestCookie, CommitRequestCookie), | ||
| (Record, ev->Get()->Record)); | ||
|
|
||
| if (!ConsoleConnected || ev->Cookie != CommitRequestCookie) { | ||
| return; | ||
| } | ||
|
|
||
| const auto& record = ev->Get()->Record; | ||
| switch (record.GetStatus()) { | ||
| case NKikimrBlobStorage::TEvControllerConsoleCommitResponse::SessionMismatch: | ||
| DisconnectFromConsole(); | ||
| ConnectToConsole(); | ||
| break; | ||
|
|
||
| case NKikimrBlobStorage::TEvControllerConsoleCommitResponse::NotCommitted: | ||
| break; | ||
|
|
||
| case NKikimrBlobStorage::TEvControllerConsoleCommitResponse::Committed: | ||
| ConfigCommittedToConsole = true; | ||
| break; | ||
| } | ||
|
|
||
| ProposedConfigHashVersion.reset(); | ||
| } | ||
|
|
||
| void TDistributedConfigKeeper::Handle(TEvTabletPipe::TEvClientConnected::TPtr ev) { | ||
| STLOG(PRI_DEBUG, BS_NODE, NWDC71, "received TEvClientConnected", (ConsolePipeId, ConsolePipeId), | ||
| (TabletId, ev->Get()->TabletId), (Status, ev->Get()->Status), (ClientId, ev->Get()->ClientId), | ||
| (ServerId, ev->Get()->ServerId)); | ||
| if (ev->Get()->ClientId == ConsolePipeId) { | ||
| if (ev->Get()->Status == NKikimrProto::OK) { | ||
| ConsoleConnected = true; | ||
| SendConfigProposeRequest(); | ||
| } else { | ||
| OnConsolePipeError(); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| void TDistributedConfigKeeper::Handle(TEvTabletPipe::TEvClientDestroyed::TPtr ev) { | ||
| STLOG(PRI_DEBUG, BS_NODE, NWDC72, "received TEvClientDestroyed", (ConsolePipeId, ConsolePipeId), | ||
| (TabletId, ev->Get()->TabletId), (ClientId, ev->Get()->ClientId), (ServerId, ev->Get()->ServerId)); | ||
| if (ev->Get()->ClientId == ConsolePipeId) { | ||
| OnConsolePipeError(); | ||
| } | ||
| } | ||
|
|
||
| void TDistributedConfigKeeper::OnConsolePipeError() { | ||
| ConsolePipeId = {}; | ||
| ConsoleConnected = false; | ||
| ConfigCommittedToConsole = false; | ||
| ProposedConfigHashVersion.reset(); | ||
| ProposeRequestInFlight = false; | ||
| ++CommitRequestCookie; // to prevent processing any messages | ||
|
|
||
| // cancel any pending requests | ||
| for (const auto& [actorId, cancellation] : PendingConsoleCancellation) { | ||
| cancellation(); | ||
| } | ||
| PendingConsoleCancellation.clear(); | ||
| PendingValidation.clear(); | ||
|
|
||
| ConnectToConsole(); | ||
| } | ||
|
|
||
| std::optional<TString> TDistributedConfigKeeper::UpdateConfigComposite(NKikimrBlobStorage::TStorageConfig& config, | ||
| const TString& yaml, const std::optional<TString>& fetched) { | ||
| TString temp; | ||
| const TString *finalFetchedConfig = fetched ? &fetched.value() : &temp; | ||
|
|
||
| if (!fetched) { // fill in 'to-be-fetched' version of config with version incremented by one | ||
| try { | ||
| auto metadata = NYamlConfig::GetMetadata(yaml); | ||
| metadata.Cluster = metadata.Cluster.value_or("unknown"); // TODO: fix this | ||
| metadata.Version = metadata.Version.value_or(0) + 1; | ||
| temp = NYamlConfig::ReplaceMetadata(yaml, metadata); | ||
| } catch (const std::exception& ex) { | ||
| return ex.what(); | ||
| } | ||
| } | ||
|
|
||
| TStringStream ss; | ||
| { | ||
| TZstdCompress zstd(&ss); | ||
| SaveSize(&zstd, yaml.size()); | ||
| zstd.Write(yaml); | ||
| SaveSize(&zstd, finalFetchedConfig->size()); | ||
| zstd.Write(*finalFetchedConfig); | ||
| } | ||
| config.SetConfigComposite(ss.Str()); | ||
|
|
||
| return {}; | ||
| } | ||
|
|
||
| } |
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
Oops, something went wrong.