mirror of
https://github.com/deuxfleurs-org/garage.git
synced 2026-08-16 00:43:11 +00:00
Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Alex Auvolat
|
3fac7ddaa3 | ||
|
Alex Auvolat
|
c384a26991 |
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
|
||||
|
||||
@@ -687,7 +687,7 @@ paths:
|
||||
operationId: "GetBucketInfo"
|
||||
summary: "Get a bucket"
|
||||
description: |
|
||||
Given a bucket identifier (`id`) or a global alias (`globalAlias`), get its information.
|
||||
Given a bucket identifier (`id`) or a global alias (`alias`), get its information.
|
||||
It includes its aliases, its web configuration, keys that have some permissions
|
||||
on it, some statistics (number of objects, size), number of dangling multipart uploads,
|
||||
and its quotas (if any).
|
||||
@@ -701,7 +701,7 @@ paths:
|
||||
example: "b4018dc61b27ccb5c64ec1b24f53454bbbd180697c758c4d47a22a8921864a87"
|
||||
schema:
|
||||
type: string
|
||||
- name: globalAlias
|
||||
- name: alias
|
||||
in: query
|
||||
description: |
|
||||
The exact global alias of one of the existing buckets.
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -71,7 +71,7 @@ The entire procedure would look something like this:
|
||||
|
||||
2. Take each node offline individually to back up its metadata folder, bring them back online once the backup is done.
|
||||
You can do all of the nodes in a single zone at once as that won't impact global cluster availability.
|
||||
Do not try to manually copy the metadata folder of a running node.
|
||||
Do not try to make a backup of the metadata folder of a running node.
|
||||
|
||||
**Since Garage v0.9.4,** you can use the `garage meta snapshot --all` command
|
||||
to take a simultaneous snapshot of the metadata database files of all your
|
||||
|
||||
@@ -129,10 +129,10 @@ 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 \
|
||||
dxflrs/garage:v1.1.0
|
||||
-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:v0.9.4
|
||||
```
|
||||
|
||||
Under Linux, you can substitute `--network host` for `-p 3900:3900 -p 3901:3901 -p 3902:3902 -p 3903:3903`
|
||||
|
||||
@@ -1,3 +1,3 @@
|
||||
# Garage helm3 chart
|
||||
|
||||
Documentation is located [here](https://garagehq.deuxfleurs.fr/documentation/cookbook/kubernetes/).
|
||||
Documentation is located [here](/doc/book/cookbook/kubernetes.md).
|
||||
|
||||
@@ -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__"
|
||||
|
||||
@@ -1,22 +0,0 @@
|
||||
{{- if eq .Values.deployment.kind "StatefulSet" -}}
|
||||
apiVersion: v1
|
||||
kind: Service
|
||||
metadata:
|
||||
name: {{ include "garage.fullname" . }}-headless
|
||||
labels:
|
||||
{{- include "garage.labels" . | nindent 4 }}
|
||||
spec:
|
||||
type: ClusterIP
|
||||
clusterIP: None
|
||||
ports:
|
||||
- port: {{ .Values.service.s3.api.port }}
|
||||
targetPort: 3900
|
||||
protocol: TCP
|
||||
name: s3-api
|
||||
- port: {{ .Values.service.s3.web.port }}
|
||||
targetPort: 3902
|
||||
protocol: TCP
|
||||
name: s3-web
|
||||
selector:
|
||||
{{- include "garage.selectorLabels" . | nindent 4 }}
|
||||
{{- end }}
|
||||
@@ -10,11 +10,12 @@ spec:
|
||||
{{- include "garage.selectorLabels" . | nindent 6 }}
|
||||
{{- if eq .Values.deployment.kind "StatefulSet" }}
|
||||
replicas: {{ .Values.deployment.replicaCount }}
|
||||
serviceName: {{ include "garage.fullname" . }}-headless
|
||||
serviceName: {{ include "garage.fullname" . }}
|
||||
podManagementPolicy: {{ .Values.deployment.podManagementPolicy }}
|
||||
{{- end }}
|
||||
template:
|
||||
metadata:
|
||||
|
||||
annotations:
|
||||
checksum/config: {{ include (print $.Template.BasePath "/configmap.yaml") . | sha256sum }}
|
||||
{{- with .Values.podAnnotations }}
|
||||
@@ -78,14 +79,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: []
|
||||
|
||||
@@ -6,7 +6,7 @@ edition = "2018"
|
||||
license = "AGPL-3.0"
|
||||
description = "Admin API server crate for the Garage object store"
|
||||
repository = "https://git.deuxfleurs.fr/Deuxfleurs/garage"
|
||||
readme = "../../../README.md"
|
||||
readme = "../../README.md"
|
||||
|
||||
[lib]
|
||||
path = "lib.rs"
|
||||
|
||||
@@ -6,7 +6,7 @@ edition = "2018"
|
||||
license = "AGPL-3.0"
|
||||
description = "Common functions for the API server crates for the Garage object store"
|
||||
repository = "https://git.deuxfleurs.fr/Deuxfleurs/garage"
|
||||
readme = "../../../README.md"
|
||||
readme = "../../README.md"
|
||||
|
||||
[lib]
|
||||
path = "lib.rs"
|
||||
|
||||
@@ -6,7 +6,7 @@ edition = "2018"
|
||||
license = "AGPL-3.0"
|
||||
description = "K2V API server crate for the Garage object store"
|
||||
repository = "https://git.deuxfleurs.fr/Deuxfleurs/garage"
|
||||
readme = "../../../README.md"
|
||||
readme = "../../README.md"
|
||||
|
||||
[lib]
|
||||
path = "lib.rs"
|
||||
|
||||
@@ -6,7 +6,7 @@ edition = "2018"
|
||||
license = "AGPL-3.0"
|
||||
description = "S3 API server crate for the Garage object store"
|
||||
repository = "https://git.deuxfleurs.fr/Deuxfleurs/garage"
|
||||
readme = "../../../README.md"
|
||||
readme = "../../README.md"
|
||||
|
||||
[lib]
|
||||
path = "lib.rs"
|
||||
|
||||
@@ -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
|
||||
|
||||
+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 {
|
||||
|
||||
+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",
|
||||
|
||||
+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>,
|
||||
|
||||
Reference in New Issue
Block a user