rust/neon
1
0
Fork 0
mirror of https://github.com/neondatabase/neon.git synced 2026-01-15 01:12:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

arg passing is mostly working

Browse Source
This commit is contained in:
Alek Westover
2023-06-20 19:35:13 -04:00
parent 6b74d1a76a
commit 890061d371
11 changed files with 107 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -18,3 +18,4 @@ test_output/
*.o
*.so
*.Po
alek/

2
alek/env.txt
View File

@@ -1 +1 @@
{'repo_dir': PosixPath('/home/alek/Desktop/neonX/test_output/test_file_download[debug-pg14]/repo'), 'rust_log_override': None, 'port_distributor': <fixtures.neon_fixtures.PortDistributor object at 0x7fdc68f4e980>, 's3_mock_server': <fixtures.neon_fixtures.MockS3Server object at 0x7fdc68f4ea40>, 'neon_cli': <fixtures.neon_fixtures.NeonCli object at 0x7fdc674b3b50>, 'endpoints': <fixtures.neon_fixtures.EndpointFactory object at 0x7fdc674b3ac0>, 'safekeepers': [Safekeeper(env=<fixtures.neon_fixtures.NeonEnv object at 0x7fdc674b3b20>, port=SafekeeperPort(pg=15004, http=15005), id=1, running=True), Safekeeper(env=<fixtures.neon_fixtures.NeonEnv object at 0x7fdc674b3b20>, port=SafekeeperPort(pg=15006, http=15007), id=2, running=True), Safekeeper(env=<fixtures.neon_fixtures.NeonEnv object at 0x7fdc674b3b20>, port=SafekeeperPort(pg=15008, http=15009), id=3, running=True)], 'broker': NeonBroker(logfile=PosixPath('/home/alek/Desktop/neonX/test_output/test_file_download[debug-pg14]/repo/storage_broker.log'), port=15001, neon_binpath=PosixPath('/home/alek/Desktop/neonX/target/debug'), handle=<Popen: returncode: None args: ['/home/alek/Desktop/neonX/target/debug/stora...>), 'remote_storage': S3Storage(bucket_name='test_file_download', bucket_region='us-east-1', access_key='test', secret_key='test', endpoint='http://127.0.0.1:15000', prefix_in_bucket=None), 'remote_storage_users': <RemoteStorageUsers.PAGESERVER: 1>, 'pg_version': '14', 'neon_binpath': PosixPath('/home/alek/Desktop/neonX/target/debug'), 'pg_distrib_dir': PosixPath('/home/alek/Desktop/neonX/pg_install'), 'endpoint_counter': 0, 'initial_tenant': `TenantId("694e035e5deed6f9ec6ba7c6cdb5b8db"), 'initial_timeline': TimelineId("5a04807f3ddf071e965c2c3c08f59813"), 'pageserver': <fixtures.neon_fixtures.NeonPageserver object at 0x7fdc674b39d0>}
{'repo_dir': PosixPath('/home/alek/Desktop/neonX/test_output/test_file_download[debug-pg14]/repo'), 'rust_log_override': None, 'port_distributor': <fixtures.neon_fixtures.PortDistributor object at 0x7efeeb54e890>, 's3_mock_server': <fixtures.neon_fixtures.MockS3Server object at 0x7efeeb54e950>, 'neon_cli': <fixtures.neon_fixtures.NeonCli object at 0x7efee9ac7a90>, 'endpoints': <fixtures.neon_fixtures.EndpointFactory object at 0x7efee9ac7a00>, 'safekeepers': [Safekeeper(env=<fixtures.neon_fixtures.NeonEnv object at 0x7efee9ac7a60>, port=SafekeeperPort(pg=15005, http=15006), id=1, running=True), Safekeeper(env=<fixtures.neon_fixtures.NeonEnv object at 0x7efee9ac7a60>, port=SafekeeperPort(pg=15007, http=15008), id=2, running=True), Safekeeper(env=<fixtures.neon_fixtures.NeonEnv object at 0x7efee9ac7a60>, port=SafekeeperPort(pg=15009, http=15010), id=3, running=True)], 'broker': NeonBroker(logfile=PosixPath('/home/alek/Desktop/neonX/test_output/test_file_download[debug-pg14]/repo/storage_broker.log'), port=15002, neon_binpath=PosixPath('/home/alek/Desktop/neonX/target/debug'), handle=<Popen: returncode: None args: ['/home/alek/Desktop/neonX/target/debug/stora...>), 'remote_storage': S3Storage(bucket_name='test_file_download', bucket_region='us-east-1', access_key='test', secret_key='test', endpoint='http://127.0.0.1:15001', prefix_in_bucket=None), 'remote_storage_users': <RemoteStorageUsers.PAGESERVER: 1>, 'pg_version': '14', 'neon_binpath': PosixPath('/home/alek/Desktop/neonX/target/debug'), 'pg_distrib_dir': PosixPath('/home/alek/Desktop/neonX/pg_install'), 'endpoint_counter': 0, 'initial_tenant': `TenantId("258f07c42a24166e3c8ee13a6b6a665c"), 'initial_timeline': TimelineId("3dc8832f5f3bdca85aad1e190fd08582"), 'pageserver': <fixtures.neon_fixtures.NeonPageserver object at 0x7efee9ac7910>}

1
alek/output.txt
View File

@@ -1 +0,0 @@
[(0, '0'), (1, '2'), (2, '4'), (3, '6'), (4, '8'), (5, '10'), (6, '12'), (7, '14'), (8, '16'), (9, '18'), (10, '20'), (11, '22'), (12, '24'), (13, '26'), (14, '28'), (15, '30'), (16, '32'), (17, '34'), (18, '36'), (19, '38'), (20, '40'), (21, '42'), (22, '44'), (23, '46'), (24, '48'), (25, '50'), (26, '52'), (27, '54'), (28, '56'), (29, '58'), (30, '60'), (31, '62'), (32, '64'), (33, '66'), (34, '68'), (35, '70'), (36, '72'), (37, '74'), (38, '76'), (39, '78'), (40, '80'), (41, '82'), (42, '84'), (43, '86'), (44, '88'), (45, '90'), (46, '92'), (47, '94'), (48, '96'), (49, '98'), (50, '100'), (51, '102'), (52, '104'), (53, '106'), (54, '108'), (55, '110'), (56, '112'), (57, '114'), (58, '116'), (59, '118'), (60, '120'), (61, '122'), (62, '124'), (63, '126'), (64, '128'), (65, '130'), (66, '132'), (67, '134'), (68, '136'), (69, '138'), (70, '140'), (71, '142'), (72, '144'), (73, '146'), (74, '148'), (75, '150'), (76, '152'), (77, '154'), (78, '156'), (79, '158'), (80, '160'), (81, '162'), (82, '164'), (83, '166'), (84, '168'), (85, '170'), (86, '172'), (87, '174'), (88, '176'), (89, '178'), (90, '180'), (91, '182'), (92, '184'), (93, '186'), (94, '188'), (95, '190'), (96, '192'), (97, '194'), (98, '196'), (99, '198')]

1
alek/sharedir.txt
View File

@@ -1 +0,0 @@
GOT even farther

1
alek/test_ext.control
View File

@@ -3,3 +3,4 @@ comment = 'Mock extension'
default_version = '1.0'
module_pathname = '$libdir/test_ext'
relocatable = true

33
compute_tools/src/bin/compute_ctl.rs
View File

@@ -41,6 +41,7 @@ use std::{thread, time::Duration};
use anyhow::{Context, Result};
use chrono::Utc;
use clap::Arg;
use serde_json::{self, Value};
use tracing::{error, info, warn};
use url::Url;
@@ -54,19 +55,35 @@ use compute_tools::monitor::launch_monitor;
use compute_tools::params::*;
use compute_tools::spec::*;
use toml_edit::Document;
fn main() -> Result<()> {
init_tracing_and_logging(DEFAULT_LOG_LEVEL)?;
let args: Vec<String> = std::env::args().collect();
std::fs::write("alek/ARG", args.join(" "))?;
let matches = cli().get_matches();
let remote_ext_config = matches
.get_one::<String>("remote-extension-config")
.get_one::<String>("remote-ext-config")
.expect("remote-extension-config is required");
// let remote_ext_config = remote_ext_config.parse::<Document>()?;
warn!("YOU MUST BUILD RUST FOR CHANGES TOAPPEAR");
std::fs::write("alek/sharedir.txt", "GOT even farther ")?;
let remote_ext_config: serde_json::Value = serde_json::from_str(&remote_ext_config)?;
let remote_ext_bucket = match &remote_ext_config["bucket"] {
Value::String(x) => x,
_ => panic!("oops"),
};
let remote_ext_region = match &remote_ext_config["region"] {
Value::String(x) => x,
_ => panic!("oops"),
};
warn!("you certainly must build changes if you want rust changes to be built");
std::fs::write("alek/yay", remote_ext_bucket.clone())?;
// compute_tools::extension_server::download_file(
// "test_ext.control",
// remote_ext_bucket.into(),
// remote_ext_region.into(),
// )
// .await?;
let http_port = *matches
.get_one::<u16>("http-port")
@@ -185,8 +202,8 @@ fn main() -> Result<()> {
live_config_allowed,
state: Mutex::new(new_state),
state_changed: Condvar::new(),
// remote_ext_bucket: remote_ext_config["bucket"].to_string(), // TODO ALEK: pass all the args!
// remote_ext_region: remote_ext_config["region"].to_string(),
remote_ext_bucket: remote_ext_bucket.clone(), // TODO ALEK: pass all the args!
remote_ext_region: remote_ext_region.clone(),
};
let compute = Arc::new(compute_node);

5
compute_tools/src/compute.rs
View File

@@ -47,8 +47,8 @@ pub struct ComputeNode {
pub state_changed: Condvar,
// S3 configuration variables:
// TODO: alek pass all args here
// pub remote_ext_bucket: String,
// pub remote_ext_region: String,
pub remote_ext_bucket: String,
pub remote_ext_region: String,
}
#[derive(Clone, Debug)]
@@ -437,7 +437,6 @@ impl ComputeNode {
}
#[instrument(skip(self))]
// ALEK: hmmm?
pub fn start_compute(&self, extension_server_port: u16) -> Result<std::process::Child> {
let compute_state = self.state.lock().unwrap().clone();
let spec = compute_state.pspec.as_ref().expect("spec must be set");

96
compute_tools/src/extension_server.rs
View File

@@ -14,7 +14,7 @@ pub async fn download_file(
// remote_ext_bucket: String,
// remote_ext_region: String,
) -> anyhow::Result<()> {
fs::write("alek/writesomething", "something")?;
// probably should be using the pgbin argv somehow to compute sharedir...
// let sharedir = get_pg_config("--sharedir");
// fs::write("alek/sharedir.txt", sharedir)?;
@@ -22,13 +22,15 @@ pub async fn download_file(
// TODO: download the extensions!
// let s3_config = create_s3_config(remote_ext_bucket, remote_ext_region);
// download_extension(&s3_config, ExtensionType::Shared);
let from_prefix = "/tmp/from_prefix";
let to_prefix = "/tmp/to_prefix";
// download_extension(&s3_config, ExtensionType::Shared).await?;
let filepath = Path::new(from_prefix).join(filename);
let copy_to_filepath = Path::new(to_prefix).join(filename);
fs::copy(filepath, copy_to_filepath)?;
// This is filler code
// let from_prefix = "/tmp/from_prefix";
// let to_prefix = "/tmp/to_prefix";
// let filepath = Path::new(from_prefix).join(filename);
// let copy_to_filepath = Path::new(to_prefix).join(filename);
// fs::copy(filepath, copy_to_filepath)?;
Ok(())
}
@@ -51,6 +53,10 @@ async fn download_helper(
) -> anyhow::Result<()> {
let file_name = remote_from_path.object_name().expect("it must exist");
info!("Downloading {:?}", file_name);
info!(
"To location {:?} (actually just downloading it with it's remote name for now at least)",
to_path
);
let mut download = remote_storage.download(&remote_from_path).await?;
let mut write_data_buffer = Vec::new();
download
@@ -84,40 +90,48 @@ pub async fn download_extension(
let libdir = get_pg_config("--libdir");
let remote_storage = GenericRemoteStorage::from_config(config)?;
match ext_type {
ExtensionType::Shared => {
// 1. Download control files from s3-bucket/public/*.control to SHAREDIR/extension
// We can do this step even before we have spec,
// because public extensions are common for all projects.
let folder = RemotePath::new(Path::new("public_extensions"))?;
let from_paths = remote_storage.list_files(Some(&folder)).await?;
for remote_from_path in from_paths {
if remote_from_path.extension() == Some("control") {
// NOTE: if you run this, it will actually write stuff to your postgress directory
// only run if you are ok with that. TODO: delete this comment
download_helper(&remote_storage, &remote_from_path, &sharedir).await?;
}
}
}
ExtensionType::Tenant(tenant_id) => {
// 2. After we have spec, before project start
// Download control files from s3-bucket/[tenant-id]/*.control to SHAREDIR/extension
let folder = RemotePath::new(Path::new(&format!("{tenant_id}")))?;
let from_paths = remote_storage.list_files(Some(&folder)).await?;
for remote_from_path in from_paths {
if remote_from_path.extension() == Some("control") {
download_helper(&remote_storage, &remote_from_path, &sharedir).await?;
}
}
}
ExtensionType::Library(library_name) => {
// 3. After we have spec, before postgres start
// Download preload_shared_libraries from s3-bucket/public/[library-name].control into LIBDIR/
let from_path = format!("neon-dev-extensions/public/{library_name}.control");
let remote_from_path = RemotePath::new(Path::new(&from_path))?;
download_helper(&remote_storage, &remote_from_path, &libdir).await?;
}
}
// just doing a testing thing
let folder = RemotePath::new(Path::new("public_extensions"))?;
let from_paths = remote_storage.list_files(Some(&folder)).await?;
let some_path = from_paths[0]
.object_name()
.expect("had a problem with somepath in extension server");
fs::write("alek/SOMEPATH", some_path)?;
// match ext_type {
// ExtensionType::Shared => {
// // 1. Download control files from s3-bucket/public/*.control to SHAREDIR/extension
// // We can do this step even before we have spec,
// // because public extensions are common for all projects.
// let folder = RemotePath::new(Path::new("public_extensions"))?;
// let from_paths = remote_storage.list_files(Some(&folder)).await?;
// for remote_from_path in from_paths {
// if remote_from_path.extension() == Some("control") {
// // NOTE: if you run this, it will actually write stuff to your postgress directory
// // only run if you are ok with that. TODO: delete this comment
// download_helper(&remote_storage, &remote_from_path, &sharedir).await?;
// }
// }
// }
// ExtensionType::Tenant(tenant_id) => {
// // 2. After we have spec, before project start
// // Download control files from s3-bucket/[tenant-id]/*.control to SHAREDIR/extension
// let folder = RemotePath::new(Path::new(&format!("{tenant_id}")))?;
// let from_paths = remote_storage.list_files(Some(&folder)).await?;
// for remote_from_path in from_paths {
// if remote_from_path.extension() == Some("control") {
// download_helper(&remote_storage, &remote_from_path, &sharedir).await?;
// }
// }
// }
// ExtensionType::Library(library_name) => {
// // 3. After we have spec, before postgres start
// // Download preload_shared_libraries from s3-bucket/public/[library-name].control into LIBDIR/
// let from_path = format!("neon-dev-extensions/public/{library_name}.control");
// let remote_from_path = RemotePath::new(Path::new(&from_path))?;
// download_helper(&remote_storage, &remote_from_path, &libdir).await?;
// }
// }
Ok(())
}

2
compute_tools/src/http/api.rs
View File

@@ -17,8 +17,6 @@ use tokio::task;
use tracing::{error, info};
use tracing_utils::http::OtelName;
use std::net::TcpStream;
use crate::extension_server;
fn status_response_from_state(state: &ComputeState) -> ComputeStatusResponse {

16
control_plane/src/bin/neon_local.rs
View File

@@ -89,7 +89,7 @@ struct TimelineTreeEl {
fn main() -> Result<()> {
let matches = cli().get_matches();
std::fs::write("notetoalekarrived", "cat")?;
std::fs::write("alek/neon_local", "Ihavearrived")?;
let (sub_name, sub_args) = match matches.subcommand() {
Some(subcommand_data) => subcommand_data,
@@ -660,6 +660,10 @@ fn handle_endpoint(ep_match: &ArgMatches, env: &local_env::LocalEnv) -> Result<(
.get_one::<String>("endpoint_id")
.ok_or_else(|| anyhow!("No endpoint ID was provided to start"))?;
let remote_ext_config = sub_args
.get_one::<String>("remote-ext-config")
.context("remote_ext_config is not an optional parameter")?;
// If --safekeepers argument is given, use only the listed safekeeper nodes.
let safekeepers =
if let Some(safekeepers_str) = sub_args.get_one::<String>("safekeepers") {
@@ -690,6 +694,7 @@ fn handle_endpoint(ep_match: &ArgMatches, env: &local_env::LocalEnv) -> Result<(
.copied()
.unwrap_or(false);
// TODO maybe this is the place
if let Some(endpoint) = endpoint {
match (&endpoint.mode, hot_standby) {
(ComputeMode::Static(_), true) => {
@@ -701,7 +706,7 @@ fn handle_endpoint(ep_match: &ArgMatches, env: &local_env::LocalEnv) -> Result<(
_ => {}
}
println!("Starting existing endpoint {endpoint_id}...");
endpoint.start(&auth_token, safekeepers)?;
endpoint.start(&auth_token, safekeepers, &remote_ext_config)?;
} else {
let branch_name = sub_args
.get_one::<String>("branch-name")
@@ -745,7 +750,8 @@ fn handle_endpoint(ep_match: &ArgMatches, env: &local_env::LocalEnv) -> Result<(
pg_version,
mode,
)?;
ep.start(&auth_token, safekeepers)?;
// TODO: alek this is where the endpoint is created / started
ep.start(&auth_token, safekeepers, &remote_ext_config)?;
}
}
"stop" => {
@@ -945,10 +951,6 @@ fn try_stop_all(env: &local_env::LocalEnv, immediate: bool) {
}
fn cli() -> Command {
std::fs::write("tryingtowritesomefiletoprovethatwegothere", "test")
.expect("file write failed in neonlocal");
info!("alek");
warn!("alek error");
let branch_name_arg = Arg::new("branch-name")
.long("branch-name")
.help("Name of the branch to be created or used as an alias for other services")

14
control_plane/src/endpoint.rs
View File

@@ -401,11 +401,19 @@ impl Endpoint {
Ok(())
}
pub fn start(&self, auth_token: &Option<String>, safekeepers: Vec<NodeId>) -> Result<()> {
pub fn start(
&self,
auth_token: &Option<String>,
safekeepers: Vec<NodeId>,
remote_ext_config: &str,
) -> Result<()> {
if self.status() == "running" {
anyhow::bail!("The endpoint is already running");
}
// ALEK: this is it
// How to tet remote-ext-config into here?
// Slurp the endpoints/<endpoint id>/postgresql.conf file into
// memory. We will include it in the spec file that we pass to
// `compute_ctl`, and `compute_ctl` will write it to the postgresql.conf
@@ -480,8 +488,12 @@ impl Endpoint {
// Launch compute_ctl
println!("Starting postgres node at '{}'", self.connstr());
println!("alek REMOTESTORAGE logfile {:?}", logfile);
let mut cmd = Command::new(self.env.neon_distrib_dir.join("compute_ctl"));
// TODO: alek right here
cmd.args(["--http-port", &self.http_address.port().to_string()])
.args(["--remote-ext-config", remote_ext_config])
.args(["--pgdata", self.pgdata().to_str().unwrap()])
.args(["--connstr", &self.connstr()])
.args([