From a85c9000f420256d99e18712fd8063fd9b5cae7c Mon Sep 17 00:00:00 2001 From: HavenCTO <194730679+HavenCTO@users.noreply.github.com> Date: Fri, 10 Jul 2026 23:26:21 -0400 Subject: [PATCH] Add IPFS storage backend support Add support for storing Quickwit splits on IPFS through the Mutable File System (MFS) of an IPFS node (e.g. Kubo). IPFS storage makes splits content-addressed (with CIDs) and enables sharing, replication, and export across the IPFS network. Implementation includes: - IpfsRpcClient for communicating with node RPC API - IpfsStorage implementing the Storage trait for MFS operations - URI scheme `ipfs://` for addressing MFS directories - Configuration via IpfsStorageConfig with customizable chunking and timeouts - Integration with StorageResolver and the existing storage factory pattern - Environment variable support (QW_IPFS_API_ENDPOINT) - Feature flags for conditional compilation (ipfs, integration-testsuite) - Integration tests against a local Kubo node - Documentation of design decisions and operational notes Kubo service added to docker-compose.yml for local development and testing. --- .env.example | 2 + docker-compose.yml | 23 + docs/configuration/storage-config.md | 43 +- docs/internals/ipfs-storage.md | 101 ++++ quickwit/.dockerignore | 6 + quickwit/Cargo.lock | 1 + quickwit/quickwit-cli/Cargo.toml | 4 + quickwit/quickwit-common/src/uri.rs | 34 +- quickwit/quickwit-config/src/lib.rs | 4 +- .../quickwit-config/src/storage_config.rs | 130 ++++- .../src/metastore_resolver.rs | 1 + quickwit/quickwit-storage/Cargo.toml | 2 + .../src/ipfs_storage/client.rs | 322 ++++++++++++ .../quickwit-storage/src/ipfs_storage/mod.rs | 468 ++++++++++++++++++ quickwit/quickwit-storage/src/lib.rs | 4 + .../quickwit-storage/src/storage_resolver.rs | 18 + .../quickwit-storage/tests/ipfs_storage.rs | 72 +++ 17 files changed, 1230 insertions(+), 5 deletions(-) create mode 100644 docs/internals/ipfs-storage.md create mode 100644 quickwit/.dockerignore create mode 100644 quickwit/quickwit-storage/src/ipfs_storage/client.rs create mode 100644 quickwit/quickwit-storage/src/ipfs_storage/mod.rs create mode 100644 quickwit/quickwit-storage/tests/ipfs_storage.rs diff --git a/.env.example b/.env.example index 9556b302b34..a308cbbebfc 100644 --- a/.env.example +++ b/.env.example @@ -12,6 +12,7 @@ AZURITE_VERSION=latest JAEGER_VERSION=latest OTEL_VERSION=latest PROMETHEUS_VERSION=latest +KUBO_VERSION=latest # Expose the services to outside network MAP_HOST_LOCALSTACK=0.0.0.0 @@ -24,3 +25,4 @@ MAP_HOST_GRAFANA=0.0.0.0 MAP_HOST_JAEGER=0.0.0.0 MAP_HOST_OTEL=0.0.0.0 MAP_HOST_PROMETHEUS=0.0.0.0 +MAP_HOST_KUBO=0.0.0.0 diff --git a/docker-compose.yml b/docker-compose.yml index 24f5ed29b47..c7f19a032ab 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -224,10 +224,33 @@ services: - all - gcp-pubsub + kubo: + image: ipfs/kubo:${KUBO_VERSION:-v0.32.1} + container_name: kubo + # Offline, local-only node for development and tests: the `test` profile + # strips the public bootstrap peers and `--offline` prevents any outbound + # network activity. Remove both to join the public IPFS network. + command: ["daemon", "--migrate=true", "--offline"] + environment: + IPFS_PROFILE: test + ports: + - "${MAP_HOST_KUBO:-127.0.0.1}:5001:5001" # RPC API + - "${MAP_HOST_KUBO:-127.0.0.1}:8082:8080" # Gateway + profiles: + - ipfs + volumes: + - kubo_data:/data/ipfs + healthcheck: + test: ["CMD", "ipfs", "id"] + interval: 1s + timeout: 5s + retries: 100 + volumes: azurite_data: fake_gcs_server_data: grafana_conf: grafana_data: + kubo_data: localstack_data: postgres_data: diff --git a/docs/configuration/storage-config.md b/docs/configuration/storage-config.md index 5c5d24af416..0794f9829d0 100644 --- a/docs/configuration/storage-config.md +++ b/docs/configuration/storage-config.md @@ -5,11 +5,12 @@ sidebar_position: 2 ## Supported Storage Providers -Quickwit currently supports four types of storage providers: +Quickwit currently supports five types of storage providers: - Amazon S3 and S3-compatible (Garage, MinIO, ...) - Azure Blob Storage - Local file storage* - Google Cloud Storage (native API) +- IPFS (InterPlanetary File System, via the RPC API of a node such as [Kubo](https://github.com/ipfs/kubo)) ## Storage URIs @@ -18,6 +19,7 @@ Storage URIs refer to different storage providers identified by a URI "protocol" - `azure://` for Azure Blob Storage - `file://` for local file systems - `gs://` for Google Cloud Storage +- `ipfs://` for IPFS (content-addressed storage) In general, you can use a storage URI or a file path anywhere you would intuitively expect a file path. For instance: - when setting the `index_uri` of an index to specify the storage provider and location; @@ -127,6 +129,45 @@ storage: access_key: your-azure-access-key ``` +### IPFS storage configuration + +Quickwit stores splits in the MFS (Mutable File System) of an IPFS node through its RPC API. MFS presents content-addressed, chunked DAGs as regular files supporting ranged reads, which matches the byte-range access pattern Quickwit uses to open splits. Every split written this way has a CID and is addressable, pinnable, and exportable (e.g. as a CAR file) from the wider IPFS network. + +An `ipfs://` URI maps to an MFS directory on the configured node: `ipfs://quickwit-indexes/my-index` reads and writes under the MFS path `/quickwit-indexes/my-index`. + +| Property | Description | Default value | +| --- | --- | --- | +| `api_endpoint` | Base URL of the IPFS node RPC API. | `http://127.0.0.1:5001` | +| `chunker` | Chunker applied when writing splits. Larger fixed-size chunks reduce the number of blocks a ranged read spans. | `size-1048576` | +| `raw_leaves` | Use raw leaf blocks (no UnixFS envelope around data blocks). | `true` | +| `request_timeout_secs` | Request timeout in seconds for individual RPC calls. | `30` | + +#### Environment variables + +| Env variable | Description | +| --- | --- | +| `QW_IPFS_API_ENDPOINT` | Base URL of the IPFS node RPC API. | + +Example of a storage configuration for IPFS in YAML format: + +```yaml +storage: + ipfs: + api_endpoint: http://127.0.0.1:5001 +``` + +:::note +The IPFS backend covers index data (splits). Keep the metastore on PostgreSQL or a file-backed URI: metastore state is small, mutable, and frequently rewritten, which is a poor fit for content addressing. +::: + +:::note +Deleting a split unlinks it from MFS but only reclaims disk space once the IPFS node runs garbage collection. Kubo does not GC by default: run the daemon with `--enable-gc` or schedule `ipfs repo gc`, especially on indexes with frequent merges. +::: + +:::caution +The IPFS storage backend requires a Quickwit binary compiled with the `ipfs` feature enabled (`--features quickwit-storage/ipfs`). Release binaries built from the default release feature set include it. +::: + ## Storage configuration examples for various object storage providers ### Garage diff --git a/docs/internals/ipfs-storage.md b/docs/internals/ipfs-storage.md new file mode 100644 index 00000000000..d38ab38821a --- /dev/null +++ b/docs/internals/ipfs-storage.md @@ -0,0 +1,101 @@ +# IPFS storage backend + +The `ipfs` feature of `quickwit-storage` adds an `ipfs://` storage backend that +stores splits on an IPFS node, making every split content-addressed: each split +file has a CID and can be pinned, replicated, and exported (e.g. as a CAR file) +across the IPFS network. + +## Design + +### Why MFS + +Quickwit's read path is built around `Storage::get_slice(path, byte_range)`: +one tail read fetches the split footer + hotcache, and subsequent small ranged +reads are mostly served from the hotcache. Raw IPLD blocks are chunk-addressed, +not byte-addressed, so something must translate byte ranges into block fetches. + +Rather than reimplementing that translation, the backend delegates it to the +node's MFS (Mutable File System, the `/api/v0/files/*` RPC surface of +[Kubo](https://github.com/ipfs/kubo)): + +- `files/write` chunks the payload into a UnixFS DAG and links it at an MFS + path. The DAG is the content-addressed artifact; the MFS path is the mutable + name Quickwit uses (`/.split`). +- `files/read?offset=&count=` resolves a byte range to the covering blocks + using the UnixFS index — the exact "contiguous byte range over a set of + blocks" view that `get_slice` needs. +- `files/stat` returns the file size and its CID. + +### Mapping + +| Quickwit | IPFS | +| --- | --- | +| `ipfs://indexes/my-index` (index URI) | MFS directory `/indexes/my-index` | +| `Storage::put` | `files/write` to `.temp`, then `files/mv` into place (atomic publish: readers never see partial splits) | +| `Storage::get_slice` | `files/read` with `offset`/`count` | +| `Storage::get_all` / `copy_to` | `files/stat` (size) + streamed `files/read`, with a length-enforcing reader that fails on truncation | +| `Storage::delete` | `files/rm --force` (missing entries are not an error) | +| `Storage::file_num_bytes` | `files/stat` | +| split CID | `IpfsStorage::file_cid(path)` (`files/stat`) | + +### Chunking and read amplification + +The dominant latency cost is the first read per split (footer + hotcache tail +range). The default chunker is `size-1048576` (1 MiB fixed-size chunks) with +raw leaves, so a typical tail read spans a small number of blocks that a local +node serves in one RPC round-trip. Both knobs are exposed in +`IpfsStorageConfig` (`chunker`, `raw_leaves`). + +### What stays conventional + +- **Metastore.** Metastore state is small, mutable, and rewritten constantly — + the opposite of content-addressed data. Keep it on PostgreSQL or a file + backend. `StorageResolver` and `MetastoreResolver` are independent, so + `index_uri: ipfs://...` + `metastore_uri: postgres://...` composes with no + coupling. +- **Split cache.** The searcher split cache still materializes hot splits on + local disk; IPFS replaces the *origin* storage, not the local cache. +- **Split identity.** A split's CID alone is not sufficient to open it: the + metastore row carries `footer_offsets`. Verifiable replication of an index + needs the split CIDs *plus* a metastore snapshot. + +## Operational notes + +- The backend talks to a single node's RPC API (`QW_IPFS_API_ENDPOINT` or + `storage.ipfs.api_endpoint`, default `http://127.0.0.1:5001`). Run the node + colocated with indexers/searchers; the IPFS network is the replication + layer, not the query path. +- The RPC API grants full control of the node. Never expose port 5001 beyond + localhost / a private network. +- The dev `kubo` service in `docker-compose.yml` runs with the `test` profile + and `--offline`: no bootstrap peers, no outbound traffic. Remove both to + join the public IPFS network — an explicit opt-in, since published CIDs + become discoverable. +- Writes are staged to `.temp` and moved into place; a crash can leave + `.temp` entries behind. They are overwritten (truncate) on retry and can be + garbage-collected safely. +- **Deletes free disk space only after IPFS garbage collection.** The backend + does not use pins: files are GC-protected by being reachable from the MFS + root. `Storage::delete` (`files/rm`) unlinks the entry, which makes its + blocks GC-*eligible*, not deleted. Kubo does not GC by default, so on + merge-heavy indexes the blockstore grows until `ipfs repo gc` runs — run the + daemon with `--enable-gc` (or GC periodically) to reclaim space from deleted + and merged-away splits. Conversely, a split whose CID was explicitly pinned + (e.g. for replication) survives GC after deletion in Quickwit, by design. + +## Testing + +Unit tests: `cargo test -p quickwit-storage --features ipfs,testsuite ipfs`. + +Integration suite (requires a running node, e.g. +`docker compose up -d kubo`): + +```bash +cargo test -p quickwit-storage \ + --features ipfs,integration-testsuite,ci-test \ + --test ipfs_storage +``` + +This runs the generic `storage_test_suite` (put/get/slice/stream/delete/bulk +delete/size/exists) plus single-part and 15 MB multi-part uploads and a CID +round-trip against the node. diff --git a/quickwit/.dockerignore b/quickwit/.dockerignore new file mode 100644 index 00000000000..163308a5355 --- /dev/null +++ b/quickwit/.dockerignore @@ -0,0 +1,6 @@ +target/ +**/target/ +.git/ +node_modules/ +quickwit-ui/node_modules/ +build.log diff --git a/quickwit/Cargo.lock b/quickwit/Cargo.lock index 72079311ab6..0f3455517d1 100644 --- a/quickwit/Cargo.lock +++ b/quickwit/Cargo.lock @@ -10068,6 +10068,7 @@ dependencies = [ "js-sys", "log", "mime", + "mime_guess", "percent-encoding", "pin-project-lite", "quinn", diff --git a/quickwit/quickwit-cli/Cargo.toml b/quickwit/quickwit-cli/Cargo.toml index 01857eb221f..b89f74b4285 100644 --- a/quickwit/quickwit-cli/Cargo.toml +++ b/quickwit/quickwit-cli/Cargo.toml @@ -92,6 +92,7 @@ jemalloc-profiled = [ "quickwit-serve/jemalloc-profiled" ] ci-test = [] +ipfs = ["quickwit-storage/ipfs"] pprof = ["quickwit-serve/pprof"] openssl-support = ["openssl-probe"] # metrics here refers to adding support for metrics ingestion within Quickwit. @@ -114,6 +115,7 @@ release-feature-set = [ "quickwit-serve/lambda", "quickwit-storage/azure", "quickwit-storage/gcs", + "quickwit-storage/ipfs", "quickwit-metastore/postgres", ] release-feature-vendored-set = [ @@ -129,6 +131,7 @@ release-feature-vendored-set = [ "quickwit-serve/lambda", "quickwit-storage/azure", "quickwit-storage/gcs", + "quickwit-storage/ipfs", "quickwit-metastore/postgres", ] release-macos-feature-vendored-set = [ @@ -143,6 +146,7 @@ release-macos-feature-vendored-set = [ "quickwit-serve/lambda", "quickwit-storage/azure", "quickwit-storage/gcs", + "quickwit-storage/ipfs", "quickwit-metastore/postgres", ] release-jemalloc-profiled = [ diff --git a/quickwit/quickwit-common/src/uri.rs b/quickwit/quickwit-common/src/uri.rs index ff191fb04cb..fd79d4d48eb 100644 --- a/quickwit/quickwit-common/src/uri.rs +++ b/quickwit/quickwit-common/src/uri.rs @@ -37,6 +37,7 @@ pub enum Protocol { Ram = 6, S3 = 7, Google = 8, + Ipfs = 9, } impl Protocol { @@ -50,6 +51,7 @@ impl Protocol { Protocol::Ram => "ram", Protocol::S3 => "s3", Protocol::Google => "gs", + Protocol::Ipfs => "ipfs", } } @@ -62,7 +64,10 @@ impl Protocol { } pub fn is_object_storage(&self) -> bool { - matches!(&self, Protocol::Azure | Protocol::S3 | Protocol::Google) + matches!( + &self, + Protocol::Azure | Protocol::S3 | Protocol::Google | Protocol::Ipfs + ) } pub fn is_database(&self) -> bool { @@ -89,6 +94,7 @@ impl FromStr for Protocol { "ram" => Ok(Protocol::Ram), "s3" => Ok(Protocol::S3), "gs" => Ok(Protocol::Google), + "ipfs" => Ok(Protocol::Ipfs), _ => bail!("unknown URI protocol `{protocol}`"), } } @@ -183,6 +189,9 @@ impl Uri { if protocol == Protocol::Google && path.components().count() < 2 { return None; } + if protocol == Protocol::Ipfs && path.components().count() < 2 { + return None; + } let parent_path = path.parent()?; Some(Self { @@ -211,6 +220,9 @@ impl Uri { if self.protocol() == Protocol::Google && path.components().count() < 2 { return None; } + if self.protocol() == Protocol::Ipfs && path.components().count() < 2 { + return None; + } path.file_name().map(Path::new) } @@ -508,6 +520,10 @@ mod tests { Uri::for_test("gs://bucket/key").protocol(), Protocol::Google ); + assert_eq!( + Uri::for_test("ipfs://indexes/key").protocol(), + Protocol::Ipfs + ); assert_eq!( Uri::for_test("postgres://localhost:5432/metastore").protocol(), Protocol::PostgreSQL @@ -661,6 +677,16 @@ mod tests { Uri::for_test("gs://bucket/foo/bar/").parent().unwrap(), "gs://bucket/foo" ); + assert!(Uri::for_test("ipfs://indexes").parent().is_none()); + assert!(Uri::for_test("ipfs://indexes/").parent().is_none()); + assert_eq!( + Uri::for_test("ipfs://indexes/foo").parent().unwrap(), + "ipfs://indexes" + ); + assert_eq!( + Uri::for_test("ipfs://indexes/foo/bar").parent().unwrap(), + "ipfs://indexes/foo" + ); } #[test] @@ -733,6 +759,12 @@ mod tests { Uri::for_test("gs://bucket/foo/").file_name().unwrap(), Path::new("foo"), ); + assert!(Uri::for_test("ipfs://indexes").file_name().is_none()); + assert!(Uri::for_test("ipfs://indexes/").file_name().is_none()); + assert_eq!( + Uri::for_test("ipfs://indexes/foo").file_name().unwrap(), + Path::new("foo"), + ); } #[test] diff --git a/quickwit/quickwit-config/src/lib.rs b/quickwit/quickwit-config/src/lib.rs index d210932b1be..022bf70c0ee 100644 --- a/quickwit/quickwit-config/src/lib.rs +++ b/quickwit/quickwit-config/src/lib.rs @@ -83,8 +83,8 @@ pub use crate::serde_utils::HumanDuration; use crate::source_config::serialize::{SourceConfigV0_7, SourceConfigV0_8, VersionedSourceConfig}; pub use crate::storage_config::{ AzureStorageConfig, ChecksumAlgorithm, FileStorageConfig, GoogleCloudStorageConfig, - RamStorageConfig, S3StorageConfig, StorageBackend, StorageBackendFlavor, StorageConfig, - StorageConfigs, + IpfsStorageConfig, RamStorageConfig, S3StorageConfig, StorageBackend, StorageBackendFlavor, + StorageConfig, StorageConfigs, }; /// Returns true if the ingest API v2 is enabled. diff --git a/quickwit/quickwit-config/src/storage_config.rs b/quickwit/quickwit-config/src/storage_config.rs index d04cd93aaa0..1b75b9e2842 100644 --- a/quickwit/quickwit-config/src/storage_config.rs +++ b/quickwit/quickwit-config/src/storage_config.rs @@ -32,6 +32,8 @@ pub enum StorageBackend { File, /// Google Cloud Storage Google, + /// IPFS (InterPlanetary File System) content-addressed storage + Ipfs, /// In-memory storage, for testing purposes Ram, /// Amazon S3 or S3-compatible storage @@ -146,6 +148,15 @@ impl StorageConfigs { }) } + pub fn find_ipfs(&self) -> Option<&IpfsStorageConfig> { + self.0 + .iter() + .find_map(|storage_config| match storage_config { + StorageConfig::Ipfs(ipfs_storage_config) => Some(ipfs_storage_config), + _ => None, + }) + } + pub fn find_file(&self) -> Option<&FileStorageConfig> { self.0 .iter() @@ -187,6 +198,7 @@ impl Deref for StorageConfigs { pub enum StorageConfig { Azure(AzureStorageConfig), File(FileStorageConfig), + Ipfs(IpfsStorageConfig), Ram(RamStorageConfig), S3(S3StorageConfig), Google(GoogleCloudStorageConfig), @@ -196,7 +208,7 @@ impl StorageConfig { pub fn redact(&mut self) { match self { Self::Azure(azure_storage_config) => azure_storage_config.redact(), - Self::File(_) | Self::Ram(_) | Self::Google(_) => {} + Self::File(_) | Self::Ipfs(_) | Self::Ram(_) | Self::Google(_) => {} Self::S3(s3_storage_config) => s3_storage_config.redact(), } } @@ -215,6 +227,13 @@ impl StorageConfig { } } + pub fn as_ipfs(&self) -> Option<&IpfsStorageConfig> { + match self { + Self::Ipfs(ipfs_storage_config) => Some(ipfs_storage_config), + _ => None, + } + } + pub fn as_ram(&self) -> Option<&RamStorageConfig> { match self { Self::Ram(ram_storage_config) => Some(ram_storage_config), @@ -249,6 +268,12 @@ impl From for StorageConfig { } } +impl From for StorageConfig { + fn from(ipfs_storage_config: IpfsStorageConfig) -> Self { + Self::Ipfs(ipfs_storage_config) + } +} + impl From for StorageConfig { fn from(ram_storage_config: RamStorageConfig) -> Self { Self::Ram(ram_storage_config) @@ -272,6 +297,7 @@ impl StorageConfig { match self { Self::Azure(_) => StorageBackend::Azure, Self::File(_) => StorageBackend::File, + Self::Ipfs(_) => StorageBackend::Ipfs, Self::Ram(_) => StorageBackend::Ram, Self::S3(_) => StorageBackend::S3, Self::Google(_) => StorageBackend::Google, @@ -450,6 +476,73 @@ impl fmt::Debug for S3StorageConfig { #[serde(deny_unknown_fields)] pub struct FileStorageConfig; +/// Configuration for the IPFS storage backend. +/// +/// Splits are stored in the MFS (Mutable File System) of an IPFS node reachable +/// through its RPC API (Kubo's default is `http://127.0.0.1:5001`). MFS presents +/// content-addressed blocks as regular files supporting ranged reads, which is +/// exactly the access pattern Quickwit requires for splits. +#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct IpfsStorageConfig { + /// Base URL of the IPFS node RPC API, e.g. `http://127.0.0.1:5001`. + #[serde(default)] + #[serde(skip_serializing_if = "Option::is_none")] + pub api_endpoint: Option, + /// Chunker applied when writing splits, e.g. `size-1048576`. + /// Larger fixed-size chunks reduce the number of blocks a ranged read spans. + #[serde(default = "IpfsStorageConfig::default_chunker")] + pub chunker: String, + /// Use raw leaf blocks (no UnixFS envelope around data blocks). + #[serde(default = "IpfsStorageConfig::default_raw_leaves")] + pub raw_leaves: bool, + /// Request timeout in seconds for individual RPC calls. + #[serde(default = "IpfsStorageConfig::default_request_timeout_secs")] + pub request_timeout_secs: u64, +} + +impl Default for IpfsStorageConfig { + fn default() -> Self { + Self { + api_endpoint: None, + chunker: Self::default_chunker(), + raw_leaves: Self::default_raw_leaves(), + request_timeout_secs: Self::default_request_timeout_secs(), + } + } +} + +impl IpfsStorageConfig { + /// Environment variable overriding the IPFS RPC API endpoint. + pub const IPFS_API_ENDPOINT_ENV_VAR: &'static str = "QW_IPFS_API_ENDPOINT"; + + /// Kubo's default RPC API endpoint. + pub const DEFAULT_API_ENDPOINT: &'static str = "http://127.0.0.1:5001"; + + fn default_chunker() -> String { + // 1 MiB fixed-size chunks: keeps hotcache/footer reads within a small + // number of blocks while staying under Kubo's block size limits. + "size-1048576".to_string() + } + + fn default_raw_leaves() -> bool { + true + } + + fn default_request_timeout_secs() -> u64 { + 30 + } + + /// Resolves the API endpoint from the environment variable + /// `QW_IPFS_API_ENDPOINT`, the config, or the Kubo default. + pub fn resolve_api_endpoint(&self) -> String { + env::var(Self::IPFS_API_ENDPOINT_ENV_VAR) + .ok() + .or_else(|| self.api_endpoint.clone()) + .unwrap_or_else(|| Self::DEFAULT_API_ENDPOINT.to_string()) + } +} + #[derive(Debug, Clone, Default, Eq, PartialEq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct RamStorageConfig; @@ -654,6 +747,41 @@ mod tests { } } + #[test] + fn test_storage_ipfs_config_serde() { + { + let ipfs_storage_config_yaml = r#" + api_endpoint: http://localhost:5001 + "#; + let ipfs_storage_config: IpfsStorageConfig = + serde_yaml::from_str(ipfs_storage_config_yaml).unwrap(); + + let expected_ipfs_config = IpfsStorageConfig { + api_endpoint: Some("http://localhost:5001".to_string()), + ..Default::default() + }; + assert_eq!(ipfs_storage_config, expected_ipfs_config); + } + { + let ipfs_storage_config_yaml = r#" + api_endpoint: http://ipfs-node:5001 + chunker: size-262144 + raw_leaves: false + request_timeout_secs: 60 + "#; + let ipfs_storage_config: IpfsStorageConfig = + serde_yaml::from_str(ipfs_storage_config_yaml).unwrap(); + + let expected_ipfs_config = IpfsStorageConfig { + api_endpoint: Some("http://ipfs-node:5001".to_string()), + chunker: "size-262144".to_string(), + raw_leaves: false, + request_timeout_secs: 60, + }; + assert_eq!(ipfs_storage_config, expected_ipfs_config); + } + } + #[test] fn test_storage_s3_config_serde() { { diff --git a/quickwit/quickwit-metastore/src/metastore_resolver.rs b/quickwit/quickwit-metastore/src/metastore_resolver.rs index bc4d2cac6e7..5db0c1ba930 100644 --- a/quickwit/quickwit-metastore/src/metastore_resolver.rs +++ b/quickwit/quickwit-metastore/src/metastore_resolver.rs @@ -60,6 +60,7 @@ impl MetastoreResolver { Protocol::File => MetastoreBackend::File, Protocol::Ram => MetastoreBackend::File, Protocol::S3 => MetastoreBackend::File, + Protocol::Ipfs => MetastoreBackend::File, Protocol::PostgreSQL => MetastoreBackend::PostgreSQL, _ => { return Err(MetastoreResolverError::UnsupportedBackend( diff --git a/quickwit/quickwit-storage/Cargo.toml b/quickwit/quickwit-storage/Cargo.toml index 1c377eaacd5..5944c33eac9 100644 --- a/quickwit/quickwit-storage/Cargo.toml +++ b/quickwit/quickwit-storage/Cargo.toml @@ -89,12 +89,14 @@ azure = [ "azure_storage_blobs/enable_reqwest_rustls", ] gcs = ["dep:opendal", "opendal/services-gcs"] +ipfs = ["dep:reqwest", "reqwest/multipart", "reqwest/stream"] ci-test = [] integration-testsuite = [ "azure", "azure_core/azurite_workaround", "azure_storage_blobs/azurite_workaround", "gcs", # Stands for Google cloud storage. + "ipfs", "dep:reqwest", ] testsuite = ["mockall"] diff --git a/quickwit/quickwit-storage/src/ipfs_storage/client.rs b/quickwit/quickwit-storage/src/ipfs_storage/client.rs new file mode 100644 index 00000000000..acc65ece75b --- /dev/null +++ b/quickwit/quickwit-storage/src/ipfs_storage/client.rs @@ -0,0 +1,322 @@ +// Copyright 2021-Present Datadog, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::io; +use std::time::Duration; + +use quickwit_common::retry::{RetryParams, Retryable, retry}; +use serde::Deserialize; +use tokio_util::io::ReaderStream; + +use crate::{PutPayload, StorageError, StorageErrorKind, StorageResolverError}; + +/// Response payload of the `files/stat` RPC. +#[derive(Debug, Deserialize)] +pub(super) struct FilesStatResponse { + /// CID of the MFS entry. This is the content address of the whole + /// (chunked) file and can be shared, pinned or exported as a CAR file. + #[serde(rename = "Hash")] + pub cid: String, + #[serde(rename = "Size")] + pub size: u64, + #[serde(rename = "Type")] + pub entry_type: String, +} + +#[derive(Debug, Deserialize)] +struct RpcErrorBody { + #[serde(rename = "Message")] + message: String, +} + +/// Error returned by the Kubo RPC API. +/// +/// Kubo reports application errors (including "file does not exist") with an +/// HTTP 500 status and a JSON body, so error classification relies on the +/// error message rather than the status code. +#[derive(Debug, thiserror::Error)] +pub(super) enum IpfsRpcError { + #[error("IPFS RPC transport error: {0}")] + Transport(#[from] reqwest::Error), + #[error("IPFS RPC error (status={status}): {message}")] + Api { status: u16, message: String }, + #[error("failed to read payload: {0}")] + PayloadIo(#[from] io::Error), +} + +impl IpfsRpcError { + pub fn is_not_found(&self) -> bool { + match self { + IpfsRpcError::Api { message, .. } => { + message.contains("does not exist") || message.contains("no such file") + } + _ => false, + } + } +} + +impl Retryable for IpfsRpcError { + fn is_retryable(&self) -> bool { + match self { + IpfsRpcError::Transport(error) => error.is_timeout() || error.is_connect(), + IpfsRpcError::Api { status, .. } => matches!(status, 429 | 502 | 503 | 504), + IpfsRpcError::PayloadIo(_) => false, + } + } +} + +impl From for StorageError { + fn from(rpc_error: IpfsRpcError) -> Self { + if rpc_error.is_not_found() { + return StorageErrorKind::NotFound.with_error(rpc_error); + } + match &rpc_error { + IpfsRpcError::Transport(error) if error.is_timeout() => { + StorageErrorKind::Timeout.with_error(rpc_error) + } + IpfsRpcError::Transport(error) if error.is_connect() => { + StorageErrorKind::Service.with_error(rpc_error) + } + IpfsRpcError::Transport(_) | IpfsRpcError::PayloadIo(_) => { + StorageErrorKind::Io.with_error(rpc_error) + } + IpfsRpcError::Api { .. } => StorageErrorKind::Internal.with_error(rpc_error), + } + } +} + +/// Minimal client for the subset of the Kubo RPC API (`/api/v0/files/*`) +/// required to serve Quickwit splits from MFS. +#[derive(Clone)] +pub(super) struct IpfsRpcClient { + http_client: reqwest::Client, + api_base_url: String, + chunker: String, + raw_leaves: bool, + request_timeout: Duration, + retry_params: RetryParams, +} + +impl IpfsRpcClient { + pub fn new( + api_endpoint: &str, + chunker: String, + raw_leaves: bool, + request_timeout: Duration, + ) -> Result { + let http_client = reqwest::Client::builder() + .connect_timeout(Duration::from_secs(10)) + .build() + .map_err(|error| StorageResolverError::FailedToOpenStorage { + kind: StorageErrorKind::Service, + message: format!("failed to build IPFS HTTP client: {error}"), + })?; + let api_base_url = format!("{}/api/v0", api_endpoint.trim_end_matches('/')); + Ok(Self { + http_client, + api_base_url, + chunker, + raw_leaves, + request_timeout, + retry_params: RetryParams::standard(), + }) + } + + fn endpoint_url(&self, endpoint: &str) -> String { + format!("{}/{endpoint}", self.api_base_url) + } + + /// Issues a small (non-streaming) RPC and returns the successful response. + async fn post_small( + &self, + endpoint: &str, + query: &[(&str, &str)], + ) -> Result { + let response = self + .http_client + .post(self.endpoint_url(endpoint)) + .timeout(self.request_timeout) + .query(query) + .send() + .await?; + into_checked_response(response).await + } + + pub async fn version(&self) -> Result<(), IpfsRpcError> { + self.post_small("version", &[]).await?; + Ok(()) + } + + pub async fn files_mkdir(&self, mfs_path: &str) -> Result<(), IpfsRpcError> { + retry(&self.retry_params, || async { + self.post_small("files/mkdir", &[("arg", mfs_path), ("parents", "true")]) + .await + }) + .await?; + Ok(()) + } + + pub async fn files_stat(&self, mfs_path: &str) -> Result { + let response = retry(&self.retry_params, || async { + self.post_small("files/stat", &[("arg", mfs_path)]).await + }) + .await?; + let stat_response: FilesStatResponse = response.json().await?; + Ok(stat_response) + } + + /// Removes an MFS entry. Missing entries are not treated as an error. + pub async fn files_rm(&self, mfs_path: &str) -> Result<(), IpfsRpcError> { + let rm_result = retry(&self.retry_params, || async { + self.post_small( + "files/rm", + &[("arg", mfs_path), ("recursive", "true"), ("force", "true")], + ) + .await + }) + .await; + match rm_result { + Ok(_) => Ok(()), + Err(rpc_error) if rpc_error.is_not_found() => Ok(()), + Err(rpc_error) => Err(rpc_error), + } + } + + pub async fn files_mv(&self, src_path: &str, dst_path: &str) -> Result<(), IpfsRpcError> { + retry(&self.retry_params, || async { + self.post_small("files/mv", &[("arg", src_path), ("arg", dst_path)]) + .await + }) + .await?; + Ok(()) + } + + /// Writes the full payload to `mfs_path`, creating parent directories and + /// truncating any existing entry. Retries restart the upload from scratch, + /// which is safe because `truncate=true` makes the write idempotent. + pub async fn files_write( + &self, + mfs_path: &str, + payload: &dyn PutPayload, + ) -> Result<(), IpfsRpcError> { + retry(&self.retry_params, || async { + self.files_write_attempt(mfs_path, payload).await + }) + .await + } + + async fn files_write_attempt( + &self, + mfs_path: &str, + payload: &dyn PutPayload, + ) -> Result<(), IpfsRpcError> { + let payload_len = payload.len(); + let payload_reader = payload.byte_stream().await?.into_async_read(); + let body = reqwest::Body::wrap_stream(ReaderStream::new(payload_reader)); + let part = reqwest::multipart::Part::stream_with_length(body, payload_len) + .file_name("data") + .mime_str("application/octet-stream")?; + let form = reqwest::multipart::Form::new().part("file", part); + let raw_leaves = if self.raw_leaves { "true" } else { "false" }; + let response = self + .http_client + .post(self.endpoint_url("files/write")) + .query(&[ + ("arg", mfs_path), + ("create", "true"), + ("truncate", "true"), + ("parents", "true"), + ("raw-leaves", raw_leaves), + ("chunker", &self.chunker), + ]) + .multipart(form) + .send() + .await?; + into_checked_response(response).await?; + Ok(()) + } + + /// Reads `count` bytes starting at `offset` and returns them in memory. + pub async fn files_read_range( + &self, + mfs_path: &str, + offset: u64, + count: u64, + ) -> Result, IpfsRpcError> { + let bytes = retry(&self.retry_params, || async { + let response = self + .files_read_response(mfs_path, offset, Some(count)) + .await?; + let bytes = response.bytes().await?; + Ok::<_, IpfsRpcError>(bytes) + }) + .await?; + Ok(bytes.to_vec()) + } + + /// Opens a streaming read starting at `offset`. When `count` is `None`, + /// the stream covers the remainder of the file. + pub async fn files_read_stream( + &self, + mfs_path: &str, + offset: u64, + count: Option, + ) -> Result { + retry(&self.retry_params, || async { + self.files_read_response(mfs_path, offset, count).await + }) + .await + } + + async fn files_read_response( + &self, + mfs_path: &str, + offset: u64, + count_opt: Option, + ) -> Result { + let offset_str = offset.to_string(); + let mut query: Vec<(&str, &str)> = vec![("arg", mfs_path), ("offset", &offset_str)]; + let count_str: String; + if let Some(count) = count_opt { + count_str = count.to_string(); + query.push(("count", &count_str)); + } + let response = self + .http_client + .post(self.endpoint_url("files/read")) + .query(&query) + .send() + .await?; + into_checked_response(response).await + } +} + +/// Turns a Kubo HTTP response into an error if the status is not a success, +/// extracting the error message from the JSON error body when present. +async fn into_checked_response( + response: reqwest::Response, +) -> Result { + let status = response.status(); + if status.is_success() { + return Ok(response); + } + let message = match response.json::().await { + Ok(error_body) => error_body.message, + Err(_) => format!("unexpected HTTP status {status}"), + }; + Err(IpfsRpcError::Api { + status: status.as_u16(), + message, + }) +} diff --git a/quickwit/quickwit-storage/src/ipfs_storage/mod.rs b/quickwit/quickwit-storage/src/ipfs_storage/mod.rs new file mode 100644 index 00000000000..698a0466a32 --- /dev/null +++ b/quickwit/quickwit-storage/src/ipfs_storage/mod.rs @@ -0,0 +1,468 @@ +// Copyright 2021-Present Datadog, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! IPFS storage backend. +//! +//! Stores Quickwit splits in the MFS (Mutable File System) of an IPFS node +//! (e.g. [Kubo](https://github.com/ipfs/kubo)) through its RPC API. MFS +//! presents content-addressed, chunked DAGs as regular files supporting +//! offset/count reads, which matches the byte-range access pattern of +//! [`Storage::get_slice`] on split files. +//! +//! URI scheme: `ipfs:///`, mapped to the MFS directory +//! `//` on the node targeted by the configured RPC API +//! endpoint. Every file written this way has a CID (retrievable via +//! `files/stat`) and is therefore addressable, pinnable, and exportable as a +//! CAR file from the wider IPFS network. + +mod client; + +use std::fmt; +use std::ops::Range; +use std::path::Path; +use std::pin::Pin; +use std::sync::Arc; +use std::task::{Context, Poll}; +use std::time::Duration; + +use async_trait::async_trait; +use futures::{StreamExt, TryStreamExt}; +use quickwit_common::uri::{Protocol, Uri}; +use quickwit_config::{IpfsStorageConfig, StorageBackend}; +use tantivy::directory::OwnedBytes; +use tokio::io::{AsyncRead, AsyncWriteExt, ReadBuf}; +use tokio_util::io::StreamReader; +use tracing::instrument; + +use self::client::IpfsRpcClient; +use crate::debouncer::DebouncedStorage; +use crate::metrics::object_storage_get_slice_in_flight_guards; +use crate::storage::SendableAsync; +use crate::{ + BulkDeleteError, DeleteFailure, PutPayload, Storage, StorageError, StorageErrorKind, + StorageFactory, StorageResolverError, StorageResult, +}; + +/// IPFS storage resolver. +pub struct IpfsStorageFactory { + storage_config: IpfsStorageConfig, +} + +impl IpfsStorageFactory { + /// Creates a new IPFS storage factory. + pub fn new(storage_config: IpfsStorageConfig) -> Self { + Self { storage_config } + } +} + +#[async_trait] +impl StorageFactory for IpfsStorageFactory { + fn backend(&self) -> StorageBackend { + StorageBackend::Ipfs + } + + async fn resolve(&self, uri: &Uri) -> Result, StorageResolverError> { + let storage = IpfsStorage::from_uri(&self.storage_config, uri)?; + Ok(Arc::new(DebouncedStorage::new(storage))) + } +} + +/// IPFS storage implementation backed by the MFS of an IPFS node. +pub struct IpfsStorage { + uri: Uri, + /// MFS directory acting as the root of this storage, e.g. `/quickwit-indexes/my-index`. + mfs_root: String, + client: IpfsRpcClient, +} + +impl fmt::Debug for IpfsStorage { + fn fmt(&self, formatter: &mut fmt::Formatter) -> fmt::Result { + formatter + .debug_struct("IpfsStorage") + .field("uri", &self.uri) + .field("mfs_root", &self.mfs_root) + .finish() + } +} + +impl IpfsStorage { + /// Creates an [`IpfsStorage`] from an `ipfs://` URI and a config. + pub fn from_uri( + storage_config: &IpfsStorageConfig, + uri: &Uri, + ) -> Result { + if uri.protocol() != Protocol::Ipfs { + let message = format!("URI `{uri}` is not a valid IPFS URI"); + return Err(StorageResolverError::InvalidUri(message)); + } + let mfs_root = mfs_root_from_uri(uri)?; + let client = IpfsRpcClient::new( + &storage_config.resolve_api_endpoint(), + storage_config.chunker.clone(), + storage_config.raw_leaves, + Duration::from_secs(storage_config.request_timeout_secs), + )?; + Ok(Self { + uri: uri.clone(), + mfs_root, + client, + }) + } + + /// Returns the CID of the file at `path`, i.e. the content address under + /// which the whole split can be fetched, pinned, or exported from IPFS. + pub async fn file_cid(&self, path: &Path) -> StorageResult { + let mfs_path = self.mfs_path(path)?; + let stat_response = self + .client + .files_stat(&mfs_path) + .await + .map_err(StorageError::from)?; + Ok(stat_response.cid) + } + + fn mfs_path(&self, relative_path: &Path) -> StorageResult { + ensure_valid_relative_path(relative_path)?; + let relative_str = relative_path.to_string_lossy(); + if relative_str.is_empty() { + return Ok(self.mfs_root.clone()); + } + Ok(format!("{}/{relative_str}", self.mfs_root)) + } + + async fn delete_single_file(&self, path: &Path) -> StorageResult<()> { + let mfs_path = self.mfs_path(path)?; + self.client + .files_rm(&mfs_path) + .await + .map_err(StorageError::from)?; + Ok(()) + } +} + +/// Extracts the MFS root directory from an `ipfs://` URI: +/// `ipfs://indexes/my-index` -> `/indexes/my-index`. +fn mfs_root_from_uri(uri: &Uri) -> Result { + let raw_path = uri + .as_str() + .strip_prefix("ipfs://") + .unwrap_or_default() + .trim_matches('/'); + if raw_path.is_empty() { + let message = format!("URI `{uri}` is missing an MFS root directory"); + return Err(StorageResolverError::InvalidUri(message)); + } + Ok(format!("/{raw_path}")) +} + +/// Rejects paths that could escape the storage root (`..`, absolute paths). +fn ensure_valid_relative_path(path: &Path) -> StorageResult<()> { + for component in path.components() { + match component { + std::path::Component::RootDir + | std::path::Component::ParentDir + | std::path::Component::Prefix(_) => { + return Err(StorageErrorKind::Unauthorized.with_error(anyhow::anyhow!( + "path `{}` is forbidden. only simple relative paths are allowed", + path.display() + ))); + } + std::path::Component::CurDir | std::path::Component::Normal(_) => {} + } + } + Ok(()) +} + +/// Wraps a byte stream and enforces that exactly `expected_len` bytes flow +/// through it, failing fast on truncated responses. +struct ExactLenReader { + inner: R, + num_bytes_read: u64, + expected_len: u64, +} + +impl AsyncRead for ExactLenReader { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + let filled_before = buf.filled().len(); + match Pin::new(&mut self.inner).poll_read(cx, buf) { + Poll::Ready(Ok(())) => { + let num_new_bytes = (buf.filled().len() - filled_before) as u64; + if num_new_bytes == 0 && self.num_bytes_read < self.expected_len { + return Poll::Ready(Err(std::io::Error::new( + std::io::ErrorKind::UnexpectedEof, + format!( + "IPFS read truncated: got {} bytes, expected {}", + self.num_bytes_read, self.expected_len + ), + ))); + } + self.num_bytes_read += num_new_bytes; + Poll::Ready(Ok(())) + } + other => other, + } + } +} + +#[async_trait] +impl Storage for IpfsStorage { + async fn check_connectivity(&self) -> anyhow::Result<()> { + self.client.version().await?; + // Creating the root directory doubles as a write-permission check. + self.client.files_mkdir(&self.mfs_root).await?; + Ok(()) + } + + #[instrument(name = "storage.ipfs.put", level = "debug", skip(self, payload), fields(payload_len = payload.len()))] + async fn put(&self, path: &Path, payload: Box) -> StorageResult<()> { + let mfs_path = self.mfs_path(path)?; + // Write to a temporary MFS entry, then move it into place. `files/mv` + // only relinks directory entries, so readers never observe a + // partially written split. + let temp_mfs_path = format!("{mfs_path}.temp"); + self.client + .files_write(&temp_mfs_path, payload.as_ref()) + .await + .map_err(StorageError::from)?; + // `files/mv` fails if the destination exists: remove any previous + // version first. Splits are immutable and ULID-named, so overwrites + // only happen when retrying a failed upload of the same content. + self.client + .files_rm(&mfs_path) + .await + .map_err(StorageError::from)?; + self.client + .files_mv(&temp_mfs_path, &mfs_path) + .await + .map_err(StorageError::from)?; + Ok(()) + } + + #[instrument(name = "storage.ipfs.copy_to", level = "debug", skip(self, output))] + async fn copy_to(&self, path: &Path, output: &mut dyn SendableAsync) -> StorageResult<()> { + let mfs_path = self.mfs_path(path)?; + let expected_len = self + .client + .files_stat(&mfs_path) + .await + .map_err(StorageError::from)? + .size; + let response = self + .client + .files_read_stream(&mfs_path, 0, None) + .await + .map_err(StorageError::from)?; + let byte_stream = response.bytes_stream().map_err(std::io::Error::other); + let mut reader = ExactLenReader { + inner: StreamReader::new(byte_stream), + num_bytes_read: 0, + expected_len, + }; + tokio::io::copy(&mut reader, output).await?; + output.flush().await?; + Ok(()) + } + + #[instrument(name = "storage.ipfs.get_slice", level = "debug", skip(self))] + async fn get_slice(&self, path: &Path, range: Range) -> StorageResult { + let mfs_path = self.mfs_path(path)?; + // Kubo interprets `count=0` on `files/read` as "read to EOF". + if range.is_empty() { + return Ok(OwnedBytes::empty()); + } + let _in_flight_guards = object_storage_get_slice_in_flight_guards(range.len()); + let bytes = self + .client + .files_read_range(&mfs_path, range.start as u64, range.len() as u64) + .await + .map_err(StorageError::from)?; + if bytes.len() != range.len() { + return Err(StorageErrorKind::Io.with_error(anyhow::anyhow!( + "IPFS ranged read returned {} bytes, expected {} (path: {mfs_path})", + bytes.len(), + range.len() + ))); + } + Ok(OwnedBytes::new(bytes)) + } + + #[instrument(name = "storage.ipfs.get_slice_stream", level = "debug", skip(self))] + async fn get_slice_stream( + &self, + path: &Path, + range: Range, + ) -> StorageResult> { + let mfs_path = self.mfs_path(path)?; + // Kubo interprets `count=0` on `files/read` as "read to EOF". + if range.is_empty() { + return Ok(Box::new(tokio::io::empty())); + } + let response = self + .client + .files_read_stream(&mfs_path, range.start as u64, Some(range.len() as u64)) + .await + .map_err(StorageError::from)?; + let byte_stream = response.bytes_stream().map_err(std::io::Error::other); + let reader = ExactLenReader { + inner: StreamReader::new(byte_stream), + num_bytes_read: 0, + expected_len: range.len() as u64, + }; + Ok(Box::new(reader)) + } + + #[instrument(name = "storage.ipfs.get_all", level = "debug", skip(self))] + async fn get_all(&self, path: &Path) -> StorageResult { + let mfs_path = self.mfs_path(path)?; + let num_bytes = self + .client + .files_stat(&mfs_path) + .await + .map_err(StorageError::from)? + .size; + if num_bytes == 0 { + return Ok(OwnedBytes::empty()); + } + let bytes = self + .client + .files_read_range(&mfs_path, 0, num_bytes) + .await + .map_err(StorageError::from)?; + if bytes.len() as u64 != num_bytes { + return Err(StorageErrorKind::Io.with_error(anyhow::anyhow!( + "IPFS full read returned {} bytes, expected {num_bytes} (path: {mfs_path})", + bytes.len(), + ))); + } + Ok(OwnedBytes::new(bytes)) + } + + #[instrument(name = "storage.ipfs.delete", level = "debug", skip(self))] + async fn delete(&self, path: &Path) -> StorageResult<()> { + self.delete_single_file(path).await + } + + #[instrument(name = "storage.ipfs.bulk_delete", level = "debug", skip(self, paths), fields(num_paths = paths.len()))] + async fn bulk_delete<'a>(&self, paths: &[&'a Path]) -> Result<(), BulkDeleteError> { + let mut successes = Vec::with_capacity(paths.len()); + let mut failures = std::collections::HashMap::new(); + let delete_futures: Vec<_> = paths + .iter() + .map(|path: &&Path| { + let path: &Path = path; + async move { + let delete_result = self.delete_single_file(path).await; + (path, delete_result) + } + }) + .collect(); + let mut delete_stream = futures::stream::iter(delete_futures).buffer_unordered(10); + while let Some((path, delete_result)) = delete_stream.next().await { + match delete_result { + Ok(()) => successes.push(path.to_path_buf()), + Err(storage_error) => { + let failure = DeleteFailure { + error: Some(storage_error), + ..Default::default() + }; + failures.insert(path.to_path_buf(), failure); + } + } + } + if failures.is_empty() { + return Ok(()); + } + Err(BulkDeleteError { + successes, + failures, + ..Default::default() + }) + } + + #[instrument(name = "storage.ipfs.file_num_bytes", level = "debug", skip(self))] + async fn file_num_bytes(&self, path: &Path) -> StorageResult { + let mfs_path = self.mfs_path(path)?; + let stat_response = self + .client + .files_stat(&mfs_path) + .await + .map_err(StorageError::from)?; + if stat_response.entry_type != "file" { + return Err(StorageErrorKind::NotFound.with_error(anyhow::anyhow!( + "MFS entry `{mfs_path}` is not a regular file, cannot determine its size" + ))); + } + Ok(stat_response.size) + } + + fn uri(&self) -> &Uri { + &self.uri + } +} + +#[cfg(test)] +mod tests { + use std::str::FromStr; + + use super::*; + + #[test] + fn test_mfs_root_from_uri() { + let uri = Uri::from_str("ipfs://indexes/my-index").unwrap(); + assert_eq!(mfs_root_from_uri(&uri).unwrap(), "/indexes/my-index"); + + let uri = Uri::from_str("ipfs://indexes").unwrap(); + assert_eq!(mfs_root_from_uri(&uri).unwrap(), "/indexes"); + + let uri = Uri::from_str("ipfs://").unwrap(); + mfs_root_from_uri(&uri).unwrap_err(); + } + + #[test] + fn test_ipfs_storage_from_uri() { + let storage_config = IpfsStorageConfig::default(); + let uri = Uri::from_str("ipfs://indexes/my-index").unwrap(); + let ipfs_storage = IpfsStorage::from_uri(&storage_config, &uri).unwrap(); + assert_eq!(ipfs_storage.uri(), &uri); + assert_eq!(ipfs_storage.mfs_root, "/indexes/my-index"); + + let s3_uri = Uri::from_str("s3://bucket/key").unwrap(); + let resolver_error = IpfsStorage::from_uri(&storage_config, &s3_uri).unwrap_err(); + assert!(matches!( + resolver_error, + StorageResolverError::InvalidUri(_) + )); + } + + #[test] + fn test_mfs_path_forbids_parent_dir() { + let storage_config = IpfsStorageConfig::default(); + let uri = Uri::from_str("ipfs://indexes/my-index").unwrap(); + let ipfs_storage = IpfsStorage::from_uri(&storage_config, &uri).unwrap(); + + let mfs_path = ipfs_storage + .mfs_path(Path::new("splits/split-1.split")) + .unwrap(); + assert_eq!(mfs_path, "/indexes/my-index/splits/split-1.split"); + + let path_error = ipfs_storage + .mfs_path(Path::new("../escape.split")) + .unwrap_err(); + assert_eq!(path_error.kind(), StorageErrorKind::Unauthorized); + } +} diff --git a/quickwit/quickwit-storage/src/lib.rs b/quickwit/quickwit-storage/src/lib.rs index 2a6338fc33c..307b5334227 100644 --- a/quickwit/quickwit-storage/src/lib.rs +++ b/quickwit/quickwit-storage/src/lib.rs @@ -41,6 +41,8 @@ pub use self::storage::Storage; mod bundle_storage; mod error; +#[cfg(feature = "ipfs")] +mod ipfs_storage; mod local_file_storage; mod object_storage; #[cfg(feature = "gcs")] @@ -67,6 +69,8 @@ pub use self::cache::{ ByteRangeCache, MemorySizedCache, QuickwitCache, StorageCache, wrap_storage_with_cache, }; pub use self::counting_storage::{CountingStorage, DownloadCounters}; +#[cfg(feature = "ipfs")] +pub use self::ipfs_storage::{IpfsStorage, IpfsStorageFactory}; pub use self::local_file_storage::{LocalFileStorage, LocalFileStorageFactory}; #[cfg(feature = "azure")] pub use self::object_storage::{AzureBlobStorage, AzureBlobStorageFactory}; diff --git a/quickwit/quickwit-storage/src/storage_resolver.rs b/quickwit/quickwit-storage/src/storage_resolver.rs index 3853f763c94..d843a3362e6 100644 --- a/quickwit/quickwit-storage/src/storage_resolver.rs +++ b/quickwit/quickwit-storage/src/storage_resolver.rs @@ -25,6 +25,8 @@ use quickwit_config::{ use crate::AzureBlobStorageFactory; #[cfg(feature = "gcs")] use crate::GoogleCloudStorageFactory; +#[cfg(feature = "ipfs")] +use crate::IpfsStorageFactory; use crate::local_file_storage::LocalFileStorageFactory; use crate::ram_storage::RamStorageFactory; use crate::{S3CompatibleObjectStorageFactory, Storage, StorageFactory, StorageResolverError}; @@ -57,6 +59,7 @@ impl StorageResolver { Protocol::Ram => StorageBackend::Ram, Protocol::S3 => StorageBackend::S3, Protocol::Google => StorageBackend::Google, + Protocol::Ipfs => StorageBackend::Ipfs, _ => { let message = format!( "Quickwit does not support {} as a storage backend", @@ -129,6 +132,21 @@ impl StorageResolver { "Quickwit was compiled without the `gcs` feature", )) } + #[cfg(feature = "ipfs")] + { + builder = builder.register(IpfsStorageFactory::new( + storage_configs.find_ipfs().cloned().unwrap_or_default(), + )); + } + #[cfg(not(feature = "ipfs"))] + { + use crate::storage_factory::UnsupportedStorage; + + builder = builder.register(UnsupportedStorage::new( + StorageBackend::Ipfs, + "Quickwit was compiled without the `ipfs` feature", + )) + } builder .build() .expect("storage factory and config backends should match") diff --git a/quickwit/quickwit-storage/tests/ipfs_storage.rs b/quickwit/quickwit-storage/tests/ipfs_storage.rs new file mode 100644 index 00000000000..851f2f653f3 --- /dev/null +++ b/quickwit/quickwit-storage/tests/ipfs_storage.rs @@ -0,0 +1,72 @@ +// Copyright 2021-Present Datadog, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// This file is an integration test that assumes an IPFS node (Kubo) is +// reachable on its RPC API. Start one with: +// `docker compose up -d kubo` (from the repository root), or point +// `QW_IPFS_API_ENDPOINT` at an existing node. + +#[cfg(all(feature = "integration-testsuite", feature = "ipfs"))] +#[cfg_attr(not(feature = "ci-test"), ignore)] +mod ipfs_storage_test_suite { + use std::str::FromStr; + + use anyhow::Context; + use quickwit_common::rand::append_random_suffix; + use quickwit_common::setup_logging_for_tests; + use quickwit_common::uri::Uri; + use quickwit_config::IpfsStorageConfig; + use quickwit_storage::{IpfsStorage, Storage}; + + #[tokio::test] + async fn ipfs_storage_test_suite() -> anyhow::Result<()> { + setup_logging_for_tests(); + + let root_dir = append_random_suffix("quickwit-integration-tests").to_lowercase(); + let storage_config = IpfsStorageConfig::default(); + let storage_uri = Uri::from_str(&format!("ipfs://{root_dir}"))?; + let mut ipfs_storage = IpfsStorage::from_uri(&storage_config, &storage_uri)?; + ipfs_storage + .check_connectivity() + .await + .context("IPFS node unreachable. start one with `docker compose up -d kubo`")?; + + quickwit_storage::storage_test_suite(&mut ipfs_storage).await?; + + let mut ipfs_storage = IpfsStorage::from_uri( + &storage_config, + &Uri::from_str(&format!("ipfs://{root_dir}/sub/prefix"))?, + )?; + ipfs_storage.check_connectivity().await?; + + quickwit_storage::storage_test_single_part_upload(&mut ipfs_storage) + .await + .context("test single-part upload failed")?; + + quickwit_storage::storage_test_multi_part_upload(&mut ipfs_storage) + .await + .context("test multi-part (large file) upload failed")?; + + // Every stored file must expose a CID: this is the content address + // that makes splits pinnable and shareable across the IPFS network. + let test_path = std::path::Path::new("cid-check.txt"); + ipfs_storage + .put(test_path, Box::new(b"content-addressed".to_vec())) + .await?; + let cid = ipfs_storage.file_cid(test_path).await?; + assert!(!cid.is_empty()); + ipfs_storage.delete(test_path).await?; + Ok(()) + } +}