mirror of
https://github.com/deuxfleurs-org/garage.git
synced 2026-08-13 23:56:53 +00:00
Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Alex Auvolat
|
01701acba1 | ||
|
trinity-1686a
|
5645c82fb8 | ||
|
Quentin Dufour
|
30c3b7a79c |
Generated
+518
-472
File diff suppressed because it is too large
Load Diff
+4
-5
@@ -132,8 +132,8 @@ opentelemetry-contrib = "0.9"
|
||||
prometheus = "0.13"
|
||||
|
||||
# used by the k2v-client crate only
|
||||
aws-sigv4 = { version = "1.1", default-features = false }
|
||||
hyper-rustls = { version = "0.26", default-features = false, features = ["http1", "http2", "ring", "rustls-native-certs"] }
|
||||
aws-sigv4 = { version = "1.1" }
|
||||
hyper-rustls = { version = "0.26", features = ["http2"] }
|
||||
log = "0.4"
|
||||
thiserror = "1.0"
|
||||
|
||||
@@ -141,9 +141,8 @@ thiserror = "1.0"
|
||||
assert-json-diff = "2.0"
|
||||
rustc_version = "0.4.0"
|
||||
static_init = "1.0"
|
||||
aws-smithy-runtime = { version = "1.8", default-features = false, features = ["tls-rustls"] }
|
||||
aws-sdk-config = { version = "1.62", default-features = false }
|
||||
aws-sdk-s3 = { version = "1.79", default-features = false, features = ["rt-tokio"] }
|
||||
aws-sdk-config = "1.62"
|
||||
aws-sdk-s3 = "=1.68"
|
||||
|
||||
[profile.dev]
|
||||
#lto = "thin" # disabled for now, adds 2-4 min to each CI build
|
||||
|
||||
@@ -69,7 +69,7 @@ $CONFIG = array(
|
||||
'hostname' => '127.0.0.1', // Can also be a domain name, eg. garage.example.com
|
||||
'port' => 3900, // Put your reverse proxy port or your S3 API port
|
||||
'use_ssl' => false, // Set it to true if you have a TLS enabled reverse proxy
|
||||
'region' => 'garage', // Garage default region is named "garage", edit according to your cluster config
|
||||
'region' => 'garage', // Garage has only one region named "garage"
|
||||
'use_path_style' => true // Garage supports only path style, must be set to true
|
||||
],
|
||||
],
|
||||
@@ -135,7 +135,7 @@ bucket but doesn't also know the secret encryption key.
|
||||
*Click on the picture to zoom*
|
||||
|
||||
Add a new external storage. Put what you want in "folder name" (eg. "shared"). Select "Amazon S3". Keep "Access Key" for the Authentication field.
|
||||
In Configuration, put your bucket name (eg. nextcloud), the host (eg. 127.0.0.1), the port (eg. 3900 or 443), the region ("garage" if you use the default, or the one your configured in your `garage.toml`). Tick the SSL box if you have put an HTTPS proxy in front of garage. You must tick the "Path access" box and you must leave the "Legacy authentication (v2)" box empty. Put your Key ID (eg. GK...) and your Secret Key in the last two input boxes. Finally click on the tick symbol on the right of your screen.
|
||||
In Configuration, put your bucket name (eg. nextcloud), the host (eg. 127.0.0.1), the port (eg. 3900 or 443), the region (garage). Tick the SSL box if you have put an HTTPS proxy in front of garage. You must tick the "Path access" box and you must leave the "Legacy authentication (v2)" box empty. Put your Key ID (eg. GK...) and your Secret Key in the last two input boxes. Finally click on the tick symbol on the right of your screen.
|
||||
|
||||
Now go to your "Files" app and a new "linked folder" has appeared with the name you chose earlier (eg. "shared").
|
||||
|
||||
@@ -238,7 +238,7 @@ object_storage:
|
||||
# Put localhost only if you have a garage instance running on that node
|
||||
endpoint: 'http://localhost:3900' # or "garage.example.com" if you have TLS on port 443
|
||||
|
||||
# Garage default region is named "garage", edit according to your config
|
||||
# Garage supports only one region for now, named garage
|
||||
region: 'garage'
|
||||
|
||||
credentials:
|
||||
@@ -441,7 +441,7 @@ media_storage_providers:
|
||||
store_synchronous: True # do we want to wait that the file has been written before returning?
|
||||
config:
|
||||
bucket: matrix # the name of our bucket, we chose matrix earlier
|
||||
region_name: garage # "garage" by default, edit according to your cluster config
|
||||
region_name: garage # only "garage" is supported for the region field
|
||||
endpoint_url: http://localhost:3900 # the path to the S3 endpoint
|
||||
access_key_id: "GKxxx" # your Key ID
|
||||
secret_access_key: "xxxx" # your Secret Key
|
||||
|
||||
@@ -86,62 +86,3 @@ helm delete --namespace garage garage
|
||||
```
|
||||
|
||||
Note that this will leave behind custom CRD `garagenodes.deuxfleurs.fr`, which must be removed manually if desired.
|
||||
|
||||
## Increase PVC size on running Garage instances
|
||||
|
||||
Since the Garage Helm chart creates the data and meta PVC based on `StatefulSet` templates, increasing the PVC size can be a bit tricky.
|
||||
|
||||
### Confirm the `StorageClass` used for Garage supports volume expansion
|
||||
|
||||
Confirm the storage class used for garage.
|
||||
|
||||
```bash
|
||||
kubectl -n garage get pvc
|
||||
NAME STATUS VOLUME CAPACITY ACCESS MODES STORAGECLASS VOLUMEATTRIBUTESCLASS AGE
|
||||
data-garage-0 Bound pvc-080360c9-8ce3-4acf-8579-1701e57b7f3f 30Gi RWO longhorn-local <unset> 77d
|
||||
data-garage-1 Bound pvc-ab8ba697-6030-4fc7-ab3c-0d6df9e3dbc0 30Gi RWO longhorn-local <unset> 5d8h
|
||||
data-garage-2 Bound pvc-3ab37551-0231-4604-986d-136d0fd950ec 30Gi RWO longhorn-local <unset> 5d5h
|
||||
meta-garage-0 Bound pvc-3b457302-3023-4169-846e-c928c5f2ea65 3Gi RWO longhorn-local <unset> 77d
|
||||
meta-garage-1 Bound pvc-49ace2b9-5c85-42df-9247-51c4cf64b460 3Gi RWO longhorn-local <unset> 5d8h
|
||||
meta-garage-2 Bound pvc-99e2e50f-42b4-4128-ae2f-b52629259723 3Gi RWO longhorn-local <unset> 5d5h
|
||||
```
|
||||
|
||||
In this case, the storage class is `longhorn-local`. Now, check if `ALLOWVOLUMEEXPANSION` is true for the used `StorageClass`.
|
||||
|
||||
```bash
|
||||
kubectl get storageclasses.storage.k8s.io longhorn-local
|
||||
NAME PROVISIONER RECLAIMPOLICY VOLUMEBINDINGMODE ALLOWVOLUMEEXPANSION AGE
|
||||
longhorn-local driver.longhorn.io Delete Immediate true 103d
|
||||
```
|
||||
|
||||
If your `StorageClass` does not support volume expansion, double check if you can enable it. Otherwise, your only real option is to spin up a new Garage cluster with increased size and migrate all data over.
|
||||
|
||||
If your `StorageClass` supports expansion, you are free to continue.
|
||||
|
||||
### Increase the size of the PVCs
|
||||
|
||||
Increase the size of all PVCs to your desired size.
|
||||
|
||||
```bash
|
||||
kubectl -n garage edit pvc data-garage-0
|
||||
kubectl -n garage edit pvc data-garage-1
|
||||
kubectl -n garage edit pvc data-garage-2
|
||||
kubectl -n garage edit pvc meta-garage-0
|
||||
kubectl -n garage edit pvc meta-garage-1
|
||||
kubectl -n garage edit pvc meta-garage-2
|
||||
```
|
||||
|
||||
### Increase the size of the `StatefulSet` PVC template
|
||||
|
||||
This is an optional step, but if not done, future instances of Garage will be created with the original size from the template.
|
||||
|
||||
```bash
|
||||
kubectl -n garage delete sts --cascade=orphan garage
|
||||
statefulset.apps "garage" deleted
|
||||
```
|
||||
|
||||
This will remove the Garage `StatefulSet` but leave the pods running. It may seem destructive but needs to be done this way since edits to the size of PVC templates are prohibited.
|
||||
|
||||
### Redeploy the `StatefulSet`
|
||||
|
||||
Now the size of future PVCs can be increased, and the Garage Helm chart can be upgraded. The new `StatefulSet` should take ownership of the orphaned pods again.
|
||||
|
||||
@@ -129,9 +129,9 @@ docker run \
|
||||
-d \
|
||||
--name garaged \
|
||||
-p 3900:3900 -p 3901:3901 -p 3902:3902 -p 3903:3903 \
|
||||
-v /path/to/garage.toml:/etc/garage.toml \
|
||||
-v /path/to/garage/meta:/var/lib/garage/meta \
|
||||
-v /path/to/garage/data:/var/lib/garage/data \
|
||||
-v /etc/garage.toml:/path/to/garage.toml \
|
||||
-v /var/lib/garage/meta:/path/to/garage/meta \
|
||||
-v /var/lib/garage/data:/path/to/garage/data \
|
||||
dxflrs/garage:v1.1.0
|
||||
```
|
||||
|
||||
|
||||
@@ -1,18 +1,24 @@
|
||||
apiVersion: v2
|
||||
name: garage
|
||||
description: S3-compatible object store for small self-hosted geo-distributed deployments
|
||||
|
||||
# A chart can be either an 'application' or a 'library' chart.
|
||||
#
|
||||
# Application charts are a collection of templates that can be packaged into versioned archives
|
||||
# to be deployed.
|
||||
#
|
||||
# Library charts provide useful utilities or functions for the chart developer. They're included as
|
||||
# a dependency of application charts to inject those utilities and functions into the rendering
|
||||
# pipeline. Library charts do not define any templates and therefore cannot be deployed.
|
||||
type: application
|
||||
|
||||
# This is the chart version. This version number should be incremented each time you make changes
|
||||
# to the chart and its templates, including the app version.
|
||||
# Versions are expected to follow Semantic Versioning (https://semver.org/)
|
||||
version: 0.7.0
|
||||
|
||||
# This is the version number of the application being deployed. This version number should be
|
||||
# incremented each time you make changes to the application. Versions are not expected to
|
||||
# follow Semantic Versioning. They should reflect the version the application is using.
|
||||
# It is recommended to use it with quotes.
|
||||
appVersion: "v1.1.0"
|
||||
home: https://garagehq.deuxfleurs.fr/
|
||||
icon: https://garagehq.deuxfleurs.fr/images/garage-logo.svg
|
||||
|
||||
keywords:
|
||||
- geo-distributed
|
||||
- read-after-write-consistency
|
||||
- s3-compatible
|
||||
|
||||
sources:
|
||||
- https://git.deuxfleurs.fr/Deuxfleurs/garage.git
|
||||
|
||||
maintainers: []
|
||||
@@ -1,15 +1,9 @@
|
||||
# garage
|
||||
|
||||
  
|
||||
  
|
||||
|
||||
S3-compatible object store for small self-hosted geo-distributed deployments
|
||||
|
||||
**Homepage:** <https://garagehq.deuxfleurs.fr/>
|
||||
|
||||
## Source Code
|
||||
|
||||
* <https://git.deuxfleurs.fr/Deuxfleurs/garage.git>
|
||||
|
||||
## Values
|
||||
|
||||
| Key | Type | Default | Description |
|
||||
@@ -29,7 +23,6 @@ S3-compatible object store for small self-hosted geo-distributed deployments
|
||||
| garage.existingConfigMap | string | `""` | if not empty string, allow using an existing ConfigMap for the garage.toml, if set, ignores garage.toml |
|
||||
| garage.garageTomlString | string | `""` | String Template for the garage configuration if set, ignores above values. Values can be templated, see https://garagehq.deuxfleurs.fr/documentation/reference-manual/configuration/ |
|
||||
| garage.kubernetesSkipCrd | bool | `false` | Set to true if you want to use k8s discovery but install the CRDs manually outside of the helm chart, for example if you operate at namespace level without cluster ressources |
|
||||
| garage.metadataAutoSnapshotInterval | string | `""` | If this value is set, Garage will automatically take a snapshot of the metadata DB file at a regular interval and save it in the metadata directory. https://garagehq.deuxfleurs.fr/documentation/reference-manual/configuration/#metadata_auto_snapshot_interval |
|
||||
| garage.replicationMode | string | `"3"` | Default to 3 replicas, see the replication_mode section at https://garagehq.deuxfleurs.fr/documentation/reference-manual/configuration/#replication-mode |
|
||||
| garage.rpcBindAddr | string | `"[::]:3901"` | |
|
||||
| garage.rpcSecret | string | `""` | If not given, a random secret will be generated and stored in a Secret object |
|
||||
@@ -56,7 +49,6 @@ S3-compatible object store for small self-hosted geo-distributed deployments
|
||||
| initImage.pullPolicy | string | `"IfNotPresent"` | |
|
||||
| initImage.repository | string | `"busybox"` | |
|
||||
| initImage.tag | string | `"stable"` | |
|
||||
| livenessProbe | object | `{}` | Specifies a livenessProbe |
|
||||
| monitoring.metrics.enabled | bool | `false` | If true, a service for monitoring is created with a prometheus.io/scrape annotation |
|
||||
| monitoring.metrics.serviceMonitor.enabled | bool | `false` | If true, a ServiceMonitor CRD is created for a prometheus operator https://github.com/coreos/prometheus-operator |
|
||||
| monitoring.metrics.serviceMonitor.interval | string | `"15s"` | |
|
||||
@@ -79,7 +71,6 @@ S3-compatible object store for small self-hosted geo-distributed deployments
|
||||
| podSecurityContext.runAsGroup | int | `1000` | |
|
||||
| podSecurityContext.runAsNonRoot | bool | `true` | |
|
||||
| podSecurityContext.runAsUser | int | `1000` | |
|
||||
| readinessProbe | object | `{}` | Specifies a readinessProbe |
|
||||
| resources | object | `{}` | |
|
||||
| securityContext.capabilities | object | `{"drop":["ALL"]}` | The default security context is heavily restricted, feel free to tune it to your requirements |
|
||||
| securityContext.readOnlyRootFilesystem | bool | `true` | |
|
||||
|
||||
@@ -19,10 +19,6 @@ data:
|
||||
|
||||
compression_level = {{ .Values.garage.compressionLevel }}
|
||||
|
||||
{{- if .Values.garage.metadataAutoSnapshotInterval }}
|
||||
metadata_auto_snapshot_interval = {{ .Values.garage.metadataAutoSnapshotInterval | quote }}
|
||||
{{- end }}
|
||||
|
||||
rpc_bind_addr = "{{ .Values.garage.rpcBindAddr }}"
|
||||
# rpc_secret will be populated by the init container from a k8s secret object
|
||||
rpc_secret = "__RPC_SECRET_REPLACE__"
|
||||
|
||||
@@ -78,14 +78,15 @@ spec:
|
||||
{{- with .Values.extraVolumeMounts }}
|
||||
{{- toYaml . | nindent 12 }}
|
||||
{{- end }}
|
||||
{{- with .Values.livenessProbe }}
|
||||
livenessProbe:
|
||||
{{- toYaml . | nindent 12 }}
|
||||
{{- end }}
|
||||
{{- with .Values.readinessProbe }}
|
||||
readinessProbe:
|
||||
{{- toYaml . | nindent 12 }}
|
||||
{{- end }}
|
||||
# TODO
|
||||
# livenessProbe:
|
||||
# httpGet:
|
||||
# path: /
|
||||
# port: 3900
|
||||
# readinessProbe:
|
||||
# httpGet:
|
||||
# path: /
|
||||
# port: 3900
|
||||
resources:
|
||||
{{- toYaml .Values.resources | nindent 12 }}
|
||||
volumes:
|
||||
|
||||
@@ -21,10 +21,6 @@ garage:
|
||||
# https://garagehq.deuxfleurs.fr/documentation/reference-manual/configuration/#compression-level
|
||||
compressionLevel: "1"
|
||||
|
||||
# -- If this value is set, Garage will automatically take a snapshot of the metadata DB file at a regular interval and save it in the metadata directory.
|
||||
# https://garagehq.deuxfleurs.fr/documentation/reference-manual/configuration/#metadata_auto_snapshot_interval
|
||||
metadataAutoSnapshotInterval: ""
|
||||
|
||||
rpcBindAddr: "[::]:3901"
|
||||
# -- If not given, a random secret will be generated and stored in a Secret object
|
||||
rpcSecret: ""
|
||||
@@ -195,21 +191,6 @@ resources: {}
|
||||
# cpu: 100m
|
||||
# memory: 512Mi
|
||||
|
||||
# -- Specifies a livenessProbe
|
||||
livenessProbe: {}
|
||||
#httpGet:
|
||||
# path: /health
|
||||
# port: 3903
|
||||
#initialDelaySeconds: 5
|
||||
#periodSeconds: 30
|
||||
# -- Specifies a readinessProbe
|
||||
readinessProbe: {}
|
||||
#httpGet:
|
||||
# path: /health
|
||||
# port: 3903
|
||||
#initialDelaySeconds: 5
|
||||
#periodSeconds: 30
|
||||
|
||||
nodeSelector: {}
|
||||
|
||||
tolerations: []
|
||||
|
||||
@@ -419,11 +419,17 @@ pub async fn handle_update_bucket(
|
||||
|
||||
if let Some(wa) = req.website_access {
|
||||
if wa.enabled {
|
||||
let (redirect_all, routing_rules) = match state.website_config.get() {
|
||||
Some(wc) => (wc.redirect_all.clone(), wc.routing_rules.clone()),
|
||||
None => (None, Vec::new()),
|
||||
};
|
||||
state.website_config.update(Some(WebsiteConfig {
|
||||
index_document: wa.index_document.ok_or_bad_request(
|
||||
"Please specify indexDocument when enabling website access.",
|
||||
)?,
|
||||
error_document: wa.error_document,
|
||||
redirect_all,
|
||||
routing_rules,
|
||||
}));
|
||||
} else {
|
||||
if wa.index_document.is_some() || wa.error_document.is_some() {
|
||||
|
||||
+172
-54
@@ -3,7 +3,7 @@ use quick_xml::de::from_reader;
|
||||
use hyper::{header::HeaderName, Request, Response, StatusCode};
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
use garage_model::bucket_table::*;
|
||||
use garage_model::bucket_table::{self, *};
|
||||
|
||||
use garage_api_common::helpers::*;
|
||||
|
||||
@@ -26,7 +26,28 @@ pub async fn handle_get_website(ctx: ReqCtx) -> Result<Response<ResBody>, Error>
|
||||
suffix: Value(website.index_document.to_string()),
|
||||
}),
|
||||
redirect_all_requests_to: None,
|
||||
routing_rules: None,
|
||||
routing_rules: RoutingRules {
|
||||
rules: website
|
||||
.routing_rules
|
||||
.clone()
|
||||
.into_iter()
|
||||
.map(|rule| RoutingRule {
|
||||
condition: rule.condition.map(|cond| Condition {
|
||||
http_error_code: cond.http_error_code.map(|c| IntValue(c as i64)),
|
||||
prefix: cond.prefix.map(Value),
|
||||
}),
|
||||
redirect: Redirect {
|
||||
hostname: rule.redirect.hostname.map(Value),
|
||||
http_redirect_code: Some(IntValue(
|
||||
rule.redirect.http_redirect_code as i64,
|
||||
)),
|
||||
protocol: rule.redirect.protocol.map(Value),
|
||||
replace_full: rule.redirect.replace_key.map(Value),
|
||||
replace_prefix: rule.redirect.replace_key_prefix.map(Value),
|
||||
},
|
||||
})
|
||||
.collect(),
|
||||
},
|
||||
};
|
||||
let xml = to_xml_with_header(&wc)?;
|
||||
Ok(Response::builder()
|
||||
@@ -97,18 +118,28 @@ pub struct WebsiteConfiguration {
|
||||
pub index_document: Option<Suffix>,
|
||||
#[serde(rename = "RedirectAllRequestsTo")]
|
||||
pub redirect_all_requests_to: Option<Target>,
|
||||
#[serde(rename = "RoutingRules")]
|
||||
pub routing_rules: Option<Vec<RoutingRule>>,
|
||||
#[serde(
|
||||
rename = "RoutingRules",
|
||||
default,
|
||||
skip_serializing_if = "RoutingRules::is_empty"
|
||||
)]
|
||||
pub routing_rules: RoutingRules,
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord, Default)]
|
||||
pub struct RoutingRules {
|
||||
#[serde(rename = "RoutingRule")]
|
||||
pub rules: Vec<RoutingRule>,
|
||||
}
|
||||
|
||||
impl RoutingRules {
|
||||
fn is_empty(&self) -> bool {
|
||||
self.rules.is_empty()
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
|
||||
pub struct RoutingRule {
|
||||
#[serde(rename = "RoutingRule")]
|
||||
pub inner: RoutingRuleInner,
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
|
||||
pub struct RoutingRuleInner {
|
||||
#[serde(rename = "Condition")]
|
||||
pub condition: Option<Condition>,
|
||||
#[serde(rename = "Redirect")]
|
||||
@@ -162,7 +193,7 @@ impl WebsiteConfiguration {
|
||||
if self.redirect_all_requests_to.is_some()
|
||||
&& (self.error_document.is_some()
|
||||
|| self.index_document.is_some()
|
||||
|| self.routing_rules.is_some())
|
||||
|| !self.routing_rules.is_empty())
|
||||
{
|
||||
return Err(Error::bad_request(
|
||||
"Bad XML: can't have RedirectAllRequestsTo and other fields",
|
||||
@@ -177,10 +208,15 @@ impl WebsiteConfiguration {
|
||||
if let Some(ref rart) = self.redirect_all_requests_to {
|
||||
rart.validate()?;
|
||||
}
|
||||
if let Some(ref rrs) = self.routing_rules {
|
||||
for rr in rrs {
|
||||
rr.inner.validate()?;
|
||||
}
|
||||
for rr in &self.routing_rules.rules {
|
||||
rr.validate()?;
|
||||
}
|
||||
if self.routing_rules.rules.len() > 1000 {
|
||||
// we will do linear scans, best to avoid overly long configuration. The
|
||||
// limit was choosen arbitrarily
|
||||
return Err(Error::bad_request(
|
||||
"Bad XML: RoutingRules can't have more than 1000 child elements",
|
||||
));
|
||||
}
|
||||
|
||||
Ok(())
|
||||
@@ -189,11 +225,7 @@ impl WebsiteConfiguration {
|
||||
pub fn into_garage_website_config(self) -> Result<WebsiteConfig, Error> {
|
||||
if self.redirect_all_requests_to.is_some() {
|
||||
Err(Error::NotImplemented(
|
||||
"S3 website redirects are not currently implemented in Garage.".into(),
|
||||
))
|
||||
} else if self.routing_rules.map(|x| !x.is_empty()).unwrap_or(false) {
|
||||
Err(Error::NotImplemented(
|
||||
"S3 routing rules are not currently implemented in Garage.".into(),
|
||||
"RedirectAllRequestsTo is not currently implemented in Garage, however its effect can be emulated using a single inconditional RoutingRule.".into(),
|
||||
))
|
||||
} else {
|
||||
Ok(WebsiteConfig {
|
||||
@@ -202,6 +234,36 @@ impl WebsiteConfiguration {
|
||||
.map(|x| x.suffix.0)
|
||||
.unwrap_or_else(|| "index.html".to_string()),
|
||||
error_document: self.error_document.map(|x| x.key.0),
|
||||
redirect_all: None,
|
||||
routing_rules: self
|
||||
.routing_rules
|
||||
.rules
|
||||
.into_iter()
|
||||
.map(|rule| {
|
||||
bucket_table::RoutingRule {
|
||||
condition: rule.condition.map(|condition| {
|
||||
bucket_table::RedirectCondition {
|
||||
http_error_code: condition.http_error_code.map(|c| c.0 as u16),
|
||||
prefix: condition.prefix.map(|p| p.0),
|
||||
}
|
||||
}),
|
||||
redirect: bucket_table::Redirect {
|
||||
hostname: rule.redirect.hostname.map(|h| h.0),
|
||||
protocol: rule.redirect.protocol.map(|p| p.0),
|
||||
// aws default to 301, which i find punitive in case of
|
||||
// missconfiguration (can be permanently cached on the
|
||||
// user agent)
|
||||
http_redirect_code: rule
|
||||
.redirect
|
||||
.http_redirect_code
|
||||
.map(|c| c.0 as u16)
|
||||
.unwrap_or(302),
|
||||
replace_key_prefix: rule.redirect.replace_prefix.map(|k| k.0),
|
||||
replace_key: rule.redirect.replace_full.map(|k| k.0),
|
||||
},
|
||||
}
|
||||
})
|
||||
.collect(),
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -242,37 +304,69 @@ impl Target {
|
||||
}
|
||||
}
|
||||
|
||||
impl RoutingRuleInner {
|
||||
impl RoutingRule {
|
||||
pub fn validate(&self) -> Result<(), Error> {
|
||||
let has_prefix = self
|
||||
.condition
|
||||
.as_ref()
|
||||
.and_then(|c| c.prefix.as_ref())
|
||||
.is_some();
|
||||
self.redirect.validate(has_prefix)
|
||||
if let Some(condition) = &self.condition {
|
||||
condition.validate()?;
|
||||
}
|
||||
self.redirect.validate()
|
||||
}
|
||||
}
|
||||
|
||||
impl Condition {
|
||||
pub fn validate(&self) -> Result<bool, Error> {
|
||||
if let Some(ref error_code) = self.http_error_code {
|
||||
// TODO do other error codes make sense? Aws only allows 4xx and 5xx
|
||||
if error_code.0 != 404 {
|
||||
return Err(Error::bad_request(
|
||||
"Bad XML: HttpErrorCodeReturnedEquals must be 404 or absent",
|
||||
));
|
||||
}
|
||||
}
|
||||
Ok(self.prefix.is_some())
|
||||
}
|
||||
}
|
||||
|
||||
impl Redirect {
|
||||
pub fn validate(&self, has_prefix: bool) -> Result<(), Error> {
|
||||
if self.replace_prefix.is_some() {
|
||||
if self.replace_full.is_some() {
|
||||
return Err(Error::bad_request(
|
||||
"Bad XML: both ReplaceKeyPrefixWith and ReplaceKeyWith are set",
|
||||
));
|
||||
}
|
||||
if !has_prefix {
|
||||
return Err(Error::bad_request(
|
||||
"Bad XML: ReplaceKeyPrefixWith is set, but KeyPrefixEquals isn't",
|
||||
));
|
||||
}
|
||||
pub fn validate(&self) -> Result<(), Error> {
|
||||
if self.replace_prefix.is_some() && self.replace_full.is_some() {
|
||||
return Err(Error::bad_request(
|
||||
"Bad XML: both ReplaceKeyPrefixWith and ReplaceKeyWith are set",
|
||||
));
|
||||
}
|
||||
if let Some(ref protocol) = self.protocol {
|
||||
if protocol.0 != "http" && protocol.0 != "https" {
|
||||
return Err(Error::bad_request("Bad XML: invalid protocol"));
|
||||
}
|
||||
}
|
||||
// TODO there are probably more invalid cases, but which ones?
|
||||
if let Some(ref http_redirect_code) = self.http_redirect_code {
|
||||
match http_redirect_code.0 {
|
||||
// aws allows all 3xx except 300, but some are non-sensical (not modified,
|
||||
// use proxy...)
|
||||
301 | 302 | 303 | 307 | 308 => {
|
||||
if self.hostname.is_none() && self.protocol.is_some() {
|
||||
return Err(Error::bad_request(
|
||||
"Bad XML: HostName must be set if Protocol is set",
|
||||
));
|
||||
}
|
||||
}
|
||||
// aws doesn't allow these codes, but netlify does, and it seems like a
|
||||
// cool feature (change the page seen without changing the url shown by the
|
||||
// user agent)
|
||||
200 | 404 => {
|
||||
if self.hostname.is_some() || self.protocol.is_some() {
|
||||
// hostname would mean different bucket, protocol doesn't make
|
||||
// sense
|
||||
return Err(Error::bad_request(
|
||||
"Bad XML: an HttpRedirectCode of 200 is not acceptable alongside HostName or Protocol",
|
||||
));
|
||||
}
|
||||
}
|
||||
_ => {
|
||||
return Err(Error::bad_request("Bad XML: invalid HttpRedirectCode"));
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
@@ -311,6 +405,15 @@ mod tests {
|
||||
<ReplaceKeyWith>fullkey</ReplaceKeyWith>
|
||||
</Redirect>
|
||||
</RoutingRule>
|
||||
<RoutingRule>
|
||||
<Condition>
|
||||
<KeyPrefixEquals></KeyPrefixEquals>
|
||||
</Condition>
|
||||
<Redirect>
|
||||
<HttpRedirectCode>404</HttpRedirectCode>
|
||||
<ReplaceKeyWith>missing</ReplaceKeyWith>
|
||||
</Redirect>
|
||||
</RoutingRule>
|
||||
</RoutingRules>
|
||||
</WebsiteConfiguration>"#;
|
||||
let conf: WebsiteConfiguration = from_str(message).unwrap();
|
||||
@@ -326,21 +429,36 @@ mod tests {
|
||||
hostname: Value("garage.tld".to_owned()),
|
||||
protocol: Some(Value("https".to_owned())),
|
||||
}),
|
||||
routing_rules: Some(vec![RoutingRule {
|
||||
inner: RoutingRuleInner {
|
||||
condition: Some(Condition {
|
||||
http_error_code: Some(IntValue(404)),
|
||||
prefix: Some(Value("prefix1".to_owned())),
|
||||
}),
|
||||
redirect: Redirect {
|
||||
hostname: Some(Value("gara.ge".to_owned())),
|
||||
protocol: Some(Value("http".to_owned())),
|
||||
http_redirect_code: Some(IntValue(303)),
|
||||
replace_prefix: Some(Value("prefix2".to_owned())),
|
||||
replace_full: Some(Value("fullkey".to_owned())),
|
||||
routing_rules: RoutingRules {
|
||||
rules: vec![
|
||||
RoutingRule {
|
||||
condition: Some(Condition {
|
||||
http_error_code: Some(IntValue(404)),
|
||||
prefix: Some(Value("prefix1".to_owned())),
|
||||
}),
|
||||
redirect: Redirect {
|
||||
hostname: Some(Value("gara.ge".to_owned())),
|
||||
protocol: Some(Value("http".to_owned())),
|
||||
http_redirect_code: Some(IntValue(303)),
|
||||
replace_prefix: Some(Value("prefix2".to_owned())),
|
||||
replace_full: Some(Value("fullkey".to_owned())),
|
||||
},
|
||||
},
|
||||
},
|
||||
}]),
|
||||
RoutingRule {
|
||||
condition: Some(Condition {
|
||||
http_error_code: None,
|
||||
prefix: Some(Value("".to_owned())),
|
||||
}),
|
||||
redirect: Redirect {
|
||||
hostname: None,
|
||||
protocol: None,
|
||||
http_redirect_code: Some(IntValue(404)),
|
||||
replace_prefix: None,
|
||||
replace_full: Some(Value("missing".to_owned())),
|
||||
},
|
||||
},
|
||||
],
|
||||
},
|
||||
};
|
||||
assert_eq! {
|
||||
ref_value,
|
||||
|
||||
@@ -17,7 +17,6 @@ use opentelemetry::{
|
||||
Context,
|
||||
};
|
||||
|
||||
use garage_net::endpoint::RpcInFlightLimiter;
|
||||
use garage_net::stream::{read_stream_to_end, stream_asyncread, ByteStream};
|
||||
|
||||
use garage_db as db;
|
||||
@@ -296,7 +295,6 @@ impl BlockManager {
|
||||
&node_id,
|
||||
BlockRpc::GetBlock(*hash, order_tag),
|
||||
priority,
|
||||
RpcInFlightLimiter::TableWrite,
|
||||
);
|
||||
tokio::select! {
|
||||
res = rpc => {
|
||||
|
||||
@@ -109,7 +109,7 @@ impl IDb for LmdbDb {
|
||||
let mut path = to.clone();
|
||||
path.push("data.mdb");
|
||||
self.db
|
||||
.copy_to_path(path, heed::CompactionOption::Enabled)?;
|
||||
.copy_to_path(path, heed::CompactionOption::Disabled)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
@@ -62,7 +62,6 @@ syslog-tracing = { workspace = true, optional = true }
|
||||
garage_api_common.workspace = true
|
||||
|
||||
aws-sdk-s3.workspace = true
|
||||
aws-smithy-runtime.workspace = true
|
||||
chrono.workspace = true
|
||||
http.workspace = true
|
||||
hmac.workspace = true
|
||||
|
||||
@@ -390,9 +390,15 @@ impl AdminRpcHandler {
|
||||
}
|
||||
|
||||
let website = if query.allow {
|
||||
let (redirect_all, routing_rules) = match bucket_state.website_config.get() {
|
||||
Some(wc) => (wc.redirect_all.clone(), wc.routing_rules.clone()),
|
||||
None => (None, Vec::new()),
|
||||
};
|
||||
Some(WebsiteConfig {
|
||||
index_document: query.index_document.clone(),
|
||||
error_document: query.error_document.clone(),
|
||||
redirect_all,
|
||||
routing_rules,
|
||||
})
|
||||
} else {
|
||||
None
|
||||
|
||||
+1
-12
@@ -13,8 +13,6 @@ use serde::{Deserialize, Serialize};
|
||||
|
||||
use format_table::format_table_to_string;
|
||||
|
||||
use garage_net::endpoint::RpcInFlightLimiter;
|
||||
|
||||
use garage_util::background::BackgroundRunner;
|
||||
use garage_util::data::*;
|
||||
use garage_util::error::Error as GarageError;
|
||||
@@ -120,7 +118,6 @@ impl AdminRpcHandler {
|
||||
&node,
|
||||
AdminRpc::LaunchRepair(opt_to_send.clone()),
|
||||
PRIO_NORMAL,
|
||||
RpcInFlightLimiter::NoLimit,
|
||||
)
|
||||
.await;
|
||||
if !matches!(resp, Ok(Ok(_))) {
|
||||
@@ -167,12 +164,7 @@ impl AdminRpcHandler {
|
||||
let node_id = (*node).into();
|
||||
match self
|
||||
.endpoint
|
||||
.call(
|
||||
&node_id,
|
||||
AdminRpc::Stats(opt),
|
||||
PRIO_NORMAL,
|
||||
RpcInFlightLimiter::NoLimit,
|
||||
)
|
||||
.call(&node_id, AdminRpc::Stats(opt), PRIO_NORMAL)
|
||||
.await
|
||||
{
|
||||
Ok(Ok(AdminRpc::Ok(s))) => writeln!(&mut ret, "{}", s).unwrap(),
|
||||
@@ -415,7 +407,6 @@ impl AdminRpcHandler {
|
||||
variable: variable.clone(),
|
||||
}),
|
||||
PRIO_NORMAL,
|
||||
RpcInFlightLimiter::NoLimit,
|
||||
)
|
||||
.await??
|
||||
{
|
||||
@@ -465,7 +456,6 @@ impl AdminRpcHandler {
|
||||
value: value.to_string(),
|
||||
}),
|
||||
PRIO_NORMAL,
|
||||
RpcInFlightLimiter::NoLimit,
|
||||
)
|
||||
.await??
|
||||
{
|
||||
@@ -498,7 +488,6 @@ impl AdminRpcHandler {
|
||||
&to,
|
||||
AdminRpc::MetaOperation(MetaOperation::Snapshot { all: false }),
|
||||
PRIO_NORMAL,
|
||||
RpcInFlightLimiter::NoLimit,
|
||||
)
|
||||
.await?
|
||||
}))
|
||||
|
||||
+3
-17
@@ -2,7 +2,6 @@ use std::collections::{HashMap, HashSet};
|
||||
use std::time::Duration;
|
||||
|
||||
use format_table::format_table;
|
||||
use garage_net::endpoint::RpcInFlightLimiter;
|
||||
use garage_util::error::*;
|
||||
|
||||
use garage_rpc::layout::*;
|
||||
@@ -201,12 +200,7 @@ pub async fn cmd_connect(
|
||||
args: ConnectNodeOpt,
|
||||
) -> Result<(), Error> {
|
||||
match rpc_cli
|
||||
.call(
|
||||
&rpc_host,
|
||||
SystemRpc::Connect(args.node),
|
||||
PRIO_NORMAL,
|
||||
RpcInFlightLimiter::NoLimit,
|
||||
)
|
||||
.call(&rpc_host, SystemRpc::Connect(args.node), PRIO_NORMAL)
|
||||
.await??
|
||||
{
|
||||
SystemRpc::Ok => {
|
||||
@@ -222,10 +216,7 @@ pub async fn cmd_admin(
|
||||
rpc_host: NodeID,
|
||||
args: AdminRpc,
|
||||
) -> Result<(), HelperError> {
|
||||
match rpc_cli
|
||||
.call(&rpc_host, args, PRIO_NORMAL, RpcInFlightLimiter::NoLimit)
|
||||
.await??
|
||||
{
|
||||
match rpc_cli.call(&rpc_host, args, PRIO_NORMAL).await?? {
|
||||
AdminRpc::Ok(msg) => {
|
||||
println!("{}", msg);
|
||||
}
|
||||
@@ -280,12 +271,7 @@ pub async fn fetch_status(
|
||||
rpc_host: NodeID,
|
||||
) -> Result<Vec<KnownNodeInfo>, Error> {
|
||||
match rpc_cli
|
||||
.call(
|
||||
&rpc_host,
|
||||
SystemRpc::GetKnownNodes,
|
||||
PRIO_NORMAL,
|
||||
RpcInFlightLimiter::NoLimit,
|
||||
)
|
||||
.call(&rpc_host, SystemRpc::GetKnownNodes, PRIO_NORMAL)
|
||||
.await??
|
||||
{
|
||||
SystemRpc::ReturnKnownNodes(nodes) => Ok(nodes),
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
use bytesize::ByteSize;
|
||||
|
||||
use format_table::format_table;
|
||||
use garage_net::endpoint::RpcInFlightLimiter;
|
||||
use garage_util::crdt::Crdt;
|
||||
use garage_util::error::*;
|
||||
|
||||
@@ -46,12 +45,7 @@ pub async fn cmd_assign_role(
|
||||
args: AssignRoleOpt,
|
||||
) -> Result<(), Error> {
|
||||
let status = match rpc_cli
|
||||
.call(
|
||||
&rpc_host,
|
||||
SystemRpc::GetKnownNodes,
|
||||
PRIO_NORMAL,
|
||||
RpcInFlightLimiter::NoLimit,
|
||||
)
|
||||
.call(&rpc_host, SystemRpc::GetKnownNodes, PRIO_NORMAL)
|
||||
.await??
|
||||
{
|
||||
SystemRpc::ReturnKnownNodes(nodes) => nodes,
|
||||
@@ -481,12 +475,7 @@ pub async fn fetch_layout(
|
||||
rpc_host: NodeID,
|
||||
) -> Result<LayoutHistory, Error> {
|
||||
match rpc_cli
|
||||
.call(
|
||||
&rpc_host,
|
||||
SystemRpc::PullClusterLayout,
|
||||
PRIO_NORMAL,
|
||||
RpcInFlightLimiter::NoLimit,
|
||||
)
|
||||
.call(&rpc_host, SystemRpc::PullClusterLayout, PRIO_NORMAL)
|
||||
.await??
|
||||
{
|
||||
SystemRpc::AdvertiseClusterLayout(t) => Ok(t),
|
||||
@@ -504,7 +493,6 @@ pub async fn send_layout(
|
||||
&rpc_host,
|
||||
SystemRpc::AdvertiseClusterLayout(layout),
|
||||
PRIO_NORMAL,
|
||||
RpcInFlightLimiter::NoLimit,
|
||||
)
|
||||
.await??;
|
||||
Ok(())
|
||||
|
||||
+1
-1
@@ -244,7 +244,7 @@ async fn cli_command(opt: Opt) -> Result<(), Error> {
|
||||
// Generate a temporary keypair for our RPC client
|
||||
let (_pk, sk) = sodiumoxide::crypto::sign::ed25519::gen_keypair();
|
||||
|
||||
let netapp = NetApp::new(GARAGE_VERSION_TAG, network_key, sk, None, None);
|
||||
let netapp = NetApp::new(GARAGE_VERSION_TAG, network_key, sk, None);
|
||||
|
||||
// Find and parse the address of the target host
|
||||
let (id, addr, is_default_addr) = if let Some(h) = opt.rpc_host {
|
||||
|
||||
@@ -5,7 +5,10 @@ use crate::json_body;
|
||||
use assert_json_diff::assert_json_eq;
|
||||
use aws_sdk_s3::{
|
||||
primitives::ByteStream,
|
||||
types::{CorsConfiguration, CorsRule, ErrorDocument, IndexDocument, WebsiteConfiguration},
|
||||
types::{
|
||||
Condition, CorsConfiguration, CorsRule, ErrorDocument, IndexDocument, Protocol, Redirect,
|
||||
RoutingRule, WebsiteConfiguration,
|
||||
},
|
||||
};
|
||||
use http::{Request, StatusCode};
|
||||
use http_body_util::BodyExt;
|
||||
@@ -533,3 +536,444 @@ async fn test_website_check_domain() {
|
||||
})
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_website_redirect_full_bucket() {
|
||||
const BCKT_NAME: &str = "my-redirect-full";
|
||||
let ctx = common::context();
|
||||
let bucket = ctx.create_bucket(BCKT_NAME);
|
||||
|
||||
let conf = WebsiteConfiguration::builder()
|
||||
.routing_rules(
|
||||
RoutingRule::builder()
|
||||
.condition(Condition::builder().key_prefix_equals("").build())
|
||||
.redirect(
|
||||
Redirect::builder()
|
||||
.protocol(Protocol::Https)
|
||||
.host_name("other.tld")
|
||||
.replace_key_prefix_with("")
|
||||
.build(),
|
||||
)
|
||||
.build(),
|
||||
)
|
||||
.build();
|
||||
|
||||
ctx.client
|
||||
.put_bucket_website()
|
||||
.bucket(&bucket)
|
||||
.website_configuration(conf)
|
||||
.send()
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let req = Request::builder()
|
||||
.method("GET")
|
||||
.uri(format!("http://127.0.0.1:{}/my-path", ctx.garage.web_port))
|
||||
.header("Host", format!("{}.web.garage", BCKT_NAME))
|
||||
.body(Body::new(Bytes::new()))
|
||||
.unwrap();
|
||||
|
||||
let client = Client::builder(TokioExecutor::new()).build_http();
|
||||
let resp = client.request(req).await.unwrap();
|
||||
assert_eq!(resp.status(), StatusCode::FOUND);
|
||||
assert_eq!(
|
||||
resp.headers()
|
||||
.get(hyper::header::LOCATION)
|
||||
.unwrap()
|
||||
.to_str()
|
||||
.unwrap(),
|
||||
"https://other.tld/my-path"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_website_redirect() {
|
||||
const BCKT_NAME: &str = "my-redirect";
|
||||
let ctx = common::context();
|
||||
let bucket = ctx.create_bucket(BCKT_NAME);
|
||||
|
||||
ctx.client
|
||||
.put_object()
|
||||
.bucket(&bucket)
|
||||
.key("index.html")
|
||||
.body(ByteStream::from_static(b"index"))
|
||||
.send()
|
||||
.await
|
||||
.unwrap();
|
||||
ctx.client
|
||||
.put_object()
|
||||
.bucket(&bucket)
|
||||
.key("404.html")
|
||||
.body(ByteStream::from_static(b"main 404"))
|
||||
.send()
|
||||
.await
|
||||
.unwrap();
|
||||
ctx.client
|
||||
.put_object()
|
||||
.bucket(&bucket)
|
||||
.key("static-file")
|
||||
.body(ByteStream::from_static(b"static file"))
|
||||
.send()
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let mut conf = WebsiteConfiguration::builder()
|
||||
.index_document(
|
||||
IndexDocument::builder()
|
||||
.suffix("home.html")
|
||||
.build()
|
||||
.unwrap(),
|
||||
)
|
||||
.error_document(ErrorDocument::builder().key("404.html").build().unwrap());
|
||||
|
||||
for (prefix, condition) in [("unconditional", false), ("conditional", true)] {
|
||||
let code = condition.then(|| "404".to_string());
|
||||
conf = conf
|
||||
// simple redirect
|
||||
.routing_rules(
|
||||
RoutingRule::builder()
|
||||
.condition(
|
||||
Condition::builder()
|
||||
.set_http_error_code_returned_equals(code.clone())
|
||||
.key_prefix_equals(format!("{prefix}/redirect-prefix/"))
|
||||
.build(),
|
||||
)
|
||||
.redirect(
|
||||
Redirect::builder()
|
||||
.http_redirect_code("302")
|
||||
.replace_key_prefix_with("other-prefix/")
|
||||
.build(),
|
||||
)
|
||||
.build(),
|
||||
)
|
||||
.routing_rules(
|
||||
RoutingRule::builder()
|
||||
.condition(
|
||||
Condition::builder()
|
||||
.set_http_error_code_returned_equals(code.clone())
|
||||
.key_prefix_equals(format!("{prefix}/redirect-prefix-307/"))
|
||||
.build(),
|
||||
)
|
||||
.redirect(
|
||||
Redirect::builder()
|
||||
.http_redirect_code("307")
|
||||
.replace_key_prefix_with("other-prefix/")
|
||||
.build(),
|
||||
)
|
||||
.build(),
|
||||
)
|
||||
// simple redirect
|
||||
.routing_rules(
|
||||
RoutingRule::builder()
|
||||
.condition(
|
||||
Condition::builder()
|
||||
.set_http_error_code_returned_equals(code.clone())
|
||||
.key_prefix_equals(format!("{prefix}/redirect-fixed/"))
|
||||
.build(),
|
||||
)
|
||||
.redirect(
|
||||
Redirect::builder()
|
||||
.http_redirect_code("302")
|
||||
.replace_key_with("fixed_key")
|
||||
.build(),
|
||||
)
|
||||
.build(),
|
||||
)
|
||||
// stream other file
|
||||
.routing_rules(
|
||||
RoutingRule::builder()
|
||||
.condition(
|
||||
Condition::builder()
|
||||
.set_http_error_code_returned_equals(code.clone())
|
||||
.key_prefix_equals(format!("{prefix}/stream-fixed/"))
|
||||
.build(),
|
||||
)
|
||||
.redirect(
|
||||
Redirect::builder()
|
||||
.http_redirect_code("200")
|
||||
.replace_key_with("static-file")
|
||||
.build(),
|
||||
)
|
||||
.build(),
|
||||
)
|
||||
// stream other file as error
|
||||
.routing_rules(
|
||||
RoutingRule::builder()
|
||||
.condition(
|
||||
Condition::builder()
|
||||
.set_http_error_code_returned_equals(code.clone())
|
||||
.key_prefix_equals(format!("{prefix}/stream-404/"))
|
||||
.build(),
|
||||
)
|
||||
.redirect(
|
||||
Redirect::builder()
|
||||
.http_redirect_code("404")
|
||||
.replace_key_with("static-file")
|
||||
.build(),
|
||||
)
|
||||
.build(),
|
||||
)
|
||||
// fail to stream other file
|
||||
.routing_rules(
|
||||
RoutingRule::builder()
|
||||
.condition(
|
||||
Condition::builder()
|
||||
.set_http_error_code_returned_equals(code.clone())
|
||||
.key_prefix_equals(format!("{prefix}/stream-missing/"))
|
||||
.build(),
|
||||
)
|
||||
.redirect(
|
||||
Redirect::builder()
|
||||
.http_redirect_code("200")
|
||||
.replace_key_with("missing-file")
|
||||
.build(),
|
||||
)
|
||||
.build(),
|
||||
);
|
||||
}
|
||||
let conf = conf.build();
|
||||
|
||||
ctx.client
|
||||
.put_bucket_website()
|
||||
.bucket(&bucket)
|
||||
.website_configuration(conf.clone())
|
||||
.send()
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let stored_cfg = ctx
|
||||
.client
|
||||
.get_bucket_website()
|
||||
.bucket(&bucket)
|
||||
.send()
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(stored_cfg.index_document, conf.index_document);
|
||||
assert_eq!(stored_cfg.error_document, conf.error_document);
|
||||
assert_eq!(stored_cfg.routing_rules, conf.routing_rules);
|
||||
|
||||
let req = |path| {
|
||||
Request::builder()
|
||||
.method("GET")
|
||||
.uri(format!(
|
||||
"http://127.0.0.1:{}/{}/path",
|
||||
ctx.garage.web_port, path
|
||||
))
|
||||
.header("Host", format!("{}.web.garage", BCKT_NAME))
|
||||
.body(Body::new(Bytes::new()))
|
||||
.unwrap()
|
||||
};
|
||||
|
||||
test_redirect_helper("unconditional", true, &req).await;
|
||||
test_redirect_helper("conditional", true, &req).await;
|
||||
for prefix in ["unconditional", "conditional"] {
|
||||
for rule_path in [
|
||||
"redirect-prefix",
|
||||
"redirect-prefix-307",
|
||||
"redirect-fixed",
|
||||
"stream-fixed",
|
||||
"stream-404",
|
||||
"stream-missing",
|
||||
] {
|
||||
ctx.client
|
||||
.put_object()
|
||||
.bucket(&bucket)
|
||||
.key(format!("{prefix}/{rule_path}/path"))
|
||||
.body(ByteStream::from_static(b"i exist"))
|
||||
.send()
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
}
|
||||
test_redirect_helper("unconditional", true, &req).await;
|
||||
test_redirect_helper("conditional", false, &req).await;
|
||||
}
|
||||
|
||||
async fn test_redirect_helper(
|
||||
prefix: &str,
|
||||
should_see_redirect: bool,
|
||||
req: impl Fn(String) -> Request<http_body_util::Full<Bytes>>,
|
||||
) {
|
||||
use http::header;
|
||||
let client = Client::builder(TokioExecutor::new()).build_http();
|
||||
let expected_body = b"i exist".as_ref();
|
||||
|
||||
let resp = client
|
||||
.request(req(format!("{prefix}/redirect-prefix")))
|
||||
.await
|
||||
.unwrap();
|
||||
if should_see_redirect {
|
||||
assert_eq!(resp.status(), StatusCode::FOUND);
|
||||
assert_eq!(
|
||||
resp.headers()
|
||||
.get(header::LOCATION)
|
||||
.unwrap()
|
||||
.to_str()
|
||||
.unwrap(),
|
||||
"/other-prefix/path"
|
||||
);
|
||||
assert!(resp
|
||||
.into_body()
|
||||
.collect()
|
||||
.await
|
||||
.unwrap()
|
||||
.to_bytes()
|
||||
.is_empty());
|
||||
} else {
|
||||
assert_eq!(resp.status(), StatusCode::OK);
|
||||
assert!(resp.headers().get(header::LOCATION).is_none());
|
||||
assert_eq!(
|
||||
resp.into_body().collect().await.unwrap().to_bytes(),
|
||||
expected_body,
|
||||
);
|
||||
}
|
||||
|
||||
let resp = client
|
||||
.request(req(format!("{prefix}/redirect-prefix-307")))
|
||||
.await
|
||||
.unwrap();
|
||||
if should_see_redirect {
|
||||
assert_eq!(resp.status(), StatusCode::TEMPORARY_REDIRECT);
|
||||
assert_eq!(
|
||||
resp.headers()
|
||||
.get(header::LOCATION)
|
||||
.unwrap()
|
||||
.to_str()
|
||||
.unwrap(),
|
||||
"/other-prefix/path"
|
||||
);
|
||||
assert!(resp
|
||||
.into_body()
|
||||
.collect()
|
||||
.await
|
||||
.unwrap()
|
||||
.to_bytes()
|
||||
.is_empty());
|
||||
} else {
|
||||
assert_eq!(resp.status(), StatusCode::OK);
|
||||
assert!(resp.headers().get(header::LOCATION).is_none());
|
||||
assert_eq!(
|
||||
resp.into_body().collect().await.unwrap().to_bytes(),
|
||||
expected_body,
|
||||
);
|
||||
}
|
||||
|
||||
let resp = client
|
||||
.request(req(format!("{prefix}/redirect-fixed")))
|
||||
.await
|
||||
.unwrap();
|
||||
if should_see_redirect {
|
||||
assert_eq!(resp.status(), StatusCode::FOUND);
|
||||
assert_eq!(
|
||||
resp.headers()
|
||||
.get(header::LOCATION)
|
||||
.unwrap()
|
||||
.to_str()
|
||||
.unwrap(),
|
||||
"/fixed_key"
|
||||
);
|
||||
assert!(resp
|
||||
.into_body()
|
||||
.collect()
|
||||
.await
|
||||
.unwrap()
|
||||
.to_bytes()
|
||||
.is_empty());
|
||||
} else {
|
||||
assert_eq!(resp.status(), StatusCode::OK);
|
||||
assert!(resp.headers().get(header::LOCATION).is_none());
|
||||
assert_eq!(
|
||||
resp.into_body().collect().await.unwrap().to_bytes(),
|
||||
expected_body,
|
||||
);
|
||||
}
|
||||
let resp = client
|
||||
.request(req(format!("{prefix}/stream-fixed")))
|
||||
.await
|
||||
.unwrap();
|
||||
if should_see_redirect {
|
||||
assert_eq!(resp.status(), StatusCode::OK);
|
||||
assert!(resp.headers().get(header::LOCATION).is_none());
|
||||
assert_eq!(
|
||||
resp.into_body().collect().await.unwrap().to_bytes(),
|
||||
b"static file".as_ref(),
|
||||
);
|
||||
} else {
|
||||
assert_eq!(resp.status(), StatusCode::OK);
|
||||
assert!(resp.headers().get(header::LOCATION).is_none());
|
||||
assert_eq!(
|
||||
resp.into_body().collect().await.unwrap().to_bytes(),
|
||||
expected_body,
|
||||
);
|
||||
}
|
||||
let resp = client
|
||||
.request(req(format!("{prefix}/stream-404")))
|
||||
.await
|
||||
.unwrap();
|
||||
if should_see_redirect {
|
||||
assert_eq!(resp.status(), StatusCode::NOT_FOUND);
|
||||
assert!(resp.headers().get(header::LOCATION).is_none());
|
||||
assert_eq!(
|
||||
resp.into_body().collect().await.unwrap().to_bytes(),
|
||||
b"static file".as_ref(),
|
||||
);
|
||||
} else {
|
||||
assert_eq!(resp.status(), StatusCode::OK);
|
||||
assert!(resp.headers().get(header::LOCATION).is_none());
|
||||
assert_eq!(
|
||||
resp.into_body().collect().await.unwrap().to_bytes(),
|
||||
expected_body,
|
||||
);
|
||||
}
|
||||
let resp = client
|
||||
.request(req(format!("{prefix}/stream-404")))
|
||||
.await
|
||||
.unwrap();
|
||||
if should_see_redirect {
|
||||
assert_eq!(resp.status(), StatusCode::NOT_FOUND);
|
||||
assert!(resp.headers().get(header::LOCATION).is_none());
|
||||
assert_eq!(
|
||||
resp.into_body().collect().await.unwrap().to_bytes(),
|
||||
b"static file".as_ref(),
|
||||
);
|
||||
} else {
|
||||
assert_eq!(resp.status(), StatusCode::OK);
|
||||
assert!(resp.headers().get(header::LOCATION).is_none());
|
||||
assert_eq!(
|
||||
resp.into_body().collect().await.unwrap().to_bytes(),
|
||||
expected_body,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_website_invalid_redirect() {
|
||||
const BCKT_NAME: &str = "my-invalid-redirect";
|
||||
let ctx = common::context();
|
||||
let bucket = ctx.create_bucket(BCKT_NAME);
|
||||
|
||||
let conf = WebsiteConfiguration::builder()
|
||||
.routing_rules(
|
||||
RoutingRule::builder()
|
||||
.condition(Condition::builder().key_prefix_equals("").build())
|
||||
.redirect(
|
||||
Redirect::builder()
|
||||
.protocol(Protocol::Https)
|
||||
.host_name("other.tld")
|
||||
.replace_key_prefix_with("")
|
||||
// we don't allow 200 with hostname
|
||||
.http_redirect_code("200")
|
||||
.build(),
|
||||
)
|
||||
.build(),
|
||||
)
|
||||
.build();
|
||||
|
||||
ctx.client
|
||||
.put_bucket_website()
|
||||
.bucket(&bucket)
|
||||
.website_configuration(conf)
|
||||
.send()
|
||||
.await
|
||||
.unwrap_err();
|
||||
}
|
||||
|
||||
+116
-1
@@ -119,7 +119,122 @@ mod v08 {
|
||||
impl garage_util::migrate::InitialFormat for Bucket {}
|
||||
}
|
||||
|
||||
pub use v08::*;
|
||||
mod v2 {
|
||||
use crate::permission::BucketKeyPerm;
|
||||
use garage_util::crdt;
|
||||
use garage_util::data::Uuid;
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
use super::v08;
|
||||
|
||||
pub use v08::{BucketQuotas, CorsRule, LifecycleExpiration, LifecycleFilter, LifecycleRule};
|
||||
|
||||
#[derive(PartialEq, Eq, Clone, Debug, Serialize, Deserialize)]
|
||||
pub struct Bucket {
|
||||
/// ID of the bucket
|
||||
pub id: Uuid,
|
||||
/// State, and configuration if not deleted, of the bucket
|
||||
pub state: crdt::Deletable<BucketParams>,
|
||||
}
|
||||
|
||||
/// Configuration for a bucket
|
||||
#[derive(PartialEq, Eq, Clone, Debug, Serialize, Deserialize)]
|
||||
pub struct BucketParams {
|
||||
/// Bucket's creation date
|
||||
pub creation_date: u64,
|
||||
/// Map of key with access to the bucket, and what kind of access they give
|
||||
pub authorized_keys: crdt::Map<String, BucketKeyPerm>,
|
||||
|
||||
/// Map of aliases that are or have been given to this bucket
|
||||
/// in the global namespace
|
||||
/// (not authoritative: this is just used as an indication to
|
||||
/// map back to aliases when doing ListBuckets)
|
||||
pub aliases: crdt::LwwMap<String, bool>,
|
||||
/// Map of aliases that are or have been given to this bucket
|
||||
/// in namespaces local to keys
|
||||
/// key = (access key id, alias name)
|
||||
pub local_aliases: crdt::LwwMap<(String, String), bool>,
|
||||
|
||||
/// Whether this bucket is allowed for website access
|
||||
/// (under all of its global alias names),
|
||||
/// and if so, the website configuration XML document
|
||||
pub website_config: crdt::Lww<Option<WebsiteConfig>>,
|
||||
/// CORS rules
|
||||
pub cors_config: crdt::Lww<Option<Vec<CorsRule>>>,
|
||||
/// Lifecycle configuration
|
||||
pub lifecycle_config: crdt::Lww<Option<Vec<LifecycleRule>>>,
|
||||
/// Bucket quotas
|
||||
pub quotas: crdt::Lww<BucketQuotas>,
|
||||
}
|
||||
|
||||
#[derive(PartialEq, Eq, Clone, Debug, Serialize, Deserialize)]
|
||||
pub struct WebsiteConfig {
|
||||
pub index_document: String,
|
||||
pub error_document: Option<String>,
|
||||
// this field is currently unused, but present so adding it in the future doesn't
|
||||
// need a new migration
|
||||
pub redirect_all: Option<RedirectAll>,
|
||||
pub routing_rules: Vec<RoutingRule>,
|
||||
}
|
||||
|
||||
#[derive(PartialEq, Eq, Clone, Debug, Serialize, Deserialize)]
|
||||
pub struct RedirectAll {
|
||||
pub hostname: String,
|
||||
pub protocol: String,
|
||||
}
|
||||
|
||||
#[derive(PartialEq, Eq, Clone, Debug, Serialize, Deserialize)]
|
||||
pub struct RoutingRule {
|
||||
pub condition: Option<RedirectCondition>,
|
||||
pub redirect: Redirect,
|
||||
}
|
||||
|
||||
#[derive(PartialEq, Eq, Clone, Debug, Serialize, Deserialize)]
|
||||
pub struct RedirectCondition {
|
||||
pub http_error_code: Option<u16>,
|
||||
pub prefix: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(PartialEq, Eq, Clone, Debug, Serialize, Deserialize)]
|
||||
pub struct Redirect {
|
||||
pub hostname: Option<String>,
|
||||
pub http_redirect_code: u16,
|
||||
pub protocol: Option<String>,
|
||||
pub replace_key_prefix: Option<String>,
|
||||
pub replace_key: Option<String>,
|
||||
}
|
||||
|
||||
impl garage_util::migrate::Migrate for Bucket {
|
||||
const VERSION_MARKER: &'static [u8] = b"G2bkt";
|
||||
|
||||
type Previous = v08::Bucket;
|
||||
|
||||
fn migrate(old: v08::Bucket) -> Bucket {
|
||||
Bucket {
|
||||
id: old.id,
|
||||
state: old.state.map(|x| BucketParams {
|
||||
creation_date: x.creation_date,
|
||||
authorized_keys: x.authorized_keys,
|
||||
aliases: x.aliases,
|
||||
local_aliases: x.local_aliases,
|
||||
website_config: x.website_config.map(|wc_opt| {
|
||||
wc_opt.map(|wc| WebsiteConfig {
|
||||
index_document: wc.index_document,
|
||||
error_document: wc.error_document,
|
||||
redirect_all: None,
|
||||
routing_rules: vec![],
|
||||
})
|
||||
}),
|
||||
cors_config: x.cors_config,
|
||||
lifecycle_config: x.lifecycle_config,
|
||||
quotas: x.quotas,
|
||||
}),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub use v2::*;
|
||||
|
||||
impl AutoCrdt for BucketQuotas {
|
||||
const WARN_IF_DIFFERENT: bool = true;
|
||||
|
||||
+7
-32
@@ -175,13 +175,7 @@ impl Garage {
|
||||
|
||||
// ---- admin tables ----
|
||||
info!("Initialize bucket_table...");
|
||||
let bucket_table = Table::new(
|
||||
BucketTable,
|
||||
control_rep_param.clone(),
|
||||
system.clone(),
|
||||
&db,
|
||||
&config.experimental.merkle_backpressure,
|
||||
);
|
||||
let bucket_table = Table::new(BucketTable, control_rep_param.clone(), system.clone(), &db);
|
||||
|
||||
info!("Initialize bucket_alias_table...");
|
||||
let bucket_alias_table = Table::new(
|
||||
@@ -189,16 +183,9 @@ impl Garage {
|
||||
control_rep_param.clone(),
|
||||
system.clone(),
|
||||
&db,
|
||||
&config.experimental.merkle_backpressure,
|
||||
);
|
||||
info!("Initialize key_table_table...");
|
||||
let key_table = Table::new(
|
||||
KeyTable,
|
||||
control_rep_param,
|
||||
system.clone(),
|
||||
&db,
|
||||
&config.experimental.merkle_backpressure,
|
||||
);
|
||||
let key_table = Table::new(KeyTable, control_rep_param, system.clone(), &db);
|
||||
|
||||
// ---- S3 tables ----
|
||||
info!("Initialize block_ref_table...");
|
||||
@@ -209,7 +196,6 @@ impl Garage {
|
||||
meta_rep_param.clone(),
|
||||
system.clone(),
|
||||
&db,
|
||||
&config.experimental.merkle_backpressure,
|
||||
);
|
||||
|
||||
info!("Initialize version_table...");
|
||||
@@ -220,12 +206,10 @@ impl Garage {
|
||||
meta_rep_param.clone(),
|
||||
system.clone(),
|
||||
&db,
|
||||
&config.experimental.merkle_backpressure,
|
||||
);
|
||||
|
||||
info!("Initialize multipart upload counter table...");
|
||||
let mpu_counter_table =
|
||||
IndexCounter::new(system.clone(), meta_rep_param.clone(), &db, &config);
|
||||
let mpu_counter_table = IndexCounter::new(system.clone(), meta_rep_param.clone(), &db);
|
||||
|
||||
info!("Initialize multipart upload table...");
|
||||
let mpu_table = Table::new(
|
||||
@@ -236,12 +220,10 @@ impl Garage {
|
||||
meta_rep_param.clone(),
|
||||
system.clone(),
|
||||
&db,
|
||||
&config.experimental.merkle_backpressure,
|
||||
);
|
||||
|
||||
info!("Initialize object counter table...");
|
||||
let object_counter_table =
|
||||
IndexCounter::new(system.clone(), meta_rep_param.clone(), &db, &config);
|
||||
let object_counter_table = IndexCounter::new(system.clone(), meta_rep_param.clone(), &db);
|
||||
|
||||
info!("Initialize object_table...");
|
||||
#[allow(clippy::redundant_clone)]
|
||||
@@ -254,7 +236,6 @@ impl Garage {
|
||||
meta_rep_param.clone(),
|
||||
system.clone(),
|
||||
&db,
|
||||
&config.experimental.merkle_backpressure,
|
||||
);
|
||||
|
||||
info!("Load lifecycle worker state...");
|
||||
@@ -264,7 +245,7 @@ impl Garage {
|
||||
|
||||
// ---- K2V ----
|
||||
#[cfg(feature = "k2v")]
|
||||
let k2v = GarageK2V::new(system.clone(), &db, meta_rep_param, &config);
|
||||
let k2v = GarageK2V::new(system.clone(), &db, meta_rep_param);
|
||||
|
||||
// ---- setup block refcount recalculation ----
|
||||
// this function can be used to fix inconsistencies in the RC table
|
||||
@@ -354,14 +335,9 @@ impl Garage {
|
||||
|
||||
#[cfg(feature = "k2v")]
|
||||
impl GarageK2V {
|
||||
fn new(
|
||||
system: Arc<System>,
|
||||
db: &db::Db,
|
||||
meta_rep_param: TableShardedReplication,
|
||||
config: &Config,
|
||||
) -> Self {
|
||||
fn new(system: Arc<System>, db: &db::Db, meta_rep_param: TableShardedReplication) -> Self {
|
||||
info!("Initialize K2V counter table...");
|
||||
let counter_table = IndexCounter::new(system.clone(), meta_rep_param.clone(), db, config);
|
||||
let counter_table = IndexCounter::new(system.clone(), meta_rep_param.clone(), db);
|
||||
|
||||
info!("Initialize K2V subscription manager...");
|
||||
let subscriptions = Arc::new(SubscriptionManager::new());
|
||||
@@ -375,7 +351,6 @@ impl GarageK2V {
|
||||
meta_rep_param,
|
||||
system.clone(),
|
||||
db,
|
||||
&config.experimental.merkle_backpressure,
|
||||
);
|
||||
|
||||
info!("Initialize K2V RPC handler...");
|
||||
|
||||
@@ -10,7 +10,6 @@ use garage_db as db;
|
||||
use garage_rpc::layout::LayoutHelper;
|
||||
use garage_rpc::system::System;
|
||||
use garage_util::background::BackgroundRunner;
|
||||
use garage_util::config::Config;
|
||||
use garage_util::data::*;
|
||||
use garage_util::error::*;
|
||||
use garage_util::migrate::Migrate;
|
||||
@@ -174,7 +173,6 @@ impl<T: CountedItem> IndexCounter<T> {
|
||||
system: Arc<System>,
|
||||
replication: TableShardedReplication,
|
||||
db: &db::Db,
|
||||
config: &Config,
|
||||
) -> Arc<Self> {
|
||||
Arc::new(Self {
|
||||
this_node: system.id,
|
||||
@@ -188,7 +186,6 @@ impl<T: CountedItem> IndexCounter<T> {
|
||||
replication,
|
||||
system,
|
||||
db,
|
||||
&config.experimental.merkle_backpressure,
|
||||
),
|
||||
})
|
||||
}
|
||||
|
||||
@@ -39,7 +39,6 @@ kuska-handshake.workspace = true
|
||||
|
||||
opentelemetry = { workspace = true, optional = true }
|
||||
opentelemetry-contrib = { workspace = true, optional = true }
|
||||
tracing.workspace = true
|
||||
|
||||
[dev-dependencies]
|
||||
pretty_env_logger.workspace = true
|
||||
|
||||
+1
-22
@@ -4,7 +4,6 @@ use std::pin::Pin;
|
||||
use std::sync::atomic::{self, AtomicU32};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::task::Poll;
|
||||
use tracing::*;
|
||||
|
||||
use arc_swap::ArcSwapOption;
|
||||
use bytes::Bytes;
|
||||
@@ -15,7 +14,7 @@ use futures::Stream;
|
||||
use kuska_handshake::async_std::{handshake_client, BoxStream};
|
||||
use tokio::net::TcpStream;
|
||||
use tokio::select;
|
||||
use tokio::sync::{mpsc, oneshot, watch, Semaphore};
|
||||
use tokio::sync::{mpsc, oneshot, watch};
|
||||
use tokio_util::compat::*;
|
||||
|
||||
#[cfg(feature = "telemetry")]
|
||||
@@ -26,7 +25,6 @@ use opentelemetry::{
|
||||
#[cfg(feature = "telemetry")]
|
||||
use opentelemetry_contrib::trace::propagator::binary::*;
|
||||
|
||||
use crate::endpoint::RpcInFlightLimiter;
|
||||
use crate::error::*;
|
||||
use crate::message::*;
|
||||
use crate::netapp::*;
|
||||
@@ -43,7 +41,6 @@ pub(crate) struct ClientConn {
|
||||
|
||||
next_query_number: AtomicU32,
|
||||
inflight: Mutex<HashMap<RequestID, oneshot::Sender<ByteStream>>>,
|
||||
rpc_table_write_inflight_limiter: Option<Semaphore>,
|
||||
}
|
||||
|
||||
impl ClientConn {
|
||||
@@ -101,14 +98,8 @@ impl ClientConn {
|
||||
next_query_number: AtomicU32::from(RequestID::default()),
|
||||
query_send: ArcSwapOption::new(Some(Arc::new(query_send))),
|
||||
inflight: Mutex::new(HashMap::new()),
|
||||
rpc_table_write_inflight_limiter: netapp.max_in_flight_table_write.map(Semaphore::new),
|
||||
});
|
||||
|
||||
info!(
|
||||
"Created conn with table write limit set to {}",
|
||||
netapp.max_in_flight_table_write.unwrap_or(0)
|
||||
);
|
||||
|
||||
netapp.connected_as_client(peer_id, conn.clone());
|
||||
|
||||
let debug_name = format!("CLI {}", hex::encode(&peer_id[..8]));
|
||||
@@ -153,21 +144,10 @@ impl ClientConn {
|
||||
req: Req<T>,
|
||||
path: &str,
|
||||
prio: RequestPriority,
|
||||
limiter: RpcInFlightLimiter,
|
||||
) -> Result<Resp<T>, Error>
|
||||
where
|
||||
T: Message,
|
||||
{
|
||||
let _permit = match (limiter, &self.rpc_table_write_inflight_limiter) {
|
||||
(RpcInFlightLimiter::TableWrite, Some(sem)) => {
|
||||
info!(
|
||||
"Available RPC table write slots: {}",
|
||||
sem.available_permits()
|
||||
);
|
||||
Some(sem.acquire().await.unwrap())
|
||||
}
|
||||
_ => None,
|
||||
};
|
||||
let query_send = self.query_send.load_full().ok_or(Error::ConnectionClosed)?;
|
||||
|
||||
let id = self
|
||||
@@ -232,7 +212,6 @@ impl ClientConn {
|
||||
let stream = Box::pin(canceller.for_stream(stream));
|
||||
|
||||
let resp_enc = RespEnc::decode(stream).await?;
|
||||
drop(_permit);
|
||||
debug!("client: got response to request {} (path {})", id, path);
|
||||
Resp::from_enc(resp_enc)
|
||||
}
|
||||
|
||||
+2
-17
@@ -57,13 +57,6 @@ where
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Copy, Clone, Default)]
|
||||
pub enum RpcInFlightLimiter {
|
||||
#[default]
|
||||
NoLimit,
|
||||
TableWrite,
|
||||
}
|
||||
|
||||
// ----
|
||||
|
||||
/// This struct represents an endpoint for message of type `M`.
|
||||
@@ -121,7 +114,6 @@ where
|
||||
target: &NodeID,
|
||||
req: T,
|
||||
prio: RequestPriority,
|
||||
limiter: RpcInFlightLimiter,
|
||||
) -> Result<Resp<M>, Error>
|
||||
where
|
||||
T: IntoReq<M>,
|
||||
@@ -144,10 +136,7 @@ where
|
||||
"Not connected: {}",
|
||||
hex::encode(&target[..8])
|
||||
))),
|
||||
Some(c) => {
|
||||
c.call(req.into_req()?, self.path.as_str(), prio, limiter)
|
||||
.await
|
||||
}
|
||||
Some(c) => c.call(req.into_req()?, self.path.as_str(), prio).await,
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -160,12 +149,8 @@ where
|
||||
target: &NodeID,
|
||||
req: M,
|
||||
prio: RequestPriority,
|
||||
limiter: RpcInFlightLimiter,
|
||||
) -> Result<<M as Message>::Response, Error> {
|
||||
Ok(self
|
||||
.call_streaming(target, req, prio, limiter)
|
||||
.await?
|
||||
.into_msg())
|
||||
Ok(self.call_streaming(target, req, prio).await?.into_msg())
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -74,8 +74,6 @@ pub struct NetApp {
|
||||
pub id: NodeID,
|
||||
/// Private key associated with our peer ID
|
||||
pub privkey: ed25519::SecretKey,
|
||||
/// Config related to netapp
|
||||
pub(crate) max_in_flight_table_write: Option<usize>,
|
||||
|
||||
pub(crate) server_conns: RwLock<HashMap<NodeID, Arc<ServerConn>>>,
|
||||
pub(crate) client_conns: RwLock<HashMap<NodeID, Arc<ClientConn>>>,
|
||||
@@ -103,7 +101,6 @@ impl NetApp {
|
||||
netid: auth::Key,
|
||||
privkey: ed25519::SecretKey,
|
||||
bind_outgoing_to: Option<IpAddr>,
|
||||
max_in_flight_table_write: Option<usize>,
|
||||
) -> Arc<Self> {
|
||||
let mut version_tag = [0u8; 16];
|
||||
version_tag[0..8].copy_from_slice(&u64::to_be_bytes(NETAPP_VERSION_TAG)[..]);
|
||||
@@ -117,7 +114,6 @@ impl NetApp {
|
||||
netid,
|
||||
id,
|
||||
privkey,
|
||||
max_in_flight_table_write,
|
||||
server_conns: RwLock::new(HashMap::new()),
|
||||
client_conns: RwLock::new(HashMap::new()),
|
||||
endpoints: RwLock::new(HashMap::new()),
|
||||
@@ -431,7 +427,6 @@ impl NetApp {
|
||||
server_port,
|
||||
},
|
||||
PRIO_NORMAL,
|
||||
RpcInFlightLimiter::NoLimit,
|
||||
)
|
||||
.await
|
||||
.map(|_| ())
|
||||
|
||||
+2
-7
@@ -406,7 +406,7 @@ impl PeeringManager {
|
||||
ping_time
|
||||
);
|
||||
let ping_response = select! {
|
||||
r = self.ping_endpoint.call(&id, ping_msg, PRIO_HIGH, RpcInFlightLimiter::NoLimit) => r,
|
||||
r = self.ping_endpoint.call(&id, ping_msg, PRIO_HIGH) => r,
|
||||
_ = tokio::time::sleep(ping_timeout) => Err(Error::Message("Ping timeout".into())),
|
||||
};
|
||||
|
||||
@@ -458,12 +458,7 @@ impl PeeringManager {
|
||||
let pex_message = PeerListMessage { list: peer_list };
|
||||
match self
|
||||
.peer_list_endpoint
|
||||
.call(
|
||||
id,
|
||||
pex_message,
|
||||
PRIO_BACKGROUND,
|
||||
RpcInFlightLimiter::NoLimit,
|
||||
)
|
||||
.call(id, pex_message, PRIO_BACKGROUND)
|
||||
.await
|
||||
{
|
||||
Err(e) => warn!("Error doing peer exchange: {}", e),
|
||||
|
||||
+1
-12
@@ -6,7 +6,6 @@ use std::time::Duration;
|
||||
use futures::future::join_all;
|
||||
use futures::stream::futures_unordered::FuturesUnordered;
|
||||
use futures::stream::StreamExt;
|
||||
use garage_net::endpoint::RpcInFlightLimiter;
|
||||
use tokio::select;
|
||||
|
||||
use opentelemetry::KeyValue;
|
||||
@@ -45,8 +44,6 @@ pub struct RequestStrategy<T> {
|
||||
rs_timeout: Timeout,
|
||||
/// Data to drop when everything completes
|
||||
rs_drop_on_complete: T,
|
||||
/// RPC In Flight Limiter
|
||||
rs_inflight_limiter: RpcInFlightLimiter,
|
||||
}
|
||||
|
||||
#[derive(Copy, Clone)]
|
||||
@@ -64,7 +61,6 @@ impl Clone for RequestStrategy<()> {
|
||||
rs_priority: self.rs_priority,
|
||||
rs_timeout: self.rs_timeout,
|
||||
rs_drop_on_complete: (),
|
||||
rs_inflight_limiter: self.rs_inflight_limiter,
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -78,7 +74,6 @@ impl RequestStrategy<()> {
|
||||
rs_priority: prio,
|
||||
rs_timeout: Timeout::Default,
|
||||
rs_drop_on_complete: (),
|
||||
rs_inflight_limiter: RpcInFlightLimiter::NoLimit,
|
||||
}
|
||||
}
|
||||
/// Add an item to be dropped on completion
|
||||
@@ -89,7 +84,6 @@ impl RequestStrategy<()> {
|
||||
rs_priority: self.rs_priority,
|
||||
rs_timeout: self.rs_timeout,
|
||||
rs_drop_on_complete: drop_on_complete,
|
||||
rs_inflight_limiter: RpcInFlightLimiter::NoLimit,
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -115,10 +109,6 @@ impl<T> RequestStrategy<T> {
|
||||
self.rs_timeout = Timeout::Custom(timeout);
|
||||
self
|
||||
}
|
||||
pub fn with_write_limiter(mut self) -> Self {
|
||||
self.rs_inflight_limiter = RpcInFlightLimiter::TableWrite;
|
||||
self
|
||||
}
|
||||
/// Extract drop_on_complete item
|
||||
fn extract_drop_on_complete(self) -> (RequestStrategy<()>, T) {
|
||||
(
|
||||
@@ -128,7 +118,6 @@ impl<T> RequestStrategy<T> {
|
||||
rs_priority: self.rs_priority,
|
||||
rs_timeout: self.rs_timeout,
|
||||
rs_drop_on_complete: (),
|
||||
rs_inflight_limiter: self.rs_inflight_limiter,
|
||||
},
|
||||
self.rs_drop_on_complete,
|
||||
)
|
||||
@@ -196,7 +185,7 @@ impl RpcHelper {
|
||||
|
||||
let node_id = to.into();
|
||||
let rpc_call = endpoint
|
||||
.call_streaming(&node_id, msg, strat.rs_priority, strat.rs_inflight_limiter)
|
||||
.call_streaming(&node_id, msg, strat.rs_priority)
|
||||
.with_context(Context::current_with_span(span))
|
||||
.record_duration(&self.0.metrics.rpc_duration, &metric_tags);
|
||||
|
||||
|
||||
+2
-12
@@ -21,7 +21,7 @@ use garage_net::{NetApp, NetworkKey, NodeID, NodeKey};
|
||||
|
||||
#[cfg(feature = "kubernetes-discovery")]
|
||||
use garage_util::config::KubernetesDiscoveryConfig;
|
||||
use garage_util::config::{Config, DataDirEnum, RpcInFlightLimiterEnum};
|
||||
use garage_util::config::{Config, DataDirEnum};
|
||||
use garage_util::data::*;
|
||||
use garage_util::error::*;
|
||||
use garage_util::persister::Persister;
|
||||
@@ -256,17 +256,7 @@ impl System {
|
||||
let bind_outgoing_to = Some(config)
|
||||
.filter(|x| x.rpc_bind_outgoing)
|
||||
.map(|x| x.rpc_bind_addr.ip());
|
||||
let maybe_max_table_write = match &config.experimental.rpc_in_flight_limiters {
|
||||
RpcInFlightLimiterEnum::None => None,
|
||||
RpcInFlightLimiterEnum::FixedSize(v) => Some(v.max_table_write),
|
||||
};
|
||||
let netapp = NetApp::new(
|
||||
GARAGE_VERSION_TAG,
|
||||
network_key,
|
||||
node_key,
|
||||
bind_outgoing_to,
|
||||
maybe_max_table_write,
|
||||
);
|
||||
let netapp = NetApp::new(GARAGE_VERSION_TAG, network_key, node_key, bind_outgoing_to);
|
||||
let system_endpoint = netapp.endpoint(SYSTEM_RPC_PATH.into());
|
||||
|
||||
// ---- setup netapp public listener and full mesh peering strategy ----
|
||||
|
||||
+35
-133
@@ -3,12 +3,10 @@ use std::convert::TryInto;
|
||||
use std::sync::Arc;
|
||||
|
||||
use serde_bytes::ByteBuf;
|
||||
use tokio::sync::SemaphorePermit;
|
||||
use tokio::sync::{Notify, Semaphore};
|
||||
use tokio::sync::Notify;
|
||||
|
||||
use garage_db as db;
|
||||
|
||||
use garage_util::config::MerkleBackpressureEnum;
|
||||
use garage_util::data::*;
|
||||
use garage_util::error::*;
|
||||
use garage_util::migrate::Migrate;
|
||||
@@ -22,67 +20,6 @@ use crate::replication::*;
|
||||
use crate::schema::*;
|
||||
use crate::util::*;
|
||||
|
||||
pub(crate) struct MerkleTodo {
|
||||
merkle_todo: db::Tree,
|
||||
merkle_todo_notify: Notify,
|
||||
merkle_todo_bounded_queue: Option<Arc<Semaphore>>,
|
||||
}
|
||||
impl Clone for MerkleTodo {
|
||||
fn clone(&self) -> Self {
|
||||
Self {
|
||||
merkle_todo: self.merkle_todo.clone(),
|
||||
merkle_todo_notify: Notify::new(),
|
||||
merkle_todo_bounded_queue: self.merkle_todo_bounded_queue.clone(),
|
||||
}
|
||||
}
|
||||
}
|
||||
impl MerkleTodo {
|
||||
fn new<F: TableSchema>(db: &db::Db, config: &MerkleBackpressureEnum) -> Self {
|
||||
let merkle_todo = db
|
||||
.open_tree(format!("{}:merkle_todo", F::TABLE_NAME))
|
||||
.expect("Unable to open DB Merkle TODO tree");
|
||||
|
||||
let merkle_todo_bounded_queue = match config {
|
||||
MerkleBackpressureEnum::None => None,
|
||||
MerkleBackpressureEnum::FixedQueue(p) => {
|
||||
Some(Arc::new(Semaphore::new(p.max_queue_size)))
|
||||
}
|
||||
};
|
||||
|
||||
Self {
|
||||
merkle_todo,
|
||||
merkle_todo_notify: Notify::new(),
|
||||
merkle_todo_bounded_queue,
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn len(&self) -> Result<usize, db::Error> {
|
||||
self.merkle_todo.len()
|
||||
}
|
||||
|
||||
pub(crate) async fn with_db<F: FnOnce(&db::Tree, SemaphorePermit)>(&self, f: F) {
|
||||
let bounded = self
|
||||
.merkle_todo_bounded_queue
|
||||
.clone()
|
||||
.unwrap_or(Arc::new(Semaphore::new(1)));
|
||||
let permit = bounded.acquire().await.unwrap();
|
||||
f(&self.merkle_todo, permit);
|
||||
}
|
||||
|
||||
pub(crate) fn appended(&self, permit: SemaphorePermit) {
|
||||
permit.forget();
|
||||
self.merkle_todo_notify.notify_one();
|
||||
}
|
||||
|
||||
pub(crate) fn processed(&self) {
|
||||
let bounded = self
|
||||
.merkle_todo_bounded_queue
|
||||
.clone()
|
||||
.unwrap_or(Arc::new(Semaphore::new(1)));
|
||||
bounded.add_permits(1);
|
||||
}
|
||||
}
|
||||
|
||||
pub struct TableData<F: TableSchema, R: TableReplication> {
|
||||
system: Arc<System>,
|
||||
|
||||
@@ -92,7 +29,8 @@ pub struct TableData<F: TableSchema, R: TableReplication> {
|
||||
pub store: db::Tree,
|
||||
|
||||
pub(crate) merkle_tree: db::Tree,
|
||||
pub(crate) merkle_todo: MerkleTodo,
|
||||
pub(crate) merkle_todo: db::Tree,
|
||||
pub(crate) merkle_todo_notify: Notify,
|
||||
|
||||
pub(crate) insert_queue: db::Tree,
|
||||
pub(crate) insert_queue_notify: Arc<Notify>,
|
||||
@@ -100,18 +38,10 @@ pub struct TableData<F: TableSchema, R: TableReplication> {
|
||||
pub(crate) gc_todo: db::Tree,
|
||||
|
||||
pub(crate) metrics: TableMetrics,
|
||||
|
||||
pub(crate) config: MerkleBackpressureEnum,
|
||||
}
|
||||
|
||||
impl<F: TableSchema, R: TableReplication> TableData<F, R> {
|
||||
pub fn new(
|
||||
system: Arc<System>,
|
||||
instance: F,
|
||||
replication: R,
|
||||
db: &db::Db,
|
||||
config: &MerkleBackpressureEnum,
|
||||
) -> Arc<Self> {
|
||||
pub fn new(system: Arc<System>, instance: F, replication: R, db: &db::Db) -> Arc<Self> {
|
||||
let store = db
|
||||
.open_tree(format!("{}:table", F::TABLE_NAME))
|
||||
.expect("Unable to open DB tree");
|
||||
@@ -119,8 +49,9 @@ impl<F: TableSchema, R: TableReplication> TableData<F, R> {
|
||||
let merkle_tree = db
|
||||
.open_tree(format!("{}:merkle_tree", F::TABLE_NAME))
|
||||
.expect("Unable to open DB Merkle tree tree");
|
||||
|
||||
let merkle_todo = MerkleTodo::new::<F>(db, config);
|
||||
let merkle_todo = db
|
||||
.open_tree(format!("{}:merkle_todo", F::TABLE_NAME))
|
||||
.expect("Unable to open DB Merkle TODO tree");
|
||||
|
||||
let insert_queue = db
|
||||
.open_tree(format!("{}:insert_queue", F::TABLE_NAME))
|
||||
@@ -145,11 +76,11 @@ impl<F: TableSchema, R: TableReplication> TableData<F, R> {
|
||||
store,
|
||||
merkle_tree,
|
||||
merkle_todo,
|
||||
merkle_todo_notify: Notify::new(),
|
||||
insert_queue,
|
||||
insert_queue_notify: Arc::new(Notify::new()),
|
||||
gc_todo,
|
||||
metrics,
|
||||
config: config.clone(),
|
||||
})
|
||||
}
|
||||
|
||||
@@ -236,8 +167,6 @@ impl<F: TableSchema, R: TableReplication> TableData<F, R> {
|
||||
// - When an entry is modified or deleted, add it to the merkle updater's todo list.
|
||||
// This has to be done atomically with the modification for the merkle updater
|
||||
// to maintain consistency. The merkle updater must then be notified with todo_notify.
|
||||
// Also to avoid overloading the merkle updater, you need to sleep a given amount of
|
||||
// time to enable backpressure (ie. slow down clients).
|
||||
// - When an entry is updated to be a tombstone, add it to the gc_todo tree
|
||||
|
||||
pub(crate) fn update_many<T: Borrow<ByteBuf>>(&self, entries: &[T]) -> Result<(), Error> {
|
||||
@@ -272,7 +201,6 @@ impl<F: TableSchema, R: TableReplication> TableData<F, R> {
|
||||
) -> Result<Option<F::E>, Error> {
|
||||
let tree_key = self.tree_key(partition_key, sort_key);
|
||||
|
||||
// transaction begins
|
||||
let changed = self.store.db().transaction(|tx| {
|
||||
let (old_entry, old_bytes, new_entry) = match tx.get(&self.store, &tree_key)? {
|
||||
Some(old_bytes) => {
|
||||
@@ -310,44 +238,31 @@ impl<F: TableSchema, R: TableReplication> TableData<F, R> {
|
||||
Ok(None)
|
||||
}
|
||||
})?;
|
||||
// transaction ends
|
||||
|
||||
// early return if nothing changed
|
||||
let (new_entry, new_bytes_hash) = match changed {
|
||||
Some((e, b)) => (e, b),
|
||||
None => {
|
||||
let maybe_bound = self.merkle_todo_bounded_queue.clone();
|
||||
if let Some(b) = &maybe_bound {
|
||||
b.add_permits(1);
|
||||
if let Some((new_entry, new_bytes_hash)) = changed {
|
||||
self.metrics.internal_update_counter.add(1);
|
||||
|
||||
let is_tombstone = new_entry.is_tombstone();
|
||||
self.merkle_todo_notify.notify_one();
|
||||
if is_tombstone {
|
||||
// We are only responsible for GC'ing this item if we are the
|
||||
// "leader" of the partition, i.e. the first node in the
|
||||
// set of nodes that replicates this partition.
|
||||
// This avoids GC loops and does not change the termination properties
|
||||
// of the GC algorithm, as in all cases GC is suspended if
|
||||
// any node of the partition is unavailable.
|
||||
let pk_hash = Hash::try_from(&tree_key[..32]).unwrap();
|
||||
// TODO: this probably breaks when the layout changes
|
||||
let nodes = self.replication.storage_nodes(&pk_hash);
|
||||
if nodes.first() == Some(&self.system.id) {
|
||||
GcTodoEntry::new(tree_key, new_bytes_hash).save(&self.gc_todo)?;
|
||||
}
|
||||
return Ok(None);
|
||||
}
|
||||
};
|
||||
|
||||
// Handle GC in case of tombstone
|
||||
let is_tombstone = new_entry.is_tombstone();
|
||||
if is_tombstone {
|
||||
// We are only responsible for GC'ing this item if we are the
|
||||
// "leader" of the partition, i.e. the first node in the
|
||||
// set of nodes that replicates this partition.
|
||||
// This avoids GC loops and does not change the termination properties
|
||||
// of the GC algorithm, as in all cases GC is suspended if
|
||||
// any node of the partition is unavailable.
|
||||
let pk_hash = Hash::try_from(&tree_key[..32]).unwrap();
|
||||
// TODO: this probably breaks when the layout changes
|
||||
let nodes = self.replication.storage_nodes(&pk_hash);
|
||||
if nodes.first() == Some(&self.system.id) {
|
||||
GcTodoEntry::new(tree_key, new_bytes_hash).save(&self.gc_todo)?;
|
||||
}
|
||||
Ok(Some(new_entry))
|
||||
} else {
|
||||
Ok(None)
|
||||
}
|
||||
|
||||
// Collect metrics
|
||||
self.metrics.internal_update_counter.add(1);
|
||||
|
||||
// Synchronize with the Merkle Worker
|
||||
self.merkle_todo_notify.notify_one(); // Wake-up it
|
||||
|
||||
Ok(Some(new_entry))
|
||||
}
|
||||
|
||||
pub(crate) fn delete_if_equal(self: &Arc<Self>, k: &[u8], v: &[u8]) -> Result<bool, Error> {
|
||||
@@ -367,16 +282,10 @@ impl<F: TableSchema, R: TableReplication> TableData<F, R> {
|
||||
_ => Ok(false),
|
||||
})?;
|
||||
|
||||
if !removed {
|
||||
let maybe_bound = self.merkle_todo_bounded_queue.clone();
|
||||
if let Some(b) = &maybe_bound {
|
||||
b.add_permits(1);
|
||||
}
|
||||
return Ok(false);
|
||||
if removed {
|
||||
self.metrics.internal_delete_counter.add(1);
|
||||
self.merkle_todo_notify.notify_one();
|
||||
}
|
||||
|
||||
self.metrics.internal_delete_counter.add(1);
|
||||
self.merkle_todo_notify.notify_one();
|
||||
Ok(removed)
|
||||
}
|
||||
|
||||
@@ -401,18 +310,11 @@ impl<F: TableSchema, R: TableReplication> TableData<F, R> {
|
||||
_ => Ok(false),
|
||||
})?;
|
||||
|
||||
if !removed {
|
||||
let maybe_bound = self.merkle_todo_bounded_queue.clone();
|
||||
if let Some(b) = &maybe_bound {
|
||||
b.add_permits(1);
|
||||
}
|
||||
return Ok(false);
|
||||
if removed {
|
||||
self.metrics.internal_delete_counter.add(1);
|
||||
self.merkle_todo_notify.notify_one();
|
||||
}
|
||||
|
||||
self.metrics.internal_delete_counter.add(1);
|
||||
self.merkle_todo_notify.notify_one();
|
||||
|
||||
Ok(true)
|
||||
Ok(removed)
|
||||
}
|
||||
|
||||
// ---- Insert queue functions ----
|
||||
|
||||
+3
-10
@@ -262,8 +262,7 @@ impl<F: TableSchema, R: TableReplication> TableGc<F, R> {
|
||||
// GC has been successful for all of these entries.
|
||||
// We now remove them all from our local table and from the GC todo list.
|
||||
for item in items {
|
||||
let _is_removed = self
|
||||
.data
|
||||
self.data
|
||||
.delete_if_equal_hash(&item.key[..], item.value_hash)
|
||||
.err_context("GC: local delete tombstones")?;
|
||||
item.remove_if_equal(&self.data.gc_todo)
|
||||
@@ -276,21 +275,14 @@ impl<F: TableSchema, R: TableReplication> TableGc<F, R> {
|
||||
|
||||
impl<F: TableSchema, R: TableReplication> EndpointHandler<GcRpc> for TableGc<F, R> {
|
||||
async fn handle(self: &Arc<Self>, message: &GcRpc, _from: NodeID) -> Result<GcRpc, Error> {
|
||||
let maybe_bounded = self.data.merkle_todo_bounded_queue.clone();
|
||||
match message {
|
||||
GcRpc::Update(items) => {
|
||||
if let Some(b) = maybe_bounded {
|
||||
b.acquire_many(items.len() as u32).await.unwrap().forget();
|
||||
}
|
||||
self.data.update_many(items)?;
|
||||
Ok(GcRpc::Ok)
|
||||
}
|
||||
GcRpc::DeleteIfEqualHash(items) => {
|
||||
if let Some(b) = maybe_bounded {
|
||||
b.acquire_many(items.len() as u32).await.unwrap().forget();
|
||||
}
|
||||
for (key, vhash) in items.iter() {
|
||||
let _is_removed = self.data.delete_if_equal_hash(&key[..], *vhash)?;
|
||||
self.data.delete_if_equal_hash(&key[..], *vhash)?;
|
||||
}
|
||||
Ok(GcRpc::Ok)
|
||||
}
|
||||
@@ -337,6 +329,7 @@ impl<F: TableSchema, R: TableReplication> Worker for GcWorker<F, R> {
|
||||
}
|
||||
|
||||
async fn wait_for_work(&mut self) -> WorkerState {
|
||||
tokio::time::sleep(self.wait_delay).await;
|
||||
WorkerState::Busy
|
||||
}
|
||||
}
|
||||
|
||||
@@ -9,7 +9,6 @@ use tokio::sync::watch;
|
||||
use garage_db as db;
|
||||
|
||||
use garage_util::background::*;
|
||||
use garage_util::config::MerkleBackpressureEnum;
|
||||
use garage_util::data::*;
|
||||
use garage_util::encode::{nonversioned_decode, nonversioned_encode};
|
||||
use garage_util::error::Error;
|
||||
@@ -71,15 +70,6 @@ impl<F: TableSchema, R: TableReplication> MerkleUpdater<F, R> {
|
||||
pub(crate) fn new(data: Arc<TableData<F, R>>) -> Arc<Self> {
|
||||
let empty_node_hash = blake2sum(&nonversioned_encode(&MerkleNode::Empty).unwrap()[..]);
|
||||
|
||||
// @FIXME: move in worker
|
||||
match &data.config {
|
||||
MerkleBackpressureEnum::None => info!("Merkle Backpressure is not activated"),
|
||||
MerkleBackpressureEnum::FixedQueue(v) => info!(
|
||||
"Merkle backpressure with a fixed queue size (qlen={}) is activated.",
|
||||
v.max_queue_size
|
||||
),
|
||||
}
|
||||
|
||||
Arc::new(Self {
|
||||
data,
|
||||
empty_node_hash,
|
||||
@@ -135,11 +125,6 @@ impl<F: TableSchema, R: TableReplication> MerkleUpdater<F, R> {
|
||||
k
|
||||
);
|
||||
}
|
||||
|
||||
let maybe_bound = self.data.merkle_todo_bounded_queue.clone();
|
||||
if let Some(b) = &maybe_bound {
|
||||
b.add_permits(1);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
+1
-19
@@ -1,16 +1,12 @@
|
||||
use opentelemetry::{global, metrics::*, KeyValue};
|
||||
use std::convert::TryInto;
|
||||
|
||||
use garage_db as db;
|
||||
|
||||
use crate::data::MerkleTodo;
|
||||
|
||||
/// TableMetrics reference all counter used for metrics
|
||||
pub struct TableMetrics {
|
||||
pub(crate) _table_size: ValueObserver<u64>,
|
||||
pub(crate) _merkle_tree_size: ValueObserver<u64>,
|
||||
pub(crate) _merkle_todo_len: ValueObserver<u64>,
|
||||
pub(crate) _merkle_todo_bounded_queue_free: ValueObserver<u64>,
|
||||
pub(crate) _gc_todo_len: ValueObserver<u64>,
|
||||
|
||||
pub(crate) get_request_counter: BoundCounter<u64>,
|
||||
@@ -29,7 +25,7 @@ impl TableMetrics {
|
||||
table_name: &'static str,
|
||||
store: db::Tree,
|
||||
merkle_tree: db::Tree,
|
||||
merkle_todo: MerkleTodo,
|
||||
merkle_todo: db::Tree,
|
||||
gc_todo: db::Tree,
|
||||
) -> Self {
|
||||
let meter = global::meter(table_name);
|
||||
@@ -76,20 +72,6 @@ impl TableMetrics {
|
||||
)
|
||||
.with_description("Merkle tree updater TODO queue length")
|
||||
.init(),
|
||||
_merkle_todo_bounded_queue_free: meter
|
||||
.u64_value_observer(
|
||||
"table.merkle_todo_bounded_queue_free",
|
||||
move |observer| {
|
||||
let maybe_bounded = merkle_todo_bounded_queue.clone();
|
||||
let free: u64 = match &maybe_bounded {
|
||||
Some(v) => v.available_permits().try_into().unwrap(),
|
||||
None => 0,
|
||||
};
|
||||
observer.observe(free, &[KeyValue::new("table_name", table_name)])
|
||||
}
|
||||
)
|
||||
.with_description("Merkle TODO queue free slots")
|
||||
.init(),
|
||||
_gc_todo_len: meter
|
||||
.u64_value_observer(
|
||||
"table.gc_todo_queue_length",
|
||||
|
||||
@@ -43,13 +43,10 @@ impl TableReplication for TableFullReplication {
|
||||
}
|
||||
fn write_quorum(&self) -> usize {
|
||||
let nmembers = self.system.cluster_layout().current().all_nodes().len();
|
||||
|
||||
let max_faults = if nmembers > 1 { 1 } else { 0 };
|
||||
|
||||
if nmembers > max_faults {
|
||||
nmembers - max_faults
|
||||
} else {
|
||||
if nmembers < 3 {
|
||||
1
|
||||
} else {
|
||||
nmembers.div_euclid(2) + 1
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+1
-7
@@ -244,14 +244,8 @@ impl<F: TableSchema, R: TableReplication> TableSyncer<F, R> {
|
||||
|
||||
// All remote nodes have written those items, now we can delete them locally
|
||||
let mut not_removed = 0;
|
||||
let maybe_bounded = self.data.merkle_todo_bounded_queue.clone();
|
||||
if let Some(b) = maybe_bounded {
|
||||
b.acquire_many(items.len() as u32).await.unwrap().forget();
|
||||
}
|
||||
|
||||
for (k, v) in items.iter() {
|
||||
let removed = self.data.delete_if_equal(&k[..], &v[..])?;
|
||||
if !removed {
|
||||
if !self.data.delete_if_equal(&k[..], &v[..])? {
|
||||
not_removed += 1;
|
||||
}
|
||||
}
|
||||
|
||||
+3
-15
@@ -14,7 +14,6 @@ use opentelemetry::{
|
||||
use garage_db as db;
|
||||
|
||||
use garage_util::background::BackgroundRunner;
|
||||
use garage_util::config::MerkleBackpressureEnum;
|
||||
use garage_util::data::*;
|
||||
use garage_util::error::Error;
|
||||
use garage_util::metrics::RecordDuration;
|
||||
@@ -69,18 +68,12 @@ impl<F: TableSchema> Rpc for TableRpc<F> {
|
||||
impl<F: TableSchema, R: TableReplication> Table<F, R> {
|
||||
// =============== PUBLIC INTERFACE FUNCTIONS (new, insert, get, etc) ===============
|
||||
|
||||
pub fn new(
|
||||
instance: F,
|
||||
replication: R,
|
||||
system: Arc<System>,
|
||||
db: &db::Db,
|
||||
config: &MerkleBackpressureEnum,
|
||||
) -> Arc<Self> {
|
||||
pub fn new(instance: F, replication: R, system: Arc<System>, db: &db::Db) -> Arc<Self> {
|
||||
let endpoint = system
|
||||
.netapp
|
||||
.endpoint(format!("garage_table/table.rs/Rpc:{}", F::TABLE_NAME));
|
||||
|
||||
let data = TableData::new(system.clone(), instance, replication, db, config);
|
||||
let data = TableData::new(system.clone(), instance, replication, db);
|
||||
|
||||
let merkle_updater = MerkleUpdater::new(data.clone());
|
||||
|
||||
@@ -138,8 +131,7 @@ impl<F: TableSchema, R: TableReplication> Table<F, R> {
|
||||
who.as_ref(),
|
||||
rpc,
|
||||
RequestStrategy::with_priority(PRIO_NORMAL)
|
||||
.with_quorum(self.data.replication.write_quorum())
|
||||
.with_write_limiter(),
|
||||
.with_quorum(self.data.replication.write_quorum()),
|
||||
)
|
||||
.await?;
|
||||
|
||||
@@ -535,10 +527,6 @@ impl<F: TableSchema, R: TableReplication> EndpointHandler<TableRpc<F>> for Table
|
||||
Ok(TableRpc::Update(values))
|
||||
}
|
||||
TableRpc::Update(pairs) => {
|
||||
let maybe_bounded = self.data.merkle_todo_bounded_queue.clone();
|
||||
if let Some(b) = maybe_bounded {
|
||||
b.acquire_many(pairs.len() as u32).await.unwrap().forget();
|
||||
}
|
||||
self.data.update_many(pairs)?;
|
||||
Ok(TableRpc::Ok)
|
||||
}
|
||||
|
||||
@@ -135,10 +135,6 @@ pub struct Config {
|
||||
/// Configuration for the admin API endpoint
|
||||
#[serde(default = "Default::default")]
|
||||
pub admin: AdminConfig,
|
||||
|
||||
/// --- Experimental
|
||||
#[serde(default = "Default::default")]
|
||||
pub experimental: ExperimentalConfig,
|
||||
}
|
||||
|
||||
/// Value for data_dir: either a single directory or a list of dirs with attributes
|
||||
@@ -259,40 +255,6 @@ pub struct KubernetesDiscoveryConfig {
|
||||
pub skip_crd: bool,
|
||||
}
|
||||
|
||||
#[derive(Deserialize, Debug, Clone, Default)]
|
||||
pub struct ExperimentalConfig {
|
||||
pub merkle_backpressure: MerkleBackpressureEnum,
|
||||
pub rpc_in_flight_limiters: RpcInFlightLimiterEnum,
|
||||
}
|
||||
|
||||
#[derive(Deserialize, Debug, Clone, Default)]
|
||||
#[serde(rename_all = "lowercase", tag = "kind")]
|
||||
pub enum MerkleBackpressureEnum {
|
||||
#[default]
|
||||
None,
|
||||
FixedQueue(MerkleFixedQueue),
|
||||
}
|
||||
|
||||
#[derive(Deserialize, Debug, Clone, Default)]
|
||||
#[serde(rename_all = "lowercase", tag = "kind")]
|
||||
pub enum RpcInFlightLimiterEnum {
|
||||
#[default]
|
||||
None,
|
||||
FixedSize(InFlightFixedSize),
|
||||
}
|
||||
|
||||
#[derive(Deserialize, Debug, Clone, Default)]
|
||||
pub struct InFlightFixedSize {
|
||||
#[serde(default = "default_max_table_write")]
|
||||
pub max_table_write: usize,
|
||||
}
|
||||
|
||||
#[derive(Deserialize, Debug, Clone, Default)]
|
||||
pub struct MerkleFixedQueue {
|
||||
#[serde(default = "default_max_queue_size")]
|
||||
pub max_queue_size: usize,
|
||||
}
|
||||
|
||||
/// Read and parse configuration
|
||||
pub fn read_config(config_file: PathBuf) -> Result<Config, Error> {
|
||||
let config = std::fs::read_to_string(config_file)?;
|
||||
@@ -319,14 +281,6 @@ fn default_compression() -> Option<i32> {
|
||||
Some(1)
|
||||
}
|
||||
|
||||
fn default_max_table_write() -> usize {
|
||||
64
|
||||
}
|
||||
|
||||
fn default_max_queue_size() -> usize {
|
||||
256
|
||||
}
|
||||
|
||||
fn deserialize_compression<'de, D>(deserializer: D) -> Result<Option<i32>, D::Error>
|
||||
where
|
||||
D: de::Deserializer<'de>,
|
||||
|
||||
@@ -9,6 +9,16 @@ pub enum Deletable<T> {
|
||||
Deleted,
|
||||
}
|
||||
|
||||
impl<T> Deletable<T> {
|
||||
/// Map value, used for migrations
|
||||
pub fn map<U, F: FnOnce(T) -> U>(self, f: F) -> Deletable<U> {
|
||||
match self {
|
||||
Self::Present(x) => Deletable::<U>::Present(f(x)),
|
||||
Self::Deleted => Deletable::<U>::Deleted,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: Crdt> Deletable<T> {
|
||||
/// Create a new deletable object that isn't deleted
|
||||
pub fn present(v: T) -> Self {
|
||||
|
||||
@@ -43,6 +43,16 @@ pub struct Lww<T> {
|
||||
v: T,
|
||||
}
|
||||
|
||||
impl<T> Lww<T> {
|
||||
/// Map value, used for migrations
|
||||
pub fn map<U, F: FnOnce(T) -> U>(self, f: F) -> Lww<U> {
|
||||
Lww::<U> {
|
||||
ts: self.ts,
|
||||
v: f(self.v),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<T> Lww<T>
|
||||
where
|
||||
T: Crdt,
|
||||
|
||||
+302
-72
@@ -25,12 +25,14 @@ use garage_api_common::cors::{
|
||||
};
|
||||
use garage_api_common::generic_server::{server_loop, UnixListenerOn};
|
||||
use garage_api_common::helpers::*;
|
||||
use garage_api_s3::api_server::ResBody;
|
||||
use garage_api_s3::error::{
|
||||
CommonErrorDerivative, Error as ApiError, OkOrBadRequest, OkOrInternalError,
|
||||
};
|
||||
use garage_api_s3::get::{handle_get_without_ctx, handle_head_without_ctx};
|
||||
use garage_api_s3::website::X_AMZ_WEBSITE_REDIRECT_LOCATION;
|
||||
|
||||
use garage_model::bucket_table::{self, RoutingRule};
|
||||
use garage_model::garage::Garage;
|
||||
|
||||
use garage_table::*;
|
||||
@@ -260,45 +262,71 @@ impl WebServer {
|
||||
// Get path
|
||||
let path = req.uri().path().to_string();
|
||||
let index = &website_config.index_document;
|
||||
let (key, may_redirect) = path_to_keys(&path, index)?;
|
||||
let routing_result = path_to_keys(&path, index, &website_config.routing_rules)?;
|
||||
|
||||
debug!(
|
||||
"Selected bucket: \"{}\" {:?}, target key: \"{}\", may redirect to: {:?}",
|
||||
bucket_name, bucket_id, key, may_redirect
|
||||
"Selected bucket: \"{}\" {:?}, routing to {:?}",
|
||||
bucket_name, bucket_id, routing_result,
|
||||
);
|
||||
|
||||
let ret_doc = match *req.method() {
|
||||
Method::OPTIONS => handle_options_for_bucket(req, &bucket_params)
|
||||
let ret_doc = match (req.method(), routing_result.main_target()) {
|
||||
(&Method::OPTIONS, _) => handle_options_for_bucket(req, &bucket_params)
|
||||
.map_err(ApiError::from)
|
||||
.map(|res| res.map(|_empty_body: EmptyBody| empty_body())),
|
||||
Method::HEAD => {
|
||||
handle_head_without_ctx(self.garage.clone(), req, bucket_id, &key, None).await
|
||||
(_, Err((url, code))) => Ok(Response::builder()
|
||||
.status(code)
|
||||
.header("Location", url)
|
||||
.body(empty_body())
|
||||
.unwrap()),
|
||||
(_, Ok((key, code))) => {
|
||||
handle_inner(self.garage.clone(), req, bucket_id, key, code).await
|
||||
}
|
||||
Method::GET => {
|
||||
handle_get_without_ctx(
|
||||
};
|
||||
|
||||
// Try handling errors if bucket configuration provided fallbacks
|
||||
let ret_doc_with_redir = match (&ret_doc, &routing_result) {
|
||||
(
|
||||
Err(ApiError::NoSuchKey),
|
||||
RoutingResult::LoadOrRedirect {
|
||||
redirect_if_exists,
|
||||
redirect_url,
|
||||
redirect_code,
|
||||
..
|
||||
},
|
||||
) => {
|
||||
let redirect = if let Some(redirect_key) = redirect_if_exists {
|
||||
self.check_key_exists(bucket_id, redirect_key.as_str())
|
||||
.await?
|
||||
} else {
|
||||
true
|
||||
};
|
||||
if redirect {
|
||||
Ok(Response::builder()
|
||||
.status(redirect_code)
|
||||
.header("Location", redirect_url)
|
||||
.body(empty_body())
|
||||
.unwrap())
|
||||
} else {
|
||||
ret_doc
|
||||
}
|
||||
}
|
||||
(
|
||||
Err(ApiError::NoSuchKey),
|
||||
RoutingResult::LoadOrAlternativeError {
|
||||
redirect_key,
|
||||
redirect_code,
|
||||
..
|
||||
},
|
||||
) => {
|
||||
handle_inner(
|
||||
self.garage.clone(),
|
||||
req,
|
||||
bucket_id,
|
||||
&key,
|
||||
None,
|
||||
Default::default(),
|
||||
redirect_key,
|
||||
*redirect_code,
|
||||
)
|
||||
.await
|
||||
}
|
||||
_ => Err(ApiError::bad_request("HTTP method not supported")),
|
||||
};
|
||||
|
||||
// Try implicit redirect on error
|
||||
let ret_doc_with_redir = match (&ret_doc, may_redirect) {
|
||||
(Err(ApiError::NoSuchKey), ImplicitRedirect::To { key, url })
|
||||
if self.check_key_exists(bucket_id, key.as_str()).await? =>
|
||||
{
|
||||
Ok(Response::builder()
|
||||
.status(StatusCode::FOUND)
|
||||
.header(LOCATION, url)
|
||||
.body(empty_body())
|
||||
.unwrap())
|
||||
}
|
||||
(Ok(ret), _) if ret.headers().contains_key(X_AMZ_WEBSITE_REDIRECT_LOCATION) => {
|
||||
let redirect_location = ret.headers().get(X_AMZ_WEBSITE_REDIRECT_LOCATION).unwrap();
|
||||
Ok(Response::builder()
|
||||
@@ -332,17 +360,17 @@ impl WebServer {
|
||||
// We want to return the error document
|
||||
// Create a fake HTTP request with path = the error document
|
||||
let req2 = Request::builder()
|
||||
.method("GET")
|
||||
.uri(format!("http://{}/{}", host, &error_document))
|
||||
.body(())
|
||||
.unwrap();
|
||||
|
||||
match handle_get_without_ctx(
|
||||
match handle_inner(
|
||||
self.garage.clone(),
|
||||
&req2,
|
||||
bucket_id,
|
||||
&error_document,
|
||||
None,
|
||||
Default::default(),
|
||||
error.http_status_code(),
|
||||
)
|
||||
.await
|
||||
{
|
||||
@@ -357,8 +385,6 @@ impl WebServer {
|
||||
error
|
||||
);
|
||||
|
||||
*error_doc.status_mut() = error.http_status_code();
|
||||
|
||||
// Preserve error message in a special header
|
||||
for error_line in error.to_string().split('\n') {
|
||||
if let Ok(v) = HeaderValue::from_bytes(error_line.as_bytes()) {
|
||||
@@ -389,6 +415,52 @@ impl WebServer {
|
||||
}
|
||||
}
|
||||
|
||||
async fn handle_inner(
|
||||
garage: Arc<Garage>,
|
||||
req: &Request<()>,
|
||||
bucket_id: Uuid,
|
||||
key: &str,
|
||||
status_code: StatusCode,
|
||||
) -> Result<Response<ResBody>, ApiError> {
|
||||
if status_code != StatusCode::OK {
|
||||
// If we are returning an error document, discard all headers from
|
||||
// the original request that would have influenced the result:
|
||||
// - Range header, we don't want to return a subrange of the error document
|
||||
// - Caching directives such as If-None-Match, etc, which are not relevant
|
||||
let cleaned_req = Request::builder().uri(req.uri()).body(()).unwrap();
|
||||
|
||||
let mut ret = match req.method() {
|
||||
&Method::HEAD => {
|
||||
handle_head_without_ctx(garage, &cleaned_req, bucket_id, key, None).await?
|
||||
}
|
||||
&Method::GET => {
|
||||
handle_get_without_ctx(
|
||||
garage,
|
||||
&cleaned_req,
|
||||
bucket_id,
|
||||
key,
|
||||
None,
|
||||
Default::default(),
|
||||
)
|
||||
.await?
|
||||
}
|
||||
_ => return Err(ApiError::bad_request("HTTP method not supported")),
|
||||
};
|
||||
|
||||
*ret.status_mut() = status_code;
|
||||
|
||||
Ok(ret)
|
||||
} else {
|
||||
match req.method() {
|
||||
&Method::HEAD => handle_head_without_ctx(garage, req, bucket_id, key, None).await,
|
||||
&Method::GET => {
|
||||
handle_get_without_ctx(garage, req, bucket_id, key, None, Default::default()).await
|
||||
}
|
||||
_ => Err(ApiError::bad_request("HTTP method not supported")),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn error_to_res(e: Error) -> Response<BoxBody<Error>> {
|
||||
// If we are here, it is either that:
|
||||
// - there was an error before trying to get the requested URL
|
||||
@@ -405,9 +477,44 @@ fn error_to_res(e: Error) -> Response<BoxBody<Error>> {
|
||||
}
|
||||
|
||||
#[derive(Debug, PartialEq)]
|
||||
enum ImplicitRedirect {
|
||||
No,
|
||||
To { key: String, url: String },
|
||||
enum RoutingResult {
|
||||
// Load a key and use `code` as status, or fallback to normal 404 handler if not found
|
||||
LoadKey {
|
||||
key: String,
|
||||
code: StatusCode,
|
||||
},
|
||||
// Load a key and use `200` as status, or fallback with a redirection using `redirect_code`
|
||||
// as status
|
||||
LoadOrRedirect {
|
||||
key: String,
|
||||
redirect_if_exists: Option<String>,
|
||||
redirect_url: String,
|
||||
redirect_code: StatusCode,
|
||||
},
|
||||
// Load a key and use `200` as status, or fallback by loading a different key and use
|
||||
// `redirect_code` as status
|
||||
LoadOrAlternativeError {
|
||||
key: String,
|
||||
redirect_key: String,
|
||||
redirect_code: StatusCode,
|
||||
},
|
||||
// Send an http redirect with `code` as status
|
||||
Redirect {
|
||||
url: String,
|
||||
code: StatusCode,
|
||||
},
|
||||
}
|
||||
|
||||
impl RoutingResult {
|
||||
// return Ok((key_to_deref, status_code)) or Err((redirect_target, status_code))
|
||||
fn main_target(&self) -> Result<(&str, StatusCode), (&str, StatusCode)> {
|
||||
match self {
|
||||
RoutingResult::LoadKey { key, code } => Ok((key, *code)),
|
||||
RoutingResult::LoadOrRedirect { key, .. } => Ok((key, StatusCode::OK)),
|
||||
RoutingResult::LoadOrAlternativeError { key, .. } => Ok((key, StatusCode::OK)),
|
||||
RoutingResult::Redirect { url, code } => Err((url, *code)),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Path to key
|
||||
@@ -417,35 +524,154 @@ enum ImplicitRedirect {
|
||||
/// which is also AWS S3 behavior.
|
||||
///
|
||||
/// Check: https://docs.aws.amazon.com/AmazonS3/latest/userguide/IndexDocumentSupport.html
|
||||
fn path_to_keys<'a>(path: &'a str, index: &str) -> Result<(String, ImplicitRedirect), Error> {
|
||||
fn path_to_keys(
|
||||
path: &str,
|
||||
index: &str,
|
||||
routing_rules: &[RoutingRule],
|
||||
) -> Result<RoutingResult, Error> {
|
||||
let path_utf8 = percent_encoding::percent_decode_str(path).decode_utf8()?;
|
||||
|
||||
let base_key = match path_utf8.strip_prefix("/") {
|
||||
Some(bk) => bk,
|
||||
None => return Err(Error::BadRequest("Path must start with a / (slash)".into())),
|
||||
};
|
||||
let is_bucket_root = base_key.len() == 0;
|
||||
|
||||
let is_bucket_root = base_key.is_empty();
|
||||
let is_trailing_slash = path_utf8.ends_with("/");
|
||||
|
||||
match (is_bucket_root, is_trailing_slash) {
|
||||
// It is not possible to store something at the root of the bucket (ie. empty key),
|
||||
// the only option is to fetch the index
|
||||
(true, _) => Ok((index.to_string(), ImplicitRedirect::No)),
|
||||
let key = if is_bucket_root || is_trailing_slash {
|
||||
// we can't store anything at the root, so we need to query the index
|
||||
// if the key end with a slash, we always query the index
|
||||
format!("{base_key}{index}")
|
||||
} else {
|
||||
// if the key doesn't end with `/`, leave it unmodified
|
||||
base_key.to_string()
|
||||
};
|
||||
|
||||
// "If you create a folder structure in your bucket, you must have an index document at each level. In each folder, the index document must have the same name, for example, index.html. When a user specifies a URL that resembles a folder lookup, the presence or absence of a trailing slash determines the behavior of the website. For example, the following URL, with a trailing slash, returns the photos/index.html index document."
|
||||
(false, true) => Ok((format!("{base_key}{index}"), ImplicitRedirect::No)),
|
||||
let mut routing_rules_iter = routing_rules.iter();
|
||||
let key = loop {
|
||||
let Some(routing_rule) = routing_rules_iter.next() else {
|
||||
break key;
|
||||
};
|
||||
|
||||
// "However, if you exclude the trailing slash from the preceding URL, Amazon S3 first looks for an object photos in the bucket. If the photos object is not found, it searches for an index document, photos/index.html. If that document is found, Amazon S3 returns a 302 Found message and points to the photos/ key. For subsequent requests to photos/, Amazon S3 returns photos/index.html. If the index document is not found, Amazon S3 returns an error."
|
||||
(false, false) => Ok((
|
||||
base_key.to_string(),
|
||||
ImplicitRedirect::To {
|
||||
key: format!("{base_key}/{index}"),
|
||||
url: format!("{path}/"),
|
||||
},
|
||||
)),
|
||||
let Ok(status_code) = StatusCode::from_u16(routing_rule.redirect.http_redirect_code) else {
|
||||
continue;
|
||||
};
|
||||
if let Some(condition) = &routing_rule.condition {
|
||||
let suffix = if let Some(prefix) = &condition.prefix {
|
||||
let Some(suffix) = key.strip_prefix(prefix) else {
|
||||
continue;
|
||||
};
|
||||
Some(suffix)
|
||||
} else {
|
||||
None
|
||||
};
|
||||
let mut target = compute_redirect_target(&routing_rule.redirect, suffix);
|
||||
let query_alternative_key =
|
||||
status_code == StatusCode::OK || status_code == StatusCode::NOT_FOUND;
|
||||
let redirect_on_error =
|
||||
condition.http_error_code == Some(StatusCode::NOT_FOUND.as_u16());
|
||||
match (query_alternative_key, redirect_on_error) {
|
||||
(false, false) => {
|
||||
return Ok(RoutingResult::Redirect {
|
||||
url: target,
|
||||
code: status_code,
|
||||
})
|
||||
}
|
||||
(true, false) => {
|
||||
// we need to remove the leading /
|
||||
target.remove(0);
|
||||
if status_code == StatusCode::OK {
|
||||
break target;
|
||||
} else {
|
||||
return Ok(RoutingResult::LoadKey {
|
||||
key: target,
|
||||
code: status_code,
|
||||
});
|
||||
}
|
||||
}
|
||||
(false, true) => {
|
||||
return Ok(RoutingResult::LoadOrRedirect {
|
||||
key,
|
||||
redirect_if_exists: None,
|
||||
redirect_url: target,
|
||||
redirect_code: status_code,
|
||||
});
|
||||
}
|
||||
(true, true) => {
|
||||
target.remove(0);
|
||||
return Ok(RoutingResult::LoadOrAlternativeError {
|
||||
key,
|
||||
redirect_key: target,
|
||||
redirect_code: status_code,
|
||||
});
|
||||
}
|
||||
}
|
||||
} else {
|
||||
let target = compute_redirect_target(&routing_rule.redirect, None);
|
||||
return Ok(RoutingResult::Redirect {
|
||||
url: target,
|
||||
code: status_code,
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
if is_bucket_root || is_trailing_slash {
|
||||
Ok(RoutingResult::LoadKey {
|
||||
key,
|
||||
code: StatusCode::OK,
|
||||
})
|
||||
} else {
|
||||
Ok(RoutingResult::LoadOrRedirect {
|
||||
redirect_if_exists: Some(format!("{key}/{index}")),
|
||||
// we can't use `path` because key might have changed substentially in case of
|
||||
// routing rules
|
||||
redirect_url: percent_encoding::percent_encode(
|
||||
format!("/{key}/").as_bytes(),
|
||||
PATH_ENCODING_SET,
|
||||
)
|
||||
.to_string(),
|
||||
key,
|
||||
redirect_code: StatusCode::FOUND,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// per https://url.spec.whatwg.org/#path-percent-encode-set
|
||||
const PATH_ENCODING_SET: &percent_encoding::AsciiSet = &percent_encoding::CONTROLS
|
||||
.add(b' ')
|
||||
.add(b'"')
|
||||
.add(b'#')
|
||||
.add(b'<')
|
||||
.add(b'>')
|
||||
.add(b'?')
|
||||
.add(b'`')
|
||||
.add(b'{')
|
||||
.add(b'}');
|
||||
|
||||
fn compute_redirect_target(redirect: &bucket_table::Redirect, suffix: Option<&str>) -> String {
|
||||
let mut res = String::new();
|
||||
if let Some(hostname) = &redirect.hostname {
|
||||
if let Some(protocol) = &redirect.protocol {
|
||||
res.push_str(protocol);
|
||||
res.push_str("://");
|
||||
} else {
|
||||
res.push_str("//");
|
||||
}
|
||||
res.push_str(hostname);
|
||||
}
|
||||
res.push('/');
|
||||
if let Some(replace_key_prefix) = &redirect.replace_key_prefix {
|
||||
res.push_str(replace_key_prefix);
|
||||
if let Some(suffix) = suffix {
|
||||
res.push_str(suffix)
|
||||
}
|
||||
} else if let Some(replace_key) = &redirect.replace_key {
|
||||
res.push_str(replace_key)
|
||||
}
|
||||
res
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
@@ -453,35 +679,39 @@ mod tests {
|
||||
#[test]
|
||||
fn path_to_keys_test() -> Result<(), Error> {
|
||||
assert_eq!(
|
||||
path_to_keys("/file%20.jpg", "index.html")?,
|
||||
(
|
||||
"file .jpg".to_string(),
|
||||
ImplicitRedirect::To {
|
||||
key: "file .jpg/index.html".to_string(),
|
||||
url: "/file%20.jpg/".to_string()
|
||||
}
|
||||
)
|
||||
path_to_keys("/file%20.jpg", "index.html", &[])?,
|
||||
RoutingResult::LoadOrRedirect {
|
||||
key: "file .jpg".to_string(),
|
||||
redirect_url: "/file%20.jpg/".to_string(),
|
||||
redirect_if_exists: Some("file .jpg/index.html".to_string()),
|
||||
redirect_code: StatusCode::FOUND,
|
||||
}
|
||||
);
|
||||
assert_eq!(
|
||||
path_to_keys("/%20t/", "index.html")?,
|
||||
(" t/index.html".to_string(), ImplicitRedirect::No)
|
||||
path_to_keys("/%20t/", "index.html", &[])?,
|
||||
RoutingResult::LoadKey {
|
||||
key: " t/index.html".to_string(),
|
||||
code: StatusCode::OK
|
||||
}
|
||||
);
|
||||
assert_eq!(
|
||||
path_to_keys("/", "index.html")?,
|
||||
("index.html".to_string(), ImplicitRedirect::No)
|
||||
path_to_keys("/", "index.html", &[])?,
|
||||
RoutingResult::LoadKey {
|
||||
key: "index.html".to_string(),
|
||||
code: StatusCode::OK
|
||||
}
|
||||
);
|
||||
assert_eq!(
|
||||
path_to_keys("/hello", "index.html")?,
|
||||
(
|
||||
"hello".to_string(),
|
||||
ImplicitRedirect::To {
|
||||
key: "hello/index.html".to_string(),
|
||||
url: "/hello/".to_string()
|
||||
}
|
||||
)
|
||||
path_to_keys("/hello", "index.html", &[])?,
|
||||
RoutingResult::LoadOrRedirect {
|
||||
key: "hello".to_string(),
|
||||
redirect_url: "/hello/".to_string(),
|
||||
redirect_if_exists: Some("hello/index.html".to_string()),
|
||||
redirect_code: StatusCode::FOUND,
|
||||
}
|
||||
);
|
||||
assert!(path_to_keys("", "index.html").is_err());
|
||||
assert!(path_to_keys("i/am/relative", "index.html").is_err());
|
||||
assert!(path_to_keys("", "index.html", &[]).is_err());
|
||||
assert!(path_to_keys("i/am/relative", "index.html", &[]).is_err());
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user