gaspersic/solana
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

AccountsDb plugin framework (#20047)

Browse Source 
Summary of Changes

Create a plugin mechanism in the accounts update path so that accounts data can be streamed out to external data stores (be it Kafka or Postgres). The plugin mechanism allows

Data stores of connection strings/credentials to be configured,
Accounts with patterns to be streamed
PostgreSQL implementation of the streaming for different destination stores to be plugged in.

The code comprises 4 major parts:

accountsdb-plugin-intf: defines the plugin interface which concrete plugin should implement.
accountsdb-plugin-manager: manages the load/unload of plugins and provide interfaces which the validator can notify of accounts update to plugins.
accountsdb-plugin-postgres: the concrete plugin implementation for PostgreSQL
The validator integrations: updated streamed right after snapshot restore and after account update from transaction processing or other real updates.
The plugin is optionally loaded on demand by new validator CLI argument -- there is no impact if the plugin is not loaded.
This commit is contained in:
Lijun Wang
2021-09-30 14:26:17 -07:00
committed by GitHub
parent 9ef1f06b3f
commit fe97cb2ddf
41 changed files with 1512 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
core/Cargo.toml
View File

@ -35,6 +35,7 @@ retain_mut = "0.1.4"
serde = "1.0.130"
serde_derive = "1.0.103"
solana-account-decoder = { path = "../account-decoder", version = "=1.8.0" }
solana-accountsdb-plugin-manager = { path = "../accountsdb-plugin-manager", version = "=1.8.0" }
solana-client = { path = "../client", version = "=1.8.0" }
solana-config-program = { path = "../programs/config", version = "=1.8.0" }
solana-entry = { path = "../entry", version = "=1.8.0" }

1
core/src/retransmit_stage.rs
View File

@ -506,6 +506,7 @@ mod tests {
None,
None,
accounts_package_sender,
None,
)
.unwrap();
let leader_schedule_cache = Arc::new(cached_leader_schedule);

53
core/src/validator.rs
View File

@ -20,6 +20,7 @@ use {
},
crossbeam_channel::{bounded, unbounded},
rand::{thread_rng, Rng},
solana_accountsdb_plugin_manager::accountsdb_plugin_service::AccountsDbPluginService,
solana_entry::poh::compute_hash_time_ns,
solana_gossip::{
cluster_info::{
@ -63,6 +64,7 @@ use {
solana_runtime::{
accounts_db::{AccountShrinkThreshold, AccountsDbConfig},
accounts_index::AccountSecondaryIndexes,
accounts_update_notifier_interface::AccountsUpdateNotifier,
bank::Bank,
bank_forks::BankForks,
commitment::BlockCommitmentCache,
@ -113,6 +115,7 @@ pub struct ValidatorConfig {
pub account_shrink_paths: Option<Vec<PathBuf>>,
pub rpc_config: JsonRpcConfig,
pub accountsdb_repl_service_config: Option<AccountsDbReplServiceConfig>,
pub accountsdb_plugin_config_file: Option<PathBuf>,
pub rpc_addrs: Option<(SocketAddr, SocketAddr)>, // (JsonRpc, JsonRpcPubSub)
pub pubsub_config: PubSubConfig,
pub snapshot_config: Option<SnapshotConfig>,
@ -173,6 +176,7 @@ impl Default for ValidatorConfig {
account_shrink_paths: None,
rpc_config: JsonRpcConfig::default(),
accountsdb_repl_service_config: None,
accountsdb_plugin_config_file: None,
rpc_addrs: None,
pubsub_config: PubSubConfig::default(),
snapshot_config: None,
@ -279,6 +283,7 @@ pub struct Validator {
ip_echo_server: Option<solana_net_utils::IpEchoServer>,
pub cluster_info: Arc<ClusterInfo>,
accountsdb_repl_service: Option<AccountsDbReplService>,
accountsdb_plugin_service: Option<AccountsDbPluginService>,
}
// in the distant future, get rid of ::new()/exit() and use Result properly...
@ -315,6 +320,27 @@ impl Validator {
warn!("identity: {}", id);
warn!("vote account: {}", vote_account);
let mut bank_notification_senders = Vec::new();
let accountsdb_plugin_service =
if let Some(accountsdb_plugin_config_file) = &config.accountsdb_plugin_config_file {
let (confirmed_bank_sender, confirmed_bank_receiver) = unbounded();
bank_notification_senders.push(confirmed_bank_sender);
let result = AccountsDbPluginService::new(
confirmed_bank_receiver,
accountsdb_plugin_config_file,
);
match result {
Ok(accountsdb_plugin_service) => Some(accountsdb_plugin_service),
Err(err) => {
error!("Failed to load the AccountsDb plugin: {:?}", err);
abort();
}
}
} else {
None
};
if config.voting_disabled {
warn!("voting disabled");
authorized_voter_keypairs.write().unwrap().clear();
@ -380,6 +406,7 @@ impl Validator {
}
let accounts_package_channel = channel();
let (
genesis_config,
bank_forks,
@ -410,6 +437,9 @@ impl Validator {
&start_progress,
config.no_poh_speed_test,
accounts_package_channel.0.clone(),
accountsdb_plugin_service
.as_ref()
.map(|plugin_service| plugin_service.get_accounts_update_notifier()),
);
*start_progress.write().unwrap() = ValidatorStartProgress::StartingServices;
@ -545,13 +575,19 @@ impl Validator {
));
}
let (confirmed_bank_sender, confirmed_bank_receiver) = unbounded();
let accountsdb_repl_service = config.accountsdb_repl_service_config.as_ref().map(|accountsdb_repl_service_config| {
let (bank_notification_sender, bank_notification_receiver) = unbounded();
bank_notification_senders.push(bank_notification_sender);
accountsdb_repl_server_factory::AccountsDbReplServerFactory::build_accountsdb_repl_server(
accountsdb_repl_service_config.clone(), confirmed_bank_receiver, bank_forks.clone())});
accountsdb_repl_service_config.clone(), bank_notification_receiver, bank_forks.clone())
});
let (bank_notification_sender, bank_notification_receiver) = unbounded();
let confirmed_bank_subscribers = if !bank_notification_senders.is_empty() {
Some(Arc::new(RwLock::new(bank_notification_senders)))
} else {
None
};
(
Some(JsonRpcService::new(
rpc_addr,
@ -596,7 +632,7 @@ impl Validator {
bank_forks.clone(),
optimistically_confirmed_bank,
rpc_subscriptions.clone(),
Some(Arc::new(RwLock::new(vec![confirmed_bank_sender]))),
confirmed_bank_subscribers,
)),
Some(bank_notification_sender),
accountsdb_repl_service,
@ -841,6 +877,7 @@ impl Validator {
validator_exit: config.validator_exit.clone(),
cluster_info,
accountsdb_repl_service,
accountsdb_plugin_service,
}
}
@ -951,6 +988,12 @@ impl Validator {
.join()
.expect("accountsdb_repl_service");
}
if let Some(accountsdb_plugin_service) = self.accountsdb_plugin_service {
accountsdb_plugin_service
.join()
.expect("accountsdb_plugin_service");
}
}
}
@ -1093,6 +1136,7 @@ fn new_banks_from_ledger(
start_progress: &Arc<RwLock<ValidatorStartProgress>>,
no_poh_speed_test: bool,
accounts_package_sender: AccountsPackageSender,
accounts_update_notifier: Option<AccountsUpdateNotifier>,
) -> (
GenesisConfig,
BankForks,
@ -1211,6 +1255,7 @@ fn new_banks_from_ledger(
.cache_block_meta_sender
.as_ref(),
accounts_package_sender,
accounts_update_notifier,
)
.unwrap_or_else(|err| {
error!("Failed to load ledger: {:?}", err);

3
core/tests/snapshots.rs
View File

@ -203,6 +203,7 @@ mod tests {
false,
false,
Some(ACCOUNTS_DB_CONFIG_FOR_TESTING),
None,
)
.unwrap();
@ -838,6 +839,7 @@ mod tests {
false,
false,
Some(ACCOUNTS_DB_CONFIG_FOR_TESTING),
None,
)?;
assert_eq!(bank, &deserialized_bank);
@ -1016,6 +1018,7 @@ mod tests {
false,
false,
Some(ACCOUNTS_DB_CONFIG_FOR_TESTING),
None,
)
.unwrap();