|
| 1 | +use crate::service::content::content_helpers::blob_key; |
1 | 2 | use crate::service::proto::ContentDigest; |
2 | | -use crate::util; |
| 3 | +use aws_config::{BehaviorVersion, Region}; |
| 4 | +use aws_sdk_s3::Client; |
| 5 | +use aws_sdk_s3::config::Credentials; |
| 6 | +use aws_sdk_s3::error::SdkError; |
| 7 | +use aws_sdk_s3::primitives::ByteStream; |
| 8 | +use std::error::Error; |
| 9 | +use std::fmt::Write; |
3 | 10 | use std::io; |
4 | | -use std::path::PathBuf; |
5 | | -use tokio::fs; |
6 | | -use tokio::io::AsyncWriteExt; |
7 | | - |
8 | | -/// Local filesystem-backed blob store. Each body is written to a file |
9 | | -/// keyed by `{digest_type}_{hex(digest_value)}` under `root`. |
10 | | -/// |
11 | | -/// This is the placeholder object-storage backend; swapping to S3 / |
12 | | -/// GCS is a matter of replacing this type. |
| 11 | + |
| 12 | +fn err_chain<E: Error>(e: &E) -> String { |
| 13 | + let mut out = e.to_string(); |
| 14 | + let mut src = e.source(); |
| 15 | + while let Some(cur) = src { |
| 16 | + let _ = write!(out, ": {cur}"); |
| 17 | + src = cur.source(); |
| 18 | + } |
| 19 | + out |
| 20 | +} |
| 21 | + |
| 22 | +fn sdk_err<E: Error>(e: E) -> io::Error { |
| 23 | + io::Error::other(err_chain(&e)) |
| 24 | +} |
| 25 | + |
| 26 | +/// S3-backed blob store config. When `access_key`/`secret_key` are |
| 27 | +/// unset, the AWS SDK's default credential chain is used (shared |
| 28 | +/// config, container/EC2 metadata, etc.). |
| 29 | +#[derive(Debug, Clone)] |
| 30 | +pub struct ContentFilestoreConfig { |
| 31 | + pub bucket: String, |
| 32 | + pub region: String, |
| 33 | + pub endpoint: Option<String>, |
| 34 | + /// RustFS and other S3-compatible stores require path-style |
| 35 | + pub force_path_style: bool, |
| 36 | + pub access_key: Option<String>, |
| 37 | + pub secret_key: Option<String>, |
| 38 | +} |
| 39 | + |
| 40 | +impl ContentFilestoreConfig { |
| 41 | + pub fn from_env() -> Result<Self, String> { |
| 42 | + let bucket = std::env::var("CONTENT_BLOB_OS_BUCKET") |
| 43 | + .map_err(|_| "CONTENT_BLOB_OS_BUCKET is required".to_string())?; |
| 44 | + let region = std::env::var("CONTENT_BLOB_OS_REGION") |
| 45 | + .unwrap_or_else(|_| "us-east-1".to_string()); |
| 46 | + let endpoint = std::env::var("CONTENT_BLOB_OS_ENDPOINT").ok(); |
| 47 | + let force_path_style = |
| 48 | + std::env::var("CONTENT_BLOB_OS_FORCE_PATH_STYLE") |
| 49 | + .map(|v| matches!(v.as_str(), "true" | "1")) |
| 50 | + .unwrap_or(false); |
| 51 | + let access_key = std::env::var("CONTENT_BLOB_OS_ACCESS_KEY").ok(); |
| 52 | + let secret_key = std::env::var("CONTENT_BLOB_OS_SECRET_KEY").ok(); |
| 53 | + Ok(Self { |
| 54 | + bucket, |
| 55 | + region, |
| 56 | + endpoint, |
| 57 | + force_path_style, |
| 58 | + access_key, |
| 59 | + secret_key, |
| 60 | + }) |
| 61 | + } |
| 62 | +} |
| 63 | + |
| 64 | +/// S3-backed blob store. Each body is stored under the configured |
| 65 | +/// bucket keyed by `{digest_type}_{hex(digest_value)}`. |
13 | 66 | #[derive(Debug, Clone)] |
14 | 67 | pub struct ContentFilestore { |
15 | | - root: PathBuf, |
| 68 | + client: Client, |
| 69 | + bucket: String, |
16 | 70 | } |
17 | 71 |
|
18 | 72 | impl ContentFilestore { |
19 | | - pub fn new(root: PathBuf) -> Self { |
20 | | - Self { root } |
| 73 | + pub async fn new(cfg: ContentFilestoreConfig) -> Self { |
| 74 | + let mut loader = aws_config::defaults(BehaviorVersion::latest()) |
| 75 | + .region(Region::new(cfg.region)); |
| 76 | + if let Some(endpoint) = &cfg.endpoint { |
| 77 | + loader = loader.endpoint_url(endpoint); |
| 78 | + } |
| 79 | + if let (Some(access), Some(secret)) = |
| 80 | + (cfg.access_key.as_deref(), cfg.secret_key.as_deref()) |
| 81 | + { |
| 82 | + loader = loader.credentials_provider(Credentials::new( |
| 83 | + access, |
| 84 | + secret, |
| 85 | + None, |
| 86 | + None, |
| 87 | + "content-blob-os", |
| 88 | + )); |
| 89 | + } |
| 90 | + let aws = loader.load().await; |
| 91 | + |
| 92 | + let s3_cfg = aws_sdk_s3::config::Builder::from(&aws) |
| 93 | + .force_path_style(cfg.force_path_style) |
| 94 | + .build(); |
| 95 | + |
| 96 | + Self { |
| 97 | + client: Client::from_conf(s3_cfg), |
| 98 | + bucket: cfg.bucket, |
| 99 | + } |
21 | 100 | } |
22 | 101 |
|
23 | | - /// Write `body` to the store under the digest's key. Overwrites any |
24 | | - /// existing file — re-uploading the same body is a no-op. |
| 102 | + /// Upload `body` under the digest's key. Re-uploading the same |
| 103 | + /// body overwrites the existing object. |
25 | 104 | pub async fn write_blob( |
26 | 105 | &self, |
27 | 106 | digest: &ContentDigest, |
28 | | - body: &[u8], |
| 107 | + body: Vec<u8>, |
29 | 108 | ) -> io::Result<()> { |
30 | | - fs::create_dir_all(&self.root).await?; |
31 | | - |
32 | | - let path = self.root.join(Self::blob_filename(digest)); |
33 | | - |
34 | | - let mut file = fs::OpenOptions::new() |
35 | | - .create(true) |
36 | | - .write(true) |
37 | | - .truncate(true) |
38 | | - .open(&path) |
39 | | - .await?; |
40 | | - file.write_all(body).await?; |
41 | | - file.flush().await?; |
| 109 | + let key = blob_key(digest); |
| 110 | + self.client |
| 111 | + .put_object() |
| 112 | + .bucket(&self.bucket) |
| 113 | + .key(key) |
| 114 | + .body(ByteStream::from(body)) |
| 115 | + .send() |
| 116 | + .await |
| 117 | + .map_err(sdk_err)?; |
42 | 118 | Ok(()) |
43 | 119 | } |
44 | 120 |
|
45 | | - /// Read a blob body by its digest. Returns `NotFound` if no file |
46 | | - /// exists for the given digest. |
| 121 | + /// Read a blob body by its digest. Returns `NotFound` when the |
| 122 | + /// object does not exist in the bucket. |
47 | 123 | pub async fn read_blob( |
48 | 124 | &self, |
49 | 125 | digest: &ContentDigest, |
50 | 126 | ) -> io::Result<Vec<u8>> { |
51 | | - let path = self.root.join(Self::blob_filename(digest)); |
52 | | - fs::read(&path).await |
53 | | - } |
54 | | - |
55 | | - fn blob_filename(digest: &ContentDigest) -> String { |
56 | | - format!("{}_{}", digest.r#type, util::hex::encode(&digest.value)) |
| 127 | + let key = blob_key(digest); |
| 128 | + let resp = self |
| 129 | + .client |
| 130 | + .get_object() |
| 131 | + .bucket(&self.bucket) |
| 132 | + .key(key) |
| 133 | + .send() |
| 134 | + .await |
| 135 | + .map_err(|e| match e { |
| 136 | + SdkError::ServiceError(svc) if svc.err().is_no_such_key() => { |
| 137 | + io::Error::new(io::ErrorKind::NotFound, "blob not found") |
| 138 | + } |
| 139 | + other => sdk_err(other), |
| 140 | + })?; |
| 141 | + let bytes = resp |
| 142 | + .body |
| 143 | + .collect() |
| 144 | + .await |
| 145 | + .map_err(sdk_err)? |
| 146 | + .into_bytes() |
| 147 | + .to_vec(); |
| 148 | + Ok(bytes) |
57 | 149 | } |
58 | 150 | } |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?v=4&size=40){kind=avatar}
0 commit comments