From 6b74d1a76a325738bceea2777848acd69f9aa924 Mon Sep 17 00:00:00 2001 From: Alek Westover Date: Mon, 19 Jun 2023 15:25:53 -0400 Subject: [PATCH] partils --- Cargo.lock | 1 + alek/env.txt | 2 +- alek/sharedir.txt | 1 + compute_tools/src/bin/compute_ctl.rs | 32 ++++---- compute_tools/src/compute.rs | 5 ++ compute_tools/src/extension_server.rs | 82 ++++++++----------- compute_tools/src/http/api.rs | 12 ++- control_plane/Cargo.toml | 1 + control_plane/src/bin/neon_local.rs | 15 ++++ test_ext.control | 5 -- test_runner/fixtures/neon_fixtures.py | 24 +++++- .../regress/test_download_extensions.py | 18 +++- 12 files changed, 119 insertions(+), 79 deletions(-) create mode 100644 alek/sharedir.txt delete mode 100644 test_ext.control diff --git a/Cargo.lock b/Cargo.lock index 0f8115d7ac..9d01df52b0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -999,6 +999,7 @@ dependencies = [ "tar", "thiserror", "toml", + "tracing", "url", "utils", "workspace_hack", diff --git a/alek/env.txt b/alek/env.txt index 195d5062b1..ef9f2ea66d 100644 --- a/alek/env.txt +++ b/alek/env.txt @@ -1 +1 @@ -{'repo_dir': PosixPath('/home/alek/Desktop/neonX/test_output/test_file_download[debug-pg14]/repo'), 'rust_log_override': None, 'port_distributor': , 's3_mock_server': , 'neon_cli': , 'endpoints': , 'safekeepers': [Safekeeper(env=, port=SafekeeperPort(pg=15004, http=15005), id=1, running=True), Safekeeper(env=, port=SafekeeperPort(pg=15006, http=15007), id=2, running=True), Safekeeper(env=, 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=), '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': , '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("1a50a17881f6b200e570982460dbf411"), 'initial_timeline': TimelineId("835f77900b841e88d7f7a8721ad3d2ba"), 'pageserver': } \ No newline at end of file +{'repo_dir': PosixPath('/home/alek/Desktop/neonX/test_output/test_file_download[debug-pg14]/repo'), 'rust_log_override': None, 'port_distributor': , 's3_mock_server': , 'neon_cli': , 'endpoints': , 'safekeepers': [Safekeeper(env=, port=SafekeeperPort(pg=15004, http=15005), id=1, running=True), Safekeeper(env=, port=SafekeeperPort(pg=15006, http=15007), id=2, running=True), Safekeeper(env=, 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=), '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': , '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': } \ No newline at end of file diff --git a/alek/sharedir.txt b/alek/sharedir.txt new file mode 100644 index 0000000000..18862e4073 --- /dev/null +++ b/alek/sharedir.txt @@ -0,0 +1 @@ +GOT even farther \ No newline at end of file diff --git a/compute_tools/src/bin/compute_ctl.rs b/compute_tools/src/bin/compute_ctl.rs index a41681ded0..317fd14326 100644 --- a/compute_tools/src/bin/compute_ctl.rs +++ b/compute_tools/src/bin/compute_ctl.rs @@ -41,7 +41,7 @@ use std::{thread, time::Duration}; use anyhow::{Context, Result}; use chrono::Utc; use clap::Arg; -use tracing::{error, info}; +use tracing::{error, info, warn}; use url::Url; use compute_api::responses::ComputeStatus; @@ -54,17 +54,19 @@ 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 matches = cli().get_matches(); - let remote_ext_bucket = *matches - .get_one::("remote-extension-bucket") - .expect("remote-extension bucket is required"); - let remote_ext_region = *matches - .get_one::("remote-extension-region") - .expect("remote-extension region is required"); + let remote_ext_config = matches + .get_one::("remote-extension-config") + .expect("remote-extension-config is required"); + // let remote_ext_config = remote_ext_config.parse::()?; + warn!("YOU MUST BUILD RUST FOR CHANGES TOAPPEAR"); + std::fs::write("alek/sharedir.txt", "GOT even farther ")?; let http_port = *matches .get_one::("http-port") @@ -183,6 +185,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(), }; let compute = Arc::new(compute_node); @@ -353,16 +357,10 @@ fn cli() -> clap::Command { .value_name("CONTROL_PLANE_API_BASE_URI"), ) .arg( - Arg::new("remote-ext-bucket") - .short("u") - .long("remote-ext-bucket") - .value_name("REMOTE_EXT_BUCKET"), - ) - .arg( - Arg::new("remote-ext-region") - .short("r") - .long("remote-ext-region") - .value_name("REMOTE_EXT_REGION"), + Arg::new("remote-ext-config") + .short('r') + .long("remote-ext-config") + .value_name("REMOTE_EXT_CONFIG"), ) } diff --git a/compute_tools/src/compute.rs b/compute_tools/src/compute.rs index f6a3c87fe1..4d14f51c43 100644 --- a/compute_tools/src/compute.rs +++ b/compute_tools/src/compute.rs @@ -45,6 +45,10 @@ pub struct ComputeNode { pub state: Mutex, /// `Condvar` to allow notifying waiters about state changes. pub state_changed: Condvar, + // S3 configuration variables: + // TODO: alek pass all args here + // pub remote_ext_bucket: String, + // pub remote_ext_region: String, } #[derive(Clone, Debug)] @@ -433,6 +437,7 @@ impl ComputeNode { } #[instrument(skip(self))] + // ALEK: hmmm? pub fn start_compute(&self, extension_server_port: u16) -> Result { let compute_state = self.state.lock().unwrap().clone(); let spec = compute_state.pspec.as_ref().expect("spec must be set"); diff --git a/compute_tools/src/extension_server.rs b/compute_tools/src/extension_server.rs index 9e545ed5e3..b2d386be7b 100644 --- a/compute_tools/src/extension_server.rs +++ b/compute_tools/src/extension_server.rs @@ -1,30 +1,28 @@ -use anyhow::{self, Context}; +use anyhow::{self}; use remote_storage::*; +use std::fs; use std::fs::File; -use std::io::{BufWriter, Read, Write}; -use std::net::{TcpListener, TcpStream}; +use std::io::{BufWriter, Write}; +use std::num::{NonZeroU32, NonZeroUsize}; use std::path::Path; use std::str; -use std::{fs, thread}; use tokio::io::AsyncReadExt; -use toml_edit; use tracing::info; -pub fn download_file(mut stream: TcpStream) -> anyhow::Result<()> { - let mut buf = [0; 512]; - - dbg!("ALEK: calling download file"); - // fs.write("world.txt", "hello")?; - - stream.read(&mut buf).expect("Error reading from stream"); - - let filename = str::from_utf8(&buf) - .context("filename is not UTF-8")? - .trim_end(); +pub async fn download_file( + filename: &str, + // remote_ext_bucket: String, + // remote_ext_region: String, +) -> anyhow::Result<()> { + fs::write("alek/writesomething", "something")?; + // let sharedir = get_pg_config("--sharedir"); + // fs::write("alek/sharedir.txt", sharedir)?; println!("requested file {}", filename); - // download_extension(get_s3_config(), ExtensionType::Shared); + // 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"; @@ -32,17 +30,11 @@ pub fn download_file(mut stream: TcpStream) -> anyhow::Result<()> { let copy_to_filepath = Path::new(to_prefix).join(filename); fs::copy(filepath, copy_to_filepath)?; - // Write back the response to the TCP stream - match stream.write("OK".as_bytes()) { - Err(e) => anyhow::bail!("Read-Server: Error writing to stream {}", e), - Ok(_) => (), - } - Ok(()) } +// FIXME: this function panics if it runs into any issues fn get_pg_config(argument: &str) -> String { - // FIXME: this function panics if it runs into any issues let config_output = std::process::Command::new("pg_config") .arg(argument) .output() @@ -129,28 +121,22 @@ pub async fn download_extension( Ok(()) } -pub fn get_s3_config() -> anyhow::Result { - // TODO: Right now we are using the same config parameters as pageserver; but should we have our own configs? - // TODO: Should we read the s3_config from CLI arguments? - let cfg_file_path = Path::new("./../.neon/pageserver.toml"); - let cfg_file_contents = std::fs::read_to_string(cfg_file_path).with_context(|| { - format!( - "Failed to read pageserver config at '{}'", - cfg_file_path.display() - ) - })?; - let toml = cfg_file_contents - .parse::() - .with_context(|| { - format!( - "Failed to parse '{}' as pageserver config", - cfg_file_path.display() - ) - })?; - let remote_storage_data = toml - .get("remote_storage") - .context("field should be present")?; - let remote_storage_config = RemoteStorageConfig::from_toml(remote_storage_data)? - .context("error configuring remote storage")?; - Ok(remote_storage_config) +// TODO: add support for more of these parameters being configurable? +pub fn create_s3_config( + remote_ext_bucket: String, + remote_ext_region: String, +) -> RemoteStorageConfig { + let config = S3Config { + bucket_name: remote_ext_bucket, + bucket_region: remote_ext_region, + prefix_in_bucket: None, + endpoint: None, + concurrency_limit: NonZeroUsize::new(100).expect("100 != 0"), + max_keys_per_list_response: None, + }; + RemoteStorageConfig { + max_concurrent_syncs: NonZeroUsize::new(100).expect("100 != 0"), + max_sync_errors: NonZeroU32::new(100).expect("100 != 0"), + storage: RemoteStorageKind::AwsS3(config), + } } diff --git a/compute_tools/src/http/api.rs b/compute_tools/src/http/api.rs index c00c5a0088..3275b269d7 100644 --- a/compute_tools/src/http/api.rs +++ b/compute_tools/src/http/api.rs @@ -17,6 +17,8 @@ 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 { @@ -135,8 +137,14 @@ async fn routes(req: Request, compute: &Arc) -> Response Response::new(Body::from("OK")), Err(e) => { error!("download_file failed: {}", e); diff --git a/control_plane/Cargo.toml b/control_plane/Cargo.toml index a341ff0263..d2c99c5f36 100644 --- a/control_plane/Cargo.toml +++ b/control_plane/Cargo.toml @@ -32,3 +32,4 @@ utils.workspace = true compute_api.workspace = true workspace_hack.workspace = true +tracing.workspace = true diff --git a/control_plane/src/bin/neon_local.rs b/control_plane/src/bin/neon_local.rs index 52af936d7b..3f3dee3c63 100644 --- a/control_plane/src/bin/neon_local.rs +++ b/control_plane/src/bin/neon_local.rs @@ -28,6 +28,7 @@ use std::path::PathBuf; use std::process::exit; use std::str::FromStr; use storage_broker::DEFAULT_LISTEN_ADDR as DEFAULT_BROKER_ADDR; +use tracing::{error, info, warn}; use utils::{ auth::{Claims, Scope}, id::{NodeId, TenantId, TenantTimelineId, TimelineId}, @@ -88,6 +89,8 @@ struct TimelineTreeEl { fn main() -> Result<()> { let matches = cli().get_matches(); + std::fs::write("notetoalekarrived", "cat")?; + let (sub_name, sub_args) = match matches.subcommand() { Some(subcommand_data) => subcommand_data, None => bail!("no subcommand provided"), @@ -942,6 +945,10 @@ 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") @@ -1002,6 +1009,12 @@ fn cli() -> Command { .help("Additional pageserver's configuration options or overrides, refer to pageserver's 'config-override' CLI parameter docs for more") .required(false); + let remote_ext_config_args = Arg::new("remote-ext-config") + .long("remote-ext-config") + .num_args(1) + .help("Configure the S3 bucket that we search for extensions in.") + .required(true); + let lsn_arg = Arg::new("lsn") .long("lsn") .help("Specify Lsn on the timeline to start from. By default, end of the timeline would be used.") @@ -1139,6 +1152,7 @@ fn cli() -> Command { .required(false)) .arg(pg_version_arg.clone()) .arg(hot_standby_arg.clone()) + .arg(remote_ext_config_args.clone()) // TODO: is this right? ) .subcommand(Command::new("start") .about("Start postgres.\n If the endpoint doesn't exist yet, it is created.") @@ -1152,6 +1166,7 @@ fn cli() -> Command { .arg(pg_version_arg) .arg(hot_standby_arg) .arg(safekeepers_arg) + .arg(remote_ext_config_args.clone()) // TODO: is this right? // update: probably only start and not create needs this ) .subcommand( Command::new("stop") diff --git a/test_ext.control b/test_ext.control deleted file mode 100644 index 2db215bcb8..0000000000 --- a/test_ext.control +++ /dev/null @@ -1,5 +0,0 @@ -# mock extension -comment = 'Mock extension' -default_version = '1.0' -module_pathname = '$libdir/test_ext' -relocatable = true diff --git a/test_runner/fixtures/neon_fixtures.py b/test_runner/fixtures/neon_fixtures.py index 85cff8a17c..d123db7c07 100644 --- a/test_runner/fixtures/neon_fixtures.py +++ b/test_runner/fixtures/neon_fixtures.py @@ -979,6 +979,8 @@ class NeonEnv: safekeeper = Safekeeper(env=self, id=id, port=port) self.safekeepers.append(safekeeper) + # TODO: alek, for reference, this is where neon_cli is init'd. I don't think I need to modify this. + # if you do, maybe do a toml+= extension config type of thing. log.info(f"Config: {toml}") self.neon_cli.init(toml) @@ -1173,6 +1175,7 @@ class AbstractNeonCli(abc.ABC): If `check_return_code`, on non-zero exit code logs failure and raises. """ + log.warn("reached rawcli") assert type(arguments) == list assert type(self.COMMAND) == str @@ -1454,6 +1457,7 @@ class NeonCli(AbstractNeonCli): tenant_id: Optional[TenantId] = None, hot_standby: bool = False, lsn: Optional[Lsn] = None, + remote_ext_config: Optional[str] = None, ) -> "subprocess.CompletedProcess[str]": args = [ "endpoint", @@ -1475,6 +1479,11 @@ class NeonCli(AbstractNeonCli): args.append(endpoint_id) if hot_standby: args.extend(["--hot-standby", "true"]) + if remote_ext_config is not None: + # TODO: I worry that this is not properly escaped... + args.extend(["--remote-ext-config", remote_ext_config]) + log.info(args) + log.warning("ALEK") res = self.raw_cli(args) res.check_returncode() @@ -1488,6 +1497,7 @@ class NeonCli(AbstractNeonCli): safekeepers: Optional[List[int]] = None, tenant_id: Optional[TenantId] = None, lsn: Optional[Lsn] = None, + remote_ext_config: Optional[str] = None, ) -> "subprocess.CompletedProcess[str]": args = [ "endpoint", @@ -1497,6 +1507,8 @@ class NeonCli(AbstractNeonCli): "--pg-version", self.env.pg_version, ] + if remote_ext_config is not None: + args.extend(["--remote-ext-config", remote_ext_config]) if lsn is not None: args.append(f"--lsn={lsn}") args.extend(["--pg-port", str(pg_port)]) @@ -2320,6 +2332,7 @@ class Endpoint(PgProtocol): hot_standby: bool = False, lsn: Optional[Lsn] = None, config_lines: Optional[List[str]] = None, + remote_ext_config: Optional[str] = None, ) -> "Endpoint": """ Create a new Postgres endpoint. @@ -2342,6 +2355,7 @@ class Endpoint(PgProtocol): hot_standby=hot_standby, pg_port=self.pg_port, http_port=self.http_port, + remote_ext_config=remote_ext_config, ) path = Path("endpoints") / self.endpoint_id / "pgdata" self.pgdata_dir = os.path.join(self.env.repo_dir, path) @@ -2356,7 +2370,7 @@ class Endpoint(PgProtocol): return self - def start(self) -> "Endpoint": + def start(self, remote_ext_config: Optional[str] = None) -> "Endpoint": """ Start the Postgres instance. Returns self. @@ -2372,6 +2386,8 @@ class Endpoint(PgProtocol): http_port=self.http_port, tenant_id=self.tenant_id, safekeepers=self.active_safekeepers, + # TODO it's either start or create, not both that starts compute_ctl + remote_ext_config=remote_ext_config, ) self.running = True @@ -2450,6 +2466,7 @@ class Endpoint(PgProtocol): hot_standby: bool = False, lsn: Optional[Lsn] = None, config_lines: Optional[List[str]] = None, + remote_ext_config: Optional[str] = None, ) -> "Endpoint": """ Create an endpoint, apply config, and start Postgres. @@ -2464,7 +2481,8 @@ class Endpoint(PgProtocol): config_lines=config_lines, hot_standby=hot_standby, lsn=lsn, - ).start() + remote_ext_config=remote_ext_config, + ).start(remote_ext_config=remote_ext_config) log.info(f"Postgres startup took {time.time() - started_at} seconds") @@ -2498,6 +2516,7 @@ class EndpointFactory: lsn: Optional[Lsn] = None, hot_standby: bool = False, config_lines: Optional[List[str]] = None, + remote_ext_config: Optional[str] = None, ) -> Endpoint: ep = Endpoint( self.env, @@ -2514,6 +2533,7 @@ class EndpointFactory: hot_standby=hot_standby, config_lines=config_lines, lsn=lsn, + remote_ext_config=remote_ext_config, ) def create( diff --git a/test_runner/regress/test_download_extensions.py b/test_runner/regress/test_download_extensions.py index ede9b37915..730e417932 100644 --- a/test_runner/regress/test_download_extensions.py +++ b/test_runner/regress/test_download_extensions.py @@ -25,6 +25,8 @@ from fixtures.types import Lsn, TenantId, TimelineId from fixtures.utils import wait_until from prometheus_client.samples import Sample +import json + def test_file_download(neon_env_builder: NeonEnvBuilder): """Tests we can download a file""" @@ -44,7 +46,7 @@ def test_file_download(neon_env_builder: NeonEnvBuilder): # we should pass the remote_storage_client to env in the builder. # 4. Upload test_ext.control file to the bucket - # Later this will be done by CI/CD + # In the non-mock version this is done by CI/CD with open("alek/test_ext.control", "rb") as data: neon_env_builder.remote_storage_client.upload_fileobj( data, neon_env_builder.remote_storage.bucket_name, TEST_EXT_PATH @@ -59,12 +61,20 @@ def test_file_download(neon_env_builder: NeonEnvBuilder): with open("pg_install/v15/lib/test_ext.control", "wb") as f: f.write(response.read()) - # env.neon_cli - tenant, _ = env.neon_cli.create_tenant() env.neon_cli.create_timeline("test_file_download", tenant_id=tenant) + + remote_ext_config = json.dumps( + { + "bucket": neon_env_builder.remote_storage.bucket_name, + "region": "eu-central-1", + } + ) + # 6. Start endpoint and ensure that test_ext is present in select * from pg_available_extensions - endpoint = env.endpoints.create_start("test_file_download", tenant_id=tenant) + endpoint = env.endpoints.create_start( + "test_file_download", tenant_id=tenant, remote_ext_config=remote_ext_config + ) with closing(endpoint.connect()) as conn: with conn.cursor() as cur: cur.execute("CREATE TABLE t(key int primary key, value text)")