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

partils

Browse Source
This commit is contained in:
Alek Westover
2023-06-19 15:25:53 -04:00
parent a936b8a92b
commit 6b74d1a76a
12 changed files with 119 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1
View File
@@ -999,6 +999,7 @@ dependencies = [
"tar",
"thiserror",
"toml",
"tracing",
"url",
"utils",
"workspace_hack",
alek/env.txt
+1 -1
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 0x7f15e2763130>, 's3_mock_server': <fixtures.neon_fixtures.MockS3Server object at 0x7f15e2762d70>, 'neon_cli': <fixtures.neon_fixtures.NeonCli object at 0x7f15e09c7a30>, 'endpoints': <fixtures.neon_fixtures.EndpointFactory object at 0x7f15e09c79a0>, 'safekeepers': [Safekeeper(env=<fixtures.neon_fixtures.NeonEnv object at 0x7f15e09c7a00>, port=SafekeeperPort(pg=15004, http=15005), id=1, running=True), Safekeeper(env=<fixtures.neon_fixtures.NeonEnv object at 0x7f15e09c7a00>, port=SafekeeperPort(pg=15006, http=15007), id=2, running=True), Safekeeper(env=<fixtures.neon_fixtures.NeonEnv object at 0x7f15e09c7a00>, 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("1a50a17881f6b200e570982460dbf411"), 'initial_timeline': TimelineId("835f77900b841e88d7f7a8721ad3d2ba"), 'pageserver': <fixtures.neon_fixtures.NeonPageserver object at 0x7f15e09c78b0>}
{'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>}
alek/sharedir.txt
+1
View File
@@ -0,0 +1 @@
GOT even farther
compute_tools/src/bin/compute_ctl.rs
+15 -17
View File
@@ -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::<String>("remote-extension-bucket")
.expect("remote-extension bucket is required");
let remote_ext_region = *matches
.get_one::<String>("remote-extension-region")
.expect("remote-extension region is required");
let remote_ext_config = matches
.get_one::<String>("remote-extension-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 http_port = *matches
.get_one::<u16>("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"),
)
}
compute_tools/src/compute.rs
+5
View File
@@ -45,6 +45,10 @@ pub struct ComputeNode {
pub state: Mutex<ComputeState>,
/// `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<std::process::Child> {
let compute_state = self.state.lock().unwrap().clone();
let spec = compute_state.pspec.as_ref().expect("spec must be set");
compute_tools/src/extension_server.rs
+34 -48
View File
@@ -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<RemoteStorageConfig> {
// 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::<toml_edit::Document>()
.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),
}
}
compute_tools/src/http/api.rs
+10 -2
View File
@@ -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<Body>, compute: &Arc<ComputeNode>) -> Response<Body
filename
);
// TODO : left off here...
match extension_server::download_file(&filename).await {
match extension_server::download_file(
filename,
// TODO alek: pass more remote_ext arguments
// compute.remote_ext_bucket.clone(),
// compute.remote_ext_region.clone(),
)
.await
{
Ok(_) => Response::new(Body::from("OK")),
Err(e) => {
error!("download_file failed: {}", e);
control_plane/Cargo.toml
+1
View File
@@ -32,3 +32,4 @@ utils.workspace = true
compute_api.workspace = true
workspace_hack.workspace = true
tracing.workspace = true
control_plane/src/bin/neon_local.rs
+15
View File
@@ -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")
test_ext.control
-5
View File
@@ -1,5 +0,0 @@
# mock extension
comment = 'Mock extension'
default_version = '1.0'
module_pathname = '$libdir/test_ext'
relocatable = true
test_runner/fixtures/neon_fixtures.py
+22 -2
View File
@@ -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(
test_runner/regress/test_download_extensions.py
+14 -4
View File
@@ -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)")