LemmyNet/pict-rs
2
0
Fork 0
mirror of https://git.asonix.dog/asonix/pict-rs synced 2024-12-21 10:51:25 +00:00
Code Releases Activity

Merge pull request 'Describe metrics' (#51) from asonix/describe-metrics into main

Browse source 
Reviewed-on: https://git.asonix.dog/asonix/pict-rs/pulls/51
This commit is contained in:
asonix 2024-02-05 01:39:20 +00:00
commit 0007649b68
15 changed files with 712 additions and 113 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
src/backgrounded.rs
View file

@ -71,7 +71,7 @@ impl Drop for Backgrounded {
fn drop(&mut self) {
let any_items = self.identifier.is_some() || self.upload_id.is_some();
metrics::counter!("pict-rs.background.upload", "completed" => (!any_items).to_string())
metrics::counter!(crate::init_metrics::BACKGROUND_UPLOAD, "completed" => (!any_items).to_string())
.increment(1);
if any_items {

6
src/concurrent_processor.rs
View file

@ -57,7 +57,7 @@ impl ProcessMap {
completed = &tracing::field::Empty,
);
metrics::counter!("pict-rs.process-map.inserted").increment(1);
metrics::counter!(crate::init_metrics::PROCESS_MAP_INSERTED).increment(1);
(CancelState::Sender { sender }, span)
}
@ -142,7 +142,7 @@ where
let res = std::task::ready!(fut.poll(cx));
if process_map.remove(key).is_some() {
metrics::counter!("pict-rs.process-map.removed").increment(1);
metrics::counter!(crate::init_metrics::PROCESS_MAP_REMOVED).increment(1);
}
if let Ok(tup) = &res {
@ -165,7 +165,7 @@ impl Drop for CancelToken {
self.span.record("completed", completed);
if !completed {
metrics::counter!("pict-rs.process-map.removed").increment(1);
metrics::counter!(crate::init_metrics::PROCESS_MAP_REMOVED).increment(1);
}
}
}

8
src/generate.rs
View file

@ -27,7 +27,7 @@ struct MetricsGuard {
impl MetricsGuard {
fn guard() -> Self {
metrics::counter!("pict-rs.generate.start").increment(1);
metrics::counter!(crate::init_metrics::GENERATE_START).increment(1);
Self {
start: Instant::now(),
armed: true,
@ -41,9 +41,9 @@ impl MetricsGuard {
impl Drop for MetricsGuard {
fn drop(&mut self) {
metrics::histogram!("pict-rs.generate.duration", "completed" => (!self.armed).to_string())
metrics::histogram!(crate::init_metrics::GENERATE_DURATION, "completed" => (!self.armed).to_string())
.record(self.start.elapsed().as_secs_f64());
metrics::counter!("pict-rs.generate.end", "completed" => (!self.armed).to_string())
metrics::counter!(crate::init_metrics::GENERATE_END, "completed" => (!self.armed).to_string())
.increment(1);
}
}
@ -85,7 +85,7 @@ pub(crate) async fn generate<S: Store + 'static>(
let (details, bytes) = process_map
.process(hash, thumbnail_path, process_fut)
.with_timeout(Duration::from_secs(state.config.media.process_timeout * 4))
.with_metrics("pict-rs.generate.process")
.with_metrics(crate::init_metrics::GENERATE_PROCESS)
.await
.map_err(|_| UploadError::ProcessTimeout)??;

4
src/ingest.rs
View file

@ -196,7 +196,7 @@ where
.body(Body::wrap_stream(crate::stream::make_send(stream)))
.send()
.instrument(tracing::info_span!("external-validation"))
.with_metrics("pict-rs.ingest.external-validation")
.with_metrics(crate::init_metrics::INGEST_EXTERNAL_VALIDATION)
.await?;
if !response.status().is_success() {
@ -303,7 +303,7 @@ impl Drop for Session {
fn drop(&mut self) {
let any_items = self.hash.is_some() || self.alias.is_some() || self.identifier.is_some();
metrics::counter!("pict-rs.ingest.end", "completed" => (!any_items).to_string())
metrics::counter!(crate::init_metrics::INGEST_END, "completed" => (!any_items).to_string())
.increment(1);
if self.hash.is_some() || self.alias.is_some() | self.identifier.is_some() {

588
src/init_metrics.rs Normal file
View file

@ -0,0 +1,588 @@
pub(super) fn init_metrics() {
describe_toplevel();
describe_queue_cleanup();
describe_payload();
describe_job();
describe_queue_process();
describe_ingest();
describe_backgrounded();
describe_concurrent_processor();
describe_repo();
describe_process();
describe_postgres();
describe_middleware();
describe_generate();
describe_object_storage();
}
fn describe_toplevel() {
metrics::describe_counter!(FILES, "How many files have been uploaded to pict-rs");
metrics::describe_counter!(
BACKGROUND_UPLOAD_CLAIM,
"How many uploaded files have been claimed"
);
}
pub(crate) const FILES: &str = "pict-rs.files";
pub(crate) const BACKGROUND_UPLOAD_CLAIM: &str = "pict-rs.background.upload.claim";
fn describe_queue_cleanup() {
metrics::describe_counter!(
CLEANUP_OUTDATED_PROXY,
"How many proxy URLs haven't been accessed in the configured timeframe and have been queued for cleanup"
);
metrics::describe_counter!(
CLEANUP_OUTDATED_VARIANT,
"How many variants haven't been accessed in the configured timeframe and have been queued for cleanup"
);
}
pub(crate) const CLEANUP_OUTDATED_PROXY: &str = "pict-rs.cleanup.outdated-proxy";
pub(crate) const CLEANUP_OUTDATED_VARIANT: &str = "pict-rs.cleanup.outdated-variant";
fn describe_payload() {
metrics::describe_counter!(
PAYLOAD_DRAIN_START,
"How many payloads have been dropped before read to completion and need draining"
);
metrics::describe_counter!(
PAYLOAD_DRAIN_END,
"How many payloads pict-rs has finished draining"
);
metrics::describe_histogram!(
PAYLOAD_DRAIN_DURATION,
"Timings for how long it took to drain dropped payloads"
);
metrics::describe_counter!(
PAYLOAD_DRAIN_FAIL_SEND,
"How many payloads pict-rs has failed to drain due to filling the drain queue"
);
}
pub(crate) const PAYLOAD_DRAIN_START: &str = "pict-rs.payload.drain.start";
pub(crate) const PAYLOAD_DRAIN_END: &str = "pict-rs.payload.drain.end";
pub(crate) const PAYLOAD_DRAIN_DURATION: &str = "pict-rs.payload.drain.duration";
pub(crate) const PAYLOAD_DRAIN_FAIL_SEND: &str = "pict-rs.payload.drain.fail-send";
fn describe_job() {
metrics::describe_counter!(
JOB_START,
"How many times pict-rs has started processing a background job"
);
metrics::describe_histogram!(
JOB_DURAION,
"Timings for how long background jobs take to complete"
);
metrics::describe_counter!(
JOB_END,
"How many times pict-rs has completed processing a background-job"
);
}
pub(crate) const JOB_START: &str = "pict-rs.job.start";
pub(crate) const JOB_DURAION: &str = "pict-rs.job.duration";
pub(crate) const JOB_END: &str = "pict-rs.job.end";
fn describe_queue_process() {
metrics::describe_counter!(
PROCESS_START,
"How many times pict-rs has spawned a background process"
);
metrics::describe_histogram!(
PROCESS_DURATION,
"Timings for how long background processes take to complete"
);
metrics::describe_counter!(PROCESS_END, "How many background processes have completed");
}
pub(crate) const PROCESS_START: &str = "pict-rs.process.start";
pub(crate) const PROCESS_DURATION: &str = "pict-rs.process.duration";
pub(crate) const PROCESS_END: &str = "pict-rs.process.end";
fn describe_ingest() {
metrics::describe_histogram!(
INGEST_EXTERNAL_VALIDATION,
"Timings for externally validating uploaded media"
);
metrics::describe_counter!(INGEST_END, "How many times media has been ingested");
}
pub(crate) const INGEST_EXTERNAL_VALIDATION: &str = "pict-rs.ingest.external-validation";
pub(crate) const INGEST_END: &str = "pict-rs.ingest.end";
fn describe_backgrounded() {
metrics::describe_counter!(
BACKGROUND_UPLOAD,
"How many times an image has been proxied to storage in the background"
);
}
pub(crate) const BACKGROUND_UPLOAD: &str = "pict-rs.background.upload";
fn describe_concurrent_processor() {
metrics::describe_counter!(
PROCESS_MAP_INSERTED,
"How many times a task has claimed rights to processing a variant"
);
metrics::describe_counter!(
PROCESS_MAP_REMOVED,
"How many times a variant has finished processing"
);
}
pub(crate) const PROCESS_MAP_INSERTED: &str = "pict-rs.process-map.inserted";
pub(crate) const PROCESS_MAP_REMOVED: &str = "pict-rs.process-map.removed";
fn describe_repo() {
metrics::describe_counter!(
QUEUE_PUSH,
"How many jobs have been pushed into the job queue"
);
metrics::describe_histogram!(
QUEUE_POP_DURATION,
"Timings for how long it takes to pop a job from the job queue"
);
metrics::describe_counter!(
QUEUE_POP,
"How many jobs have been popped from the job queue"
);
metrics::describe_histogram!(
UPLOAD_WAIT_DURATION,
"Timings for how long an upload is waited on"
);
metrics::describe_counter!(UPLOAD_WAIT, "How many times an upload has been waited on");
}
pub(crate) const QUEUE_PUSH: &str = "pict-rs.queue.push";
pub(crate) const QUEUE_POP_DURATION: &str = "pict-rs.queue.pop.duration";
pub(crate) const QUEUE_POP: &str = "pict-rs.queue.pop";
pub(crate) const UPLOAD_WAIT_DURATION: &str = "pict-rs.upload.wait.duration";
pub(crate) const UPLOAD_WAIT: &str = "pict-rs.upload.wait";
fn describe_process() {
metrics::describe_counter!(
BACKGROUND_UPLOAD_INGEST,
"How many files have been ingested in the background"
);
metrics::describe_histogram!(
BACKGROUND_UPLOAD_INGEST_DURATION,
"Timings for ingesting media in the background"
);
}
pub(crate) const BACKGROUND_UPLOAD_INGEST: &str = "pict-rs.background.upload.ingest";
pub(crate) const BACKGROUND_UPLOAD_INGEST_DURATION: &str =
"pict-rs.background.upload.ingest.duration";
fn describe_postgres() {
metrics::describe_counter!(
POSTGRES_POOL_CONNECTION_CREATE,
"How many connections to postgres have been made"
);
metrics::describe_counter!(
POSTGRES_POOL_CONNECTION_RECYCLE,
"How many connections to postgres have been recycled"
);
metrics::describe_counter!(
POSTGRES_POOL_GET,
"How many times a connection has been retrieved from the connection pool"
);
metrics::describe_histogram!(
POSTGRES_POOL_GET_DURATION,
"How long pict-rs spent waiting for postgres connections from the connection pool"
);
metrics::describe_counter!(
POSTGRES_JOB_NOTIFIER_NOTIFIED,
"How many background job notifications pict-rs has successfully processed from postgres"
);
metrics::describe_counter!(
POSTGRES_UPLOAD_NOTIFIER_NOTIFIED,
"How many upload completion notifications pict-rs has successfully processed from postgres"
);
metrics::describe_counter!(
POSTGRES_NOTIFICATION,
"How many notifications pict-rs has received from postgres"
);
metrics::describe_histogram!(
POSTGRES_HASHES_COUNT,
"Timings for counting the total number of hashes pict-rs is storing"
);
metrics::describe_histogram!(
POSTGRES_HASHES_BOUND,
"Timings for retrieving a timestamp for a given hash"
);
metrics::describe_histogram!(
POSTGRES_HASHES_ORDERED_HASH,
"Timings for retrieving the most recent hash and timestamp before a provided time"
);
metrics::describe_histogram!(
POSTGRES_HASHES_NEXT_HASHES,
"Timings for retrieving the next page of hashes given an ordered-hash bound"
);
metrics::describe_histogram!(
POSTGRES_HASHES_PREV_HASH,
"Timings for retrieving the hash to act as the next hash page's bound"
);
metrics::describe_histogram!(
POSTGRES_HASHES_FIRST_HASHES,
"Timings for retrieving the first page of hashes"
);
metrics::describe_histogram!(
POSTGRES_HASHES_CREATE_HASH,
"Timings for inserting a new hash"
);
metrics::describe_histogram!(
POSTGRES_HASHES_UPDATE_IDENTIFIER,
"Timings for updating the identifier for a provided hash"
);
metrics::describe_histogram!(
POSTGRES_HASHES_IDENTIFIER,
"Timings for fetching the identifier for a provided hash"
);
metrics::describe_histogram!(
POSTGRES_VARIANTS_RELATE_VARIANT_IDENTIFIER,
"Timings for inserting a variant and identifier for a provided hash"
);
metrics::describe_histogram!(
POSTGRES_VARIANTS_IDENTIFIER,
"Timings for fetching an identifier for a provided hash and variant"
);
metrics::describe_histogram!(
POSTGRES_VARIANTS_FOR_HASH,
"Timings for fetching all variants and identifiers for a provided hash"
);
metrics::describe_histogram!(
POSTGRES_VARIANTS_REMOVE,
"Timings for removing a variant for a provided hash"
);
metrics::describe_histogram!(
POSTGRES_HASHES_RELATE_MOTION_IDENTIFIER,
"Timings for relating a still image identifier for a provided hash representing a video"
);
metrics::describe_histogram!(
POSTGRES_HASHES_MOTION_IDENTIFIER,
"Timings for fetching a still image identifier for a provided hash representing a video"
);
metrics::describe_histogram!(
POSTGRES_VARIANTS_CLEANUP,
"Timings for deleting all variants for a provided hash"
);
metrics::describe_histogram!(
POSTGRES_HASHES_CLEANUP,
"Timings for deleting a provided hash"
);
metrics::describe_histogram!(
POSTGRES_ALIASES_CREATE,
"Timings for creating an alias for a provided hash"
);
metrics::describe_histogram!(
POSTGRES_ALIASES_DELETE_TOKEN,
"Timings for fetching a delete token for a provided alias"
);
metrics::describe_histogram!(
POSTGRES_ALIASES_HASH,
"Timings for fetching a hash for a provided alias"
);
metrics::describe_histogram!(
POSTGRES_ALIASES_FOR_HASH,
"Timings for fetching all aliases for a provided hash"
);
metrics::describe_histogram!(
POSTGRES_ALIASES_CLEANUP,
"Timings for deleting a provided alias"
);
metrics::describe_histogram!(POSTGRES_SETTINGS_SET, "Timings for setting a given setting");
metrics::describe_histogram!(
POSTGRES_SETTINGS_GET,
"Timings for getting a provided setting"
);
metrics::describe_histogram!(
POSTGRES_SETTINGS_REMOVE,
"Timings for removing a provided setting"
);
metrics::describe_histogram!(
POSTGRES_DETAILS_RELATE,
"Timings for relating details to a provided identifier"
);
metrics::describe_histogram!(
POSTGRES_DETAILS_GET,
"Timings for getting details for a provided identifier"
);
metrics::describe_histogram!(
POSTGRES_DETAILS_CLEANUP,
"Timings for deleting details for a provided identifier"
);
metrics::describe_histogram!(
POSTGRES_QUEUE_COUNT,
"Timings for counting the size of the job queue"
);
metrics::describe_histogram!(
POSTGRES_QUEUE_PUSH,
"Timings for inserting a new job into the job queue"
);
metrics::describe_histogram!(
POSTGRES_QUEUE_LISTEN,
"Timings for initializing the queue listener"
);
metrics::describe_histogram!(
POSTGRES_QUEUE_REQUEUE,
"Timings for marking stale jobs as ready to pop"
);
metrics::describe_histogram!(
POSTGRES_QUEUE_CLAIM,
"Timings for claiming a job from the job queue"
);
metrics::describe_histogram!(
POSTGRES_QUEUE_HEARTBEAT,
"Timings for updating the provided job's keepalive heartbeat"
);
metrics::describe_histogram!(
POSTGRES_QUEUE_COMPLETE,
"Timings for removing a completed job from the queue"
);
metrics::describe_histogram!(
POSTGRES_STORE_MIGRATION_COUNT,
"Timings for fetching the count of files successfully migrated between stores"
);
metrics::describe_histogram!(
POSTGRES_STORE_MIGRATION_MARK_MIGRATED,
"Timings for marking a given identifier as having been migrated between stores"
);
metrics::describe_histogram!(
POSTGRES_STORE_MIGRATION_IS_MIGRATED,
"Timings for checking if a given identifier has been migrated between stores"
);
metrics::describe_histogram!(
POSTGRES_STORE_MIGRATION_CLEAR,
"Timings for clearing all records of identifiers migrated between stores. This occurs on successful migration"
);
metrics::describe_histogram!(
POSTGRES_PROXY_RELATE_URL,
"Timings for relating a provided proxy URL to an alias"
);
metrics::describe_histogram!(
POSTGRES_PROXY_RELATED,
"Timings for fetching a related alias for a provided proxy URL"
);
metrics::describe_histogram!(
POSTGRES_PROXY_REMOVE_RELATION,
"Timings for removing a proxy URL for a provied alias"
);
metrics::describe_histogram!(
POSTGRES_ALIAS_ACCESS_SET_ACCESSED,
"Timings for marking a given alias as having been accessed"
);
metrics::describe_histogram!(
POSTGRES_ALIAS_ACCESS_ACCESSED_AT,
"Timings for checking when a given alias was last accessed"
);
metrics::describe_histogram!(
POSTGRES_ALIAS_ACCESS_OLDER_ALIASES,
"Timings for fetching a page of aliases last accessed earlier than a given timestamp"
);
metrics::describe_histogram!(
POSTGRES_VARIANT_ACCESS_SET_ACCESSED,
"Timings for marking a given variant as having been accessed"
);
metrics::describe_histogram!(
POSTGRES_VARIANT_ACCESS_ACCESSED_AT,
"Timings for checking when a given variant was last accessed"
);
metrics::describe_histogram!(
POSTGRES_VARIANT_ACCESS_OLDER_VARIANTS,
"Timings for fetching a page of variants last accessed earlier than a given timestamp"
);
metrics::describe_histogram!(
POSTGRES_UPLOADS_CREATE,
"Timings for inserting a new upload ID"
);
metrics::describe_histogram!(
POSTGRES_UPLOADS_LISTEN,
"Timings for initializing the upload listener"
);
metrics::describe_histogram!(
POSTGRES_UPLOADS_WAIT,
"Timings for checking if a given upload is completed"
);
metrics::describe_histogram!(
POSTGRES_UPLOADS_CLAIM,
"Timings for claiming a given completed upload"
);
metrics::describe_histogram!(
POSTGRES_UPLOADS_COMPLETE,
"Timings for marking a given upload as completed"
);
}
pub(crate) const POSTGRES_POOL_CONNECTION_CREATE: &str = "pict-rs.postgres.pool.connection.create";
pub(crate) const POSTGRES_POOL_CONNECTION_RECYCLE: &str =
"pict-rs.postgres.pool.connection.recycle";
pub(crate) const POSTGRES_POOL_GET: &str = "pict-rs.postgres.pool.get";
pub(crate) const POSTGRES_POOL_GET_DURATION: &str = "pict-rs.postgres.pool.duration";
pub(crate) const POSTGRES_JOB_NOTIFIER_NOTIFIED: &str = "pict-rs.postgres.job-notifier.notified";
pub(crate) const POSTGRES_UPLOAD_NOTIFIER_NOTIFIED: &str =
"pict-rs.postgres.upload-notifier.notified";
pub(crate) const POSTGRES_NOTIFICATION: &str = "pict-rs.postgres.notification";
pub(crate) const POSTGRES_HASHES_COUNT: &str = "pict-rs.postgres.hashes.count";
pub(crate) const POSTGRES_HASHES_BOUND: &str = "pict-rs.postgres.hashes.bound";
pub(crate) const POSTGRES_HASHES_ORDERED_HASH: &str = "pict-rs.postgres.hashes.ordered-hash";
pub(crate) const POSTGRES_HASHES_NEXT_HASHES: &str = "pict-rs.postgres.hashes.next-hashes";
pub(crate) const POSTGRES_HASHES_PREV_HASH: &str = "pict-rs.postgres.hashes.prev-hash";
pub(crate) const POSTGRES_HASHES_FIRST_HASHES: &str = "pict-rs.postgres.hashes.first-hashes";
pub(crate) const POSTGRES_HASHES_CREATE_HASH: &str = "pict-rs.postgres.hashes.create-hash";
pub(crate) const POSTGRES_HASHES_UPDATE_IDENTIFIER: &str =
"pict-rs.postgres.hashes.update-identifier";
pub(crate) const POSTGRES_HASHES_IDENTIFIER: &str = "pict-rs.postgres.identifier";
pub(crate) const POSTGRES_VARIANTS_RELATE_VARIANT_IDENTIFIER: &str =
"pict-rs.postgres.variants.relate-variant-identifier";
pub(crate) const POSTGRES_VARIANTS_IDENTIFIER: &str = "pict-rs.postgres.variants.identifier";
pub(crate) const POSTGRES_VARIANTS_FOR_HASH: &str = "pict-rs.postgres.variants.for-hash";
pub(crate) const POSTGRES_VARIANTS_REMOVE: &str = "pict-rs.postgres.variants.remove";
pub(crate) const POSTGRES_HASHES_RELATE_MOTION_IDENTIFIER: &str =
"pict-rs.postgres.hashes.relate-motion-identifier";
pub(crate) const POSTGRES_HASHES_MOTION_IDENTIFIER: &str =
"pict-rs.postgres.hashes.motion-identifier";
pub(crate) const POSTGRES_VARIANTS_CLEANUP: &str = "pict-rs.postgres.variants.cleanup";
pub(crate) const POSTGRES_HASHES_CLEANUP: &str = "pict-rs.postgres.hashes.cleanup";
pub(crate) const POSTGRES_ALIASES_CREATE: &str = "pict-rs.postgres.aliases.create";
pub(crate) const POSTGRES_ALIASES_DELETE_TOKEN: &str = "pict-rs.postgres.aliases.delete-token";
pub(crate) const POSTGRES_ALIASES_HASH: &str = "pict-rs.postgres.aliases.hash";
pub(crate) const POSTGRES_ALIASES_FOR_HASH: &str = "pict-rs.postgres.aliases.for-hash";
pub(crate) const POSTGRES_ALIASES_CLEANUP: &str = "pict-rs.postgres.aliases.cleanup";
pub(crate) const POSTGRES_SETTINGS_SET: &str = "pict-rs.postgres.settings.set";
pub(crate) const POSTGRES_SETTINGS_GET: &str = "pict-rs.postgres.settings.get";
pub(crate) const POSTGRES_SETTINGS_REMOVE: &str = "pict-rs.postgres.settings.remove";
pub(crate) const POSTGRES_DETAILS_RELATE: &str = "pict-rs.postgres.details.relate";
pub(crate) const POSTGRES_DETAILS_GET: &str = "pict-rs.postgres.details.get";
pub(crate) const POSTGRES_DETAILS_CLEANUP: &str = "pict-rs.postgres.details.cleanup";
pub(crate) const POSTGRES_QUEUE_COUNT: &str = "pict-rs.postgres.queue.count";
pub(crate) const POSTGRES_QUEUE_PUSH: &str = "pict-rs.postgres.queue.push";
pub(crate) const POSTGRES_QUEUE_LISTEN: &str = "pict-rs.postgres.queue.listen";
pub(crate) const POSTGRES_QUEUE_REQUEUE: &str = "pict-rs.postgres.queue.requeue";
pub(crate) const POSTGRES_QUEUE_CLAIM: &str = "pict-rs.postgres.queue.claim";
pub(crate) const POSTGRES_QUEUE_HEARTBEAT: &str = "pict-rs.postgres.queue.heartbeat";
pub(crate) const POSTGRES_QUEUE_COMPLETE: &str = "pict-rs.postgres.queue.complete";
pub(crate) const POSTGRES_STORE_MIGRATION_COUNT: &str = "pict-rs.postgres.store-migration.count";
pub(crate) const POSTGRES_STORE_MIGRATION_MARK_MIGRATED: &str =
"pict-rs.postgres.store-migration.mark-migrated";
pub(crate) const POSTGRES_STORE_MIGRATION_IS_MIGRATED: &str =
"pict-rs.postgres.store-migration.is-migrated";
pub(crate) const POSTGRES_STORE_MIGRATION_CLEAR: &str = "pict-rs.postgres.store-migration.clear";
pub(crate) const POSTGRES_PROXY_RELATE_URL: &str = "pict-rs.postgres.proxy.relate-url";
pub(crate) const POSTGRES_PROXY_RELATED: &str = "pict-rs.postgres.proxy.related";
pub(crate) const POSTGRES_PROXY_REMOVE_RELATION: &str = "pict-rs.postgres.proxy.remove-relation";
pub(crate) const POSTGRES_ALIAS_ACCESS_SET_ACCESSED: &str =
"pict-rs.postgres.alias-access.set-accessed";
pub(crate) const POSTGRES_ALIAS_ACCESS_ACCESSED_AT: &str =
"pict-rs.postgres.alias-access.accessed-at";
pub(crate) const POSTGRES_ALIAS_ACCESS_OLDER_ALIASES: &str =
"pict-rs.postgres.alias-access.older-aliases";
pub(crate) const POSTGRES_VARIANT_ACCESS_SET_ACCESSED: &str =
"pict-rs.postgres.variant-access.set-accessed";
pub(crate) const POSTGRES_VARIANT_ACCESS_ACCESSED_AT: &str =
"pict-rs.postgres.variant-access.accessed-at";
pub(crate) const POSTGRES_VARIANT_ACCESS_OLDER_VARIANTS: &str =
"pict-rs.postgres.variant-access.older-variants";
pub(crate) const POSTGRES_UPLOADS_CREATE: &str = "pict-rs.postgres.uploads.create";
pub(crate) const POSTGRES_UPLOADS_LISTEN: &str = "pict-rs.postgres.uploads.listen";
pub(crate) const POSTGRES_UPLOADS_WAIT: &str = "pict-rs.postgres.uploads.wait";
pub(crate) const POSTGRES_UPLOADS_CLAIM: &str = "pict-rs.postgres.uploads.claim";
pub(crate) const POSTGRES_UPLOADS_COMPLETE: &str = "pict-rs.postgres.uploads.complete";
fn describe_middleware() {
metrics::describe_counter!(
REQUEST_START,
"How many requests have been made to pict-rs, by requested path"
);
metrics::describe_counter!(
REQUEST_END,
"How many requests pict-rs has finished serving, by requested path"
);
metrics::describe_histogram!(
REQUEST_TIMINGS,
"How long pict-rs takes to serve requests, by requested path"
);
}
pub(crate) const REQUEST_START: &str = "pict-rs.request.start";
pub(crate) const REQUEST_END: &str = "pict-rs.request.end";
pub(crate) const REQUEST_TIMINGS: &str = "pict-rs.request.timings";
fn describe_generate() {
metrics::describe_counter!(
GENERATE_START,
"Counter describing how many times a variant has begun processing"
);
metrics::describe_histogram!(
GENERATE_DURATION,
"Timings for processing variants (i.e. generating thumbnails)"
);
metrics::describe_counter!(GENERATE_END, "Counter describing how many times a variant has finished processing, and whether it completed or aborted");
metrics::describe_histogram!(
GENERATE_PROCESS,
"Timings for processing media or waiting for media to be processed"
);
}
pub(crate) const GENERATE_START: &str = "pict-rs.generate.start";
pub(crate) const GENERATE_DURATION: &str = "pict-rs.generate.duration";
pub(crate) const GENERATE_END: &str = "pict-rs.generate.end";
pub(crate) const GENERATE_PROCESS: &str = "pict-rs.generate.process";
fn describe_object_storage() {
metrics::describe_histogram!(
OBJECT_STORAGE_HEAD_BUCKET_REQUEST,
"Timings for HEAD requests for the pict-rs Bucket in object storage"
);
metrics::describe_histogram!(
OBJECT_STORAGE_PUT_OBJECT_REQUEST,
"Timings for PUT requests for uploading media to object storage"
);
metrics::describe_histogram!(OBJECT_STORAGE_CREATE_MULTIPART_REQUEST, "Timings for creating a multipart request to object storage. This is the first step in uploading larger files.");
metrics::describe_histogram!(OBJECT_STORAGE_CREATE_UPLOAD_PART_REQUEST, "Timings for uploading part of a large file to object storage as a multipart part. This is one step in uploading larger files.");
metrics::describe_histogram!(
OBJECT_STORAGE_ABORT_MULTIPART_REQUEST,
"Timings for aborting a multipart upload to object storage"
);
metrics::describe_histogram!(
OBJECT_STORAGE_GET_OBJECT_REQUEST,
"Timings for requesting media from object storage"
);
metrics::describe_histogram!(
OBJECT_STORAGE_GET_OBJECT_REQUEST_STREAM,
"Timings for streaming an object from object storage"
);
metrics::describe_histogram!(
OBJECT_STORAGE_HEAD_OBJECT_REQUEST,
"Timings for requesting metadata for media from object storage"
);
metrics::describe_histogram!(
OBJECT_STORAGE_DELETE_OBJECT_REQUEST,
"Timings for requesting media in object storage be deleted"
);
metrics::describe_histogram!(
OBJECT_STORAGE_COMPLETE_MULTIPART_REQUEST,
"Timings for completing a multipart request to object storage"
);
}
pub(crate) const OBJECT_STORAGE_HEAD_BUCKET_REQUEST: &str =
"pict-rs.object-storage.head-bucket-request";
pub(crate) const OBJECT_STORAGE_PUT_OBJECT_REQUEST: &str =
"pict-rs.object-storage.put-object-request";
pub(crate) const OBJECT_STORAGE_CREATE_MULTIPART_REQUEST: &str =
"pict-rs.object-storage.create-multipart-request";
pub(crate) const OBJECT_STORAGE_CREATE_UPLOAD_PART_REQUEST: &str =
"pict-rs.object-storage.create-upload-part-request";
pub(crate) const OBJECT_STORAGE_ABORT_MULTIPART_REQUEST: &str =
"pict-rs.object-storage.abort-multipart-request";
pub(crate) const OBJECT_STORAGE_GET_OBJECT_REQUEST: &str =
"pict-rs.object-storage.get-object-request";
pub(crate) const OBJECT_STORAGE_GET_OBJECT_REQUEST_STREAM: &str =
"pict-rs.object-storage.get-object-request.stream";
pub(crate) const OBJECT_STORAGE_HEAD_OBJECT_REQUEST: &str =
"pict-rs.object-storage.head-object-request";
pub(crate) const OBJECT_STORAGE_DELETE_OBJECT_REQUEST: &str =
"pict-rs.object-storage.delete-object-request";
pub(crate) const OBJECT_STORAGE_COMPLETE_MULTIPART_REQUEST: &str =
"pict-rs.object-storage.complete-multipart-request";

19
src/lib.rs
View file

@ -14,6 +14,7 @@ mod formats;
mod future;
mod generate;
mod ingest;
mod init_metrics;
mod init_tracing;
mod magick;
mod middleware;
@ -172,7 +173,8 @@ impl<S: Store + 'static> FormData for Upload<S> {
Field::array(Field::file(move |filename, _, stream| {
let state = state.clone();
metrics::counter!("pict-rs.files", "upload" => "inline").increment(1);
metrics::counter!(crate::init_metrics::FILES, "upload" => "inline")
.increment(1);
let span = tracing::info_span!("file-upload", ?filename);
@ -221,7 +223,8 @@ impl<S: Store + 'static> FormData for Import<S> {
Field::array(Field::file(move |filename, _, stream| {
let state = state.clone();
metrics::counter!("pict-rs.files", "import" => "inline").increment(1);
metrics::counter!(crate::init_metrics::FILES, "import" => "inline")
.increment(1);
let span = tracing::info_span!("file-import", ?filename);
@ -335,7 +338,8 @@ impl<S: Store + 'static> FormData for BackgroundedUpload<S> {
Field::array(Field::file(move |filename, _, stream| {
let state = state.clone();
metrics::counter!("pict-rs.files", "upload" => "background").increment(1);
metrics::counter!(crate::init_metrics::FILES, "upload" => "background")
.increment(1);
let span = tracing::info_span!("file-proxy", ?filename);
@ -423,7 +427,7 @@ async fn claim_upload<S: Store + 'static>(
Ok(wait_res) => {
let upload_result = wait_res?;
state.repo.claim(upload_id).await?;
metrics::counter!("pict-rs.background.upload.claim").increment(1);
metrics::counter!(crate::init_metrics::BACKGROUND_UPLOAD_CLAIM).increment(1);
match upload_result {
UploadResult::Success { alias, token } => {
@ -514,7 +518,7 @@ async fn do_download_inline<S: Store + 'static>(
stream: impl Stream<Item = Result<web::Bytes, Error>> + 'static,
state: &State<S>,
) -> Result<HttpResponse, Error> {
metrics::counter!("pict-rs.files", "download" => "inline").increment(1);
metrics::counter!(crate::init_metrics::FILES, "download" => "inline").increment(1);
let (alias, delete_token, details) = ingest_inline(stream, state).await?;
@ -533,7 +537,7 @@ async fn do_download_backgrounded<S: Store + 'static>(
stream: impl Stream<Item = Result<web::Bytes, Error>> + 'static,
state: web::Data<State<S>>,
) -> Result<HttpResponse, Error> {
metrics::counter!("pict-rs.files", "download" => "background").increment(1);
metrics::counter!(crate::init_metrics::FILES, "download" => "background").increment(1);
let backgrounded = Backgrounded::proxy(&state, stream).await?;
@ -1962,6 +1966,9 @@ impl PictRsConfiguration {
pub async fn run(self) -> color_eyre::Result<()> {
let PictRsConfiguration { config, operation } = self;
// describe all the metrics pict-rs produces
init_metrics::init_metrics();
let tmp_dir = TmpDir::init(&config.server.temporary_directory).await?;
let policy_dir = magick::write_magick_policy(&config.media, &tmp_dir).await?;

11
src/middleware/metrics.rs
View file

@ -26,7 +26,7 @@ struct MetricsGuardWithStatus {
impl MetricsGuard {
fn new(matched_path: Option<String>) -> Self {
metrics::counter!("pict-rs.request.start", "path" => format!("{matched_path:?}"))
metrics::counter!(crate::init_metrics::REQUEST_START, "path" => format!("{matched_path:?}"))
.increment(1);
Self {
@ -50,16 +50,17 @@ impl MetricsGuard {
impl Drop for MetricsGuard {
fn drop(&mut self) {
if self.armed {
metrics::counter!("pict-rs.request.complete", "path" => format!("{:?}", self.matched_path)).increment(1);
metrics::histogram!("pict-rs.request.timings", "path" => format!("{:?}", self.matched_path)).record(self.start.elapsed().as_secs_f64());
metrics::counter!(crate::init_metrics::REQUEST_END, "path" => format!("{:?}", self.matched_path))
.increment(1);
metrics::histogram!(crate::init_metrics::REQUEST_TIMINGS, "path" => format!("{:?}", self.matched_path)).record(self.start.elapsed().as_secs_f64());
}
}
}
impl Drop for MetricsGuardWithStatus {
fn drop(&mut self) {
metrics::counter!("pict-rs.request.complete", "path" => format!("{:?}", self.matched_path), "status" => self.status.to_string()).increment(1);
metrics::histogram!("pict-rs.request.timings", "path" => format!("{:?}", self.matched_path), "status" => self.status.to_string()).record(self.start.elapsed().as_secs_f64());
metrics::counter!(crate::init_metrics::REQUEST_END, "path" => format!("{:?}", self.matched_path), "status" => self.status.to_string()).increment(1);
metrics::histogram!(crate::init_metrics::REQUEST_TIMINGS, "path" => format!("{:?}", self.matched_path), "status" => self.status.to_string()).record(self.start.elapsed().as_secs_f64());
}
}

8
src/middleware/payload.rs
View file

@ -22,7 +22,7 @@ struct MetricsGuard {
impl MetricsGuard {
fn guard() -> Self {
metrics::counter!("pict-rs.payload.drain.start").increment(1);
metrics::counter!(crate::init_metrics::PAYLOAD_DRAIN_START).increment(1);
MetricsGuard {
start: Instant::now(),
@ -37,10 +37,10 @@ impl MetricsGuard {
impl Drop for MetricsGuard {
fn drop(&mut self) {
metrics::counter!("pict-rs.payload.drain.end", "completed" => (!self.armed).to_string())
metrics::counter!(crate::init_metrics::PAYLOAD_DRAIN_END, "completed" => (!self.armed).to_string())
.increment(1);
metrics::histogram!("pict-rs.payload.drain.duration", "completed" => (!self.armed).to_string())
metrics::histogram!(crate::init_metrics::PAYLOAD_DRAIN_DURATION, "completed" => (!self.armed).to_string())
.record(self.start.elapsed().as_secs_f64());
}
}
@ -137,7 +137,7 @@ impl Drop for PayloadStream {
if let Some(payload) = self.inner.take() {
tracing::debug!("Dropped unclosed payload, draining");
if self.sender.try_send(payload).is_err() {
metrics::counter!("pict-rs.payload.drain.fail-send").increment(1);
metrics::counter!(crate::init_metrics::PAYLOAD_DRAIN_FAIL_SEND).increment(1);
tracing::error!("Failed to send unclosed payload for draining");
}
}

7
src/process.rs
View file

@ -27,7 +27,8 @@ struct MetricsGuard {
impl MetricsGuard {
fn guard(command: Arc<str>) -> Self {
metrics::counter!("pict-rs.process.start", "command" => command.to_string()).increment(1);
metrics::counter!(crate::init_metrics::PROCESS_START, "command" => command.to_string())
.increment(1);
Self {
start: Instant::now(),
@ -44,13 +45,13 @@ impl MetricsGuard {
impl Drop for MetricsGuard {
fn drop(&mut self) {
metrics::histogram!(
"pict-rs.process.duration",
crate::init_metrics::PROCESS_DURATION,
"command" => self.command.to_string(),
"completed" => (!self.armed).to_string(),
)
.record(self.start.elapsed().as_secs_f64());
metrics::counter!("pict-rs.process.end", "completed" => (!self.armed).to_string() , "command" => self.command.to_string()).increment(1);
metrics::counter!(crate::init_metrics::PROCESS_END, "completed" => (!self.armed).to_string() , "command" => self.command.to_string()).increment(1);
}
}

6
src/queue.rs
View file

@ -203,7 +203,7 @@ struct MetricsGuard {
impl MetricsGuard {
fn guard(worker_id: uuid::Uuid, queue: &'static str) -> Self {
metrics::counter!("pict-rs.job.start", "queue" => queue, "worker-id" => worker_id.to_string()).increment(1);
metrics::counter!(crate::init_metrics::JOB_START, "queue" => queue, "worker-id" => worker_id.to_string()).increment(1);
Self {
worker_id,
@ -220,8 +220,8 @@ impl MetricsGuard {
impl Drop for MetricsGuard {
fn drop(&mut self) {
metrics::histogram!("pict-rs.job.duration", "queue" => self.queue, "worker-id" => self.worker_id.to_string(), "completed" => (!self.armed).to_string()).record(self.start.elapsed().as_secs_f64());
metrics::counter!("pict-rs.job.end", "queue" => self.queue, "worker-id" => self.worker_id.to_string(), "completed" => (!self.armed).to_string()).increment(1);
metrics::histogram!(crate::init_metrics::JOB_DURAION, "queue" => self.queue, "worker-id" => self.worker_id.to_string(), "completed" => (!self.armed).to_string()).record(self.start.elapsed().as_secs_f64());
metrics::counter!(crate::init_metrics::JOB_END, "queue" => self.queue, "worker-id" => self.worker_id.to_string(), "completed" => (!self.armed).to_string()).increment(1);
}
}

4
src/queue/cleanup.rs
View file

@ -166,7 +166,7 @@ async fn outdated_variants(repo: &ArcRepo, config: &Configuration) -> Result<(),
let mut count = 0;
while let Some(res) = variant_stream.next().await {
metrics::counter!("pict-rs.cleanup.outdated-variant").increment(1);
metrics::counter!(crate::init_metrics::CLEANUP_OUTDATED_VARIANT).increment(1);
tracing::trace!("outdated_variants: looping");
let (hash, variant) = res?;
@ -193,7 +193,7 @@ async fn outdated_proxies(repo: &ArcRepo, config: &Configuration) -> Result<(),
let mut count = 0;
while let Some(res) = alias_stream.next().await {
metrics::counter!("pict-rs.cleanup.outdated-proxy").increment(1);
metrics::counter!(crate::init_metrics::CLEANUP_OUTDATED_PROXY).increment(1);
tracing::trace!("outdated_proxies: looping");
let alias = res?;

4
src/queue/process.rs
View file

@ -87,8 +87,8 @@ impl UploadGuard {
impl Drop for UploadGuard {
fn drop(&mut self) {
metrics::counter!("pict-rs.background.upload.ingest", "completed" => (!self.armed).to_string()).increment(1);
metrics::histogram!("pict-rs.background.upload.ingest.duration", "completed" => (!self.armed).to_string()).record(self.start.elapsed().as_seconds_f64());
metrics::counter!(crate::init_metrics::BACKGROUND_UPLOAD_INGEST, "completed" => (!self.armed).to_string()).increment(1);
metrics::histogram!(crate::init_metrics::BACKGROUND_UPLOAD_INGEST_DURATION, "completed" => (!self.armed).to_string()).record(self.start.elapsed().as_seconds_f64());
if self.armed {
tracing::warn!(

10
src/repo/metrics.rs
View file

@ -55,21 +55,21 @@ impl WaitMetricsGuard {
impl Drop for PushMetricsGuard {
fn drop(&mut self) {
metrics::counter!("pict-rs.queue.push", "completed" => (!self.armed).to_string(), "queue" => self.queue).increment(1);
metrics::counter!(crate::init_metrics::QUEUE_PUSH, "completed" => (!self.armed).to_string(), "queue" => self.queue).increment(1);
}
}
impl Drop for PopMetricsGuard {
fn drop(&mut self) {
metrics::histogram!("pict-rs.queue.pop.duration", "completed" => (!self.armed).to_string(), "queue" => self.queue).record(self.start.elapsed().as_secs_f64());
metrics::counter!("pict-rs.queue.pop", "completed" => (!self.armed).to_string(), "queue" => self.queue).increment(1);
metrics::histogram!(crate::init_metrics::QUEUE_POP_DURATION, "completed" => (!self.armed).to_string(), "queue" => self.queue).record(self.start.elapsed().as_secs_f64());
metrics::counter!(crate::init_metrics::QUEUE_POP, "completed" => (!self.armed).to_string(), "queue" => self.queue).increment(1);
}
}
impl Drop for WaitMetricsGuard {
fn drop(&mut self) {
metrics::histogram!("pict-rs.upload.wait.duration", "completed" => (!self.armed).to_string()).record(self.start.elapsed().as_secs_f64());
metrics::counter!("pict-rs.upload.wait", "completed" => (!self.armed).to_string())
metrics::histogram!(crate::init_metrics::UPLOAD_WAIT_DURATION, "completed" => (!self.armed).to_string()).record(self.start.elapsed().as_secs_f64());
metrics::counter!(crate::init_metrics::UPLOAD_WAIT, "completed" => (!self.armed).to_string())
.increment(1);
}
}

120
src/repo/postgres.rs
View file

@ -253,11 +253,11 @@ fn build_pool(
.create_timeout(Some(Duration::from_secs(2)))
.recycle_timeout(Some(Duration::from_secs(2)))
.post_create(Hook::sync_fn(|_, _| {
metrics::counter!("pict-rs.postgres.pool.connection.create").increment(1);
metrics::counter!(crate::init_metrics::POSTGRES_POOL_CONNECTION_CREATE).increment(1);
Ok(())
}))
.post_recycle(Hook::sync_fn(|_, _| {
metrics::counter!("pict-rs.postgres.pool.connection.recycle").increment(1);
metrics::counter!(crate::init_metrics::POSTGRES_POOL_CONNECTION_RECYCLE).increment(1);
Ok(())
}))
.max_size(max_size)
@ -355,9 +355,9 @@ impl GetConnectionMetricsGuard {
impl Drop for GetConnectionMetricsGuard {
fn drop(&mut self) {
metrics::counter!("pict-rs.postgres.pool.get", "completed" => (!self.armed).to_string())
metrics::counter!(crate::init_metrics::POSTGRES_POOL_GET, "completed" => (!self.armed).to_string())
.increment(1);
metrics::histogram!("pict-rs.postgres.pool.get.duration", "completed" => (!self.armed).to_string()).record(self.start.elapsed().as_secs_f64());
metrics::histogram!(crate::init_metrics::POSTGRES_POOL_GET_DURATION, "completed" => (!self.armed).to_string()).record(self.start.elapsed().as_secs_f64());
}
}
@ -454,7 +454,7 @@ impl<'a> JobNotifierState<'a> {
.or_insert_with(crate::sync::notify)
.notify_one();
metrics::counter!("pict-rs.postgres.job-notifier.notified", "queue" => queue_name.to_string()).increment(1);
metrics::counter!(crate::init_metrics::POSTGRES_JOB_NOTIFIER_NOTIFIED, "queue" => queue_name.to_string()).increment(1);
}
}
@ -472,7 +472,7 @@ impl<'a> UploadNotifierState<'a> {
.and_then(|weak| weak.upgrade())
{
notifier.notify_waiters();
metrics::counter!("pict-rs.postgres.upload-notifier.notified").increment(1);
metrics::counter!(crate::init_metrics::POSTGRES_UPLOAD_NOTIFIER_NOTIFIED).increment(1);
}
}
}
@ -497,7 +497,7 @@ async fn delegate_notifications(
while let Ok(notification) = receiver.recv_async().await {
tracing::trace!("delegate_notifications: looping");
metrics::counter!("pict-rs.postgres.notification").increment(1);
metrics::counter!(crate::init_metrics::POSTGRES_NOTIFICATION).increment(1);
match notification.channel() {
"queue_status_channel" => {
@ -611,7 +611,7 @@ impl HashRepo for PostgresRepo {
let count = hashes
.count()
.get_result::<i64>(&mut conn)
.with_metrics("pict-rs.postgres.hashes.count")
.with_metrics(crate::init_metrics::POSTGRES_HASHES_COUNT)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -630,7 +630,7 @@ impl HashRepo for PostgresRepo {
.select(created_at)
.filter(hash.eq(&input_hash))
.get_result(&mut conn)
.with_metrics("pict-rs.postgres.hashes.bound")
.with_metrics(crate::init_metrics::POSTGRES_HASHES_BOUND)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -661,7 +661,7 @@ impl HashRepo for PostgresRepo {
.filter(created_at.lt(timestamp))
.order(created_at.desc())
.get_result::<(time::PrimitiveDateTime, Hash)>(&mut conn)
.with_metrics("pict-rs.postgres.hashes.ordered-hash")
.with_metrics(crate::init_metrics::POSTGRES_HASHES_ORDERED_HASH)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -700,7 +700,7 @@ impl HashRepo for PostgresRepo {
.then_order_by(hash.desc())
.limit(limit as i64 + 1)
.get_results::<Hash>(&mut conn)
.with_metrics("pict-rs.postgres.hashes.next-hashes")
.with_metrics(crate::init_metrics::POSTGRES_HASHES_NEXT_HASHES)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -714,7 +714,7 @@ impl HashRepo for PostgresRepo {
.then_order_by(hash)
.limit(limit as i64)
.get_results::<Hash>(&mut conn)
.with_metrics("pict-rs.postgres.hashes.prev-hashes")
.with_metrics(crate::init_metrics::POSTGRES_HASHES_PREV_HASH)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -729,7 +729,7 @@ impl HashRepo for PostgresRepo {
.then_order_by(hash.desc())
.limit(limit as i64 + 1)
.get_results::<Hash>(&mut conn)
.with_metrics("pict-rs.postgres.hashes.first-hashes")
.with_metrics(crate::init_metrics::POSTGRES_HASHES_FIRST_HASHES)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -768,7 +768,7 @@ impl HashRepo for PostgresRepo {
created_at.eq(&timestamp),
))
.execute(&mut conn)
.with_metrics("pict-rs.postgres.hashes.create-hash")
.with_metrics(crate::init_metrics::POSTGRES_HASHES_CREATE_HASH)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?;
@ -797,7 +797,7 @@ impl HashRepo for PostgresRepo {
.filter(hash.eq(&input_hash))
.set(identifier.eq(input_identifier.as_ref()))
.execute(&mut conn)
.with_metrics("pict-rs.postgres.hashes.update-identifier")
.with_metrics(crate::init_metrics::POSTGRES_HASHES_UPDATE_IDENTIFIER)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -816,7 +816,7 @@ impl HashRepo for PostgresRepo {
.select(identifier)
.filter(hash.eq(&input_hash))
.get_result::<String>(&mut conn)
.with_metrics("pict-rs.postgres.hashes.identifier")
.with_metrics(crate::init_metrics::POSTGRES_HASHES_IDENTIFIER)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -844,7 +844,7 @@ impl HashRepo for PostgresRepo {
identifier.eq(input_identifier.as_ref()),
))
.execute(&mut conn)
.with_metrics("pict-rs.postgres.variants.relate-variant-identifier")
.with_metrics(crate::init_metrics::POSTGRES_VARIANTS_RELATE_VARIANT_IDENTIFIER)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?;
@ -874,7 +874,7 @@ impl HashRepo for PostgresRepo {
.filter(hash.eq(&input_hash))
.filter(variant.eq(&input_variant))
.get_result::<String>(&mut conn)
.with_metrics("pict-rs.postgres.variants.identifier")
.with_metrics(crate::init_metrics::POSTGRES_VARIANTS_IDENTIFIER)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -895,7 +895,7 @@ impl HashRepo for PostgresRepo {
.select((variant, identifier))
.filter(hash.eq(&input_hash))
.get_results::<(String, String)>(&mut conn)
.with_metrics("pict-rs.postgres.variants.for-hash")
.with_metrics(crate::init_metrics::POSTGRES_VARIANTS_FOR_HASH)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -921,7 +921,7 @@ impl HashRepo for PostgresRepo {
.filter(hash.eq(&input_hash))
.filter(variant.eq(&input_variant))
.execute(&mut conn)
.with_metrics("pict-rs.postgres.variants.remove")
.with_metrics(crate::init_metrics::POSTGRES_VARIANTS_REMOVE)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -944,7 +944,7 @@ impl HashRepo for PostgresRepo {
.filter(hash.eq(&input_hash))
.set(motion_identifier.eq(input_identifier.as_ref()))
.execute(&mut conn)
.with_metrics("pict-rs.postgres.hashes.relate-motion-identifier")
.with_metrics(crate::init_metrics::POSTGRES_HASHES_RELATE_MOTION_IDENTIFIER)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -963,7 +963,7 @@ impl HashRepo for PostgresRepo {
.select(motion_identifier)
.filter(hash.eq(&input_hash))
.get_result::<Option<String>>(&mut conn)
.with_metrics("pict-rs.postgres.hashes.motion-identifier")
.with_metrics(crate::init_metrics::POSTGRES_HASHES_MOTION_IDENTIFIER)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -984,13 +984,13 @@ impl HashRepo for PostgresRepo {
diesel::delete(schema::variants::dsl::variants)
.filter(schema::variants::dsl::hash.eq(&input_hash))
.execute(conn)
.with_metrics("pict-rs.postgres.variants.cleanup")
.with_metrics(crate::init_metrics::POSTGRES_VARIANTS_CLEANUP)
.await?;
diesel::delete(schema::hashes::dsl::hashes)
.filter(schema::hashes::dsl::hash.eq(&input_hash))
.execute(conn)
.with_metrics("pict-rs.postgres.hashes.cleanup")
.with_metrics(crate::init_metrics::POSTGRES_HASHES_CLEANUP)
.await
})
})
@ -1021,7 +1021,7 @@ impl AliasRepo for PostgresRepo {
token.eq(delete_token),
))
.execute(&mut conn)
.with_metrics("pict-rs.postgres.aliases.create")
.with_metrics(crate::init_metrics::POSTGRES_ALIASES_CREATE)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?;
@ -1046,7 +1046,7 @@ impl AliasRepo for PostgresRepo {
.select(token)
.filter(alias.eq(input_alias))
.get_result(&mut conn)
.with_metrics("pict-rs.postgres.aliases.delete-token")
.with_metrics(crate::init_metrics::POSTGRES_ALIASES_DELETE_TOKEN)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -1066,7 +1066,7 @@ impl AliasRepo for PostgresRepo {
.select(hash)
.filter(alias.eq(input_alias))
.get_result(&mut conn)
.with_metrics("pict-rs.postgres.aliases.hash")
.with_metrics(crate::init_metrics::POSTGRES_ALIASES_HASH)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -1086,7 +1086,7 @@ impl AliasRepo for PostgresRepo {
.select(alias)
.filter(hash.eq(&input_hash))
.get_results(&mut conn)
.with_metrics("pict-rs.postgres.aliases.for-hash")
.with_metrics(crate::init_metrics::POSTGRES_ALIASES_FOR_HASH)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -1104,7 +1104,7 @@ impl AliasRepo for PostgresRepo {
diesel::delete(aliases)
.filter(alias.eq(input_alias))
.execute(&mut conn)
.with_metrics("pict-rs.postgres.aliases.cleanup")
.with_metrics(crate::init_metrics::POSTGRES_ALIASES_CLEANUP)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -1130,7 +1130,7 @@ impl SettingsRepo for PostgresRepo {
.do_update()
.set(value.eq(&input_value))
.execute(&mut conn)
.with_metrics("pict-rs.postgres.settings.set")
.with_metrics(crate::init_metrics::POSTGRES_SETTINGS_SET)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -1149,7 +1149,7 @@ impl SettingsRepo for PostgresRepo {
.select(value)
.filter(key.eq(input_key))
.get_result::<String>(&mut conn)
.with_metrics("pict-rs.postgres.settings.get")
.with_metrics(crate::init_metrics::POSTGRES_SETTINGS_GET)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -1172,7 +1172,7 @@ impl SettingsRepo for PostgresRepo {
diesel::delete(settings)
.filter(key.eq(input_key))
.execute(&mut conn)
.with_metrics("pict-rs.postgres.settings.remove")
.with_metrics(crate::init_metrics::POSTGRES_SETTINGS_REMOVE)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -1200,7 +1200,7 @@ impl DetailsRepo for PostgresRepo {
diesel::insert_into(details)
.values((identifier.eq(input_identifier.as_ref()), json.eq(&value)))
.execute(&mut conn)
.with_metrics("pict-rs.postgres.details.relate")
.with_metrics(crate::init_metrics::POSTGRES_DETAILS_RELATE)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -1219,7 +1219,7 @@ impl DetailsRepo for PostgresRepo {
.select(json)
.filter(identifier.eq(input_identifier.as_ref()))
.get_result::<serde_json::Value>(&mut conn)
.with_metrics("pict-rs.postgres.details.get")
.with_metrics(crate::init_metrics::POSTGRES_DETAILS_GET)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -1242,7 +1242,7 @@ impl DetailsRepo for PostgresRepo {
diesel::delete(details)
.filter(identifier.eq(input_identifier.as_ref()))
.execute(&mut conn)
.with_metrics("pict-rs.postgres.details.cleanup")
.with_metrics(crate::init_metrics::POSTGRES_DETAILS_CLEANUP)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -1262,7 +1262,7 @@ impl QueueRepo for PostgresRepo {
let count = job_queue
.count()
.get_result::<i64>(&mut conn)
.with_metrics("pict-rs.postgres.job_queue.count")
.with_metrics(crate::init_metrics::POSTGRES_QUEUE_COUNT)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -1292,7 +1292,7 @@ impl QueueRepo for PostgresRepo {
))
.returning(id)
.get_result::<Uuid>(&mut conn)
.with_metrics("pict-rs.postgres.queue.push")
.with_metrics(crate::init_metrics::POSTGRES_QUEUE_PUSH)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -1337,7 +1337,7 @@ impl QueueRepo for PostgresRepo {
diesel::sql_query("LISTEN queue_status_channel;")
.execute(&mut notifier_conn)
.with_metrics("pict-rs.postgres.queue.listen")
.with_metrics(crate::init_metrics::POSTGRES_QUEUE_LISTEN)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -1357,7 +1357,7 @@ impl QueueRepo for PostgresRepo {
worker.eq(Option::<Uuid>::None),
))
.execute(&mut conn)
.with_metrics("pict-rs.postgres.queue.requeue")
.with_metrics(crate::init_metrics::POSTGRES_QUEUE_REQUEUE)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -1392,7 +1392,7 @@ impl QueueRepo for PostgresRepo {
))
.returning((id, job))
.get_result(&mut conn)
.with_metrics("pict-rs.postgres.queue.claim")
.with_metrics(crate::init_metrics::POSTGRES_QUEUE_CLAIM)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -1439,7 +1439,7 @@ impl QueueRepo for PostgresRepo {
)
.set(heartbeat.eq(timestamp))
.execute(&mut conn)
.with_metrics("pict-rs.postgres.queue.heartbeat")
.with_metrics(crate::init_metrics::POSTGRES_QUEUE_HEARTBEAT)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -1466,7 +1466,7 @@ impl QueueRepo for PostgresRepo {
.and(worker.eq(worker_id)),
)
.execute(&mut conn)
.with_metrics("pict-rs.postgres.queue.complete")
.with_metrics(crate::init_metrics::POSTGRES_QUEUE_COMPLETE)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -1487,7 +1487,7 @@ impl StoreMigrationRepo for PostgresRepo {
let count = store_migrations
.count()
.get_result::<i64>(&mut conn)
.with_metrics("pict-rs.postgres.store-migration.count")
.with_metrics(crate::init_metrics::POSTGRES_STORE_MIGRATION_COUNT)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -1514,7 +1514,7 @@ impl StoreMigrationRepo for PostgresRepo {
.on_conflict(old_identifier)
.do_nothing()
.execute(&mut conn)
.with_metrics("pict-rs.postgres.store-migration.mark-migrated")
.with_metrics(crate::init_metrics::POSTGRES_STORE_MIGRATION_MARK_MIGRATED)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -1533,7 +1533,7 @@ impl StoreMigrationRepo for PostgresRepo {
store_migrations.filter(old_identifier.eq(input_old_identifier.as_ref())),
))
.get_result(&mut conn)
.with_metrics("pict-rs.postgres.store-migration.is-migrated")
.with_metrics(crate::init_metrics::POSTGRES_STORE_MIGRATION_IS_MIGRATED)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -1550,7 +1550,7 @@ impl StoreMigrationRepo for PostgresRepo {
diesel::delete(store_migrations)
.execute(&mut conn)
.with_metrics("pict-rs.postgres.store-migration.clear")
.with_metrics(crate::init_metrics::POSTGRES_STORE_MIGRATION_CLEAR)
.with_timeout(Duration::from_secs(20))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -1571,7 +1571,7 @@ impl ProxyRepo for PostgresRepo {
diesel::insert_into(proxies)
.values((url.eq(input_url.as_str()), alias.eq(&input_alias)))
.execute(&mut conn)
.with_metrics("pict-rs.postgres.proxy.relate-url")
.with_metrics(crate::init_metrics::POSTGRES_PROXY_RELATE_URL)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -1590,7 +1590,7 @@ impl ProxyRepo for PostgresRepo {
.select(alias)
.filter(url.eq(input_url.as_str()))
.get_result(&mut conn)
.with_metrics("pict-rs.postgres.proxy.related")
.with_metrics(crate::init_metrics::POSTGRES_PROXY_RELATED)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -1609,7 +1609,7 @@ impl ProxyRepo for PostgresRepo {
diesel::delete(proxies)
.filter(alias.eq(&input_alias))
.execute(&mut conn)
.with_metrics("pict-rs.postgres.proxy.remove-relation")
.with_metrics(crate::init_metrics::POSTGRES_PROXY_REMOVE_RELATION)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -1637,7 +1637,7 @@ impl AliasAccessRepo for PostgresRepo {
.filter(alias.eq(&input_alias))
.set(accessed.eq(timestamp))
.execute(&mut conn)
.with_metrics("pict-rs.postgres.alias-access.set-accessed")
.with_metrics(crate::init_metrics::POSTGRES_ALIAS_ACCESS_SET_ACCESSED)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -1659,7 +1659,7 @@ impl AliasAccessRepo for PostgresRepo {
.select(accessed)
.filter(alias.eq(&input_alias))
.get_result::<time::PrimitiveDateTime>(&mut conn)
.with_metrics("pict-rs.postgres.alias-access.accessed-at")
.with_metrics(crate::init_metrics::POSTGRES_ALIAS_ACCESS_ACCESSED_AT)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -1689,7 +1689,7 @@ impl AliasAccessRepo for PostgresRepo {
.order(accessed.desc())
.limit(100)
.get_results(&mut conn)
.with_metrics("pict-rs.postgres.alias-access.older-aliases")
.with_metrics(crate::init_metrics::POSTGRES_ALIAS_ACCESS_OLDER_ALIASES)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -1725,7 +1725,7 @@ impl VariantAccessRepo for PostgresRepo {
.filter(hash.eq(&input_hash).and(variant.eq(&input_variant)))
.set(accessed.eq(timestamp))
.execute(&mut conn)
.with_metrics("pict-rs.postgres.variant-access.set-accessed")
.with_metrics(crate::init_metrics::POSTGRES_VARIANT_ACCESS_SET_ACCESSED)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -1748,7 +1748,7 @@ impl VariantAccessRepo for PostgresRepo {
.select(accessed)
.filter(hash.eq(&input_hash).and(variant.eq(&input_variant)))
.get_result(&mut conn)
.with_metrics("pict-rs.postgres.variant-access.accessed-at")
.with_metrics(crate::init_metrics::POSTGRES_VARIANT_ACCESS_ACCESSED_AT)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -1778,7 +1778,7 @@ impl VariantAccessRepo for PostgresRepo {
.order(accessed.desc())
.limit(100)
.get_results(&mut conn)
.with_metrics("pict-rs.postgres.variant-access.older-variants")
.with_metrics(crate::init_metrics::POSTGRES_VARIANT_ACCESS_OLDER_VARIANTS)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -1843,7 +1843,7 @@ impl UploadRepo for PostgresRepo {
.default_values()
.returning(id)
.get_result(&mut conn)
.with_metrics("pict-rs.postgres.uploads.create")
.with_metrics(crate::init_metrics::POSTGRES_UPLOADS_CREATE)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -1868,7 +1868,7 @@ impl UploadRepo for PostgresRepo {
diesel::sql_query("LISTEN upload_completion_channel;")
.execute(&mut notifier_conn)
.with_metrics("pict-rs.postgres.uploads.listen")
.with_metrics(crate::init_metrics::POSTGRES_UPLOADS_LISTEN)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -1882,7 +1882,7 @@ impl UploadRepo for PostgresRepo {
.select(result)
.filter(id.eq(upload_id.id))
.get_result(&mut conn)
.with_metrics("pict-rs.postgres.uploads.wait")
.with_metrics(crate::init_metrics::POSTGRES_UPLOADS_WAIT)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -1924,7 +1924,7 @@ impl UploadRepo for PostgresRepo {
diesel::delete(uploads)
.filter(id.eq(upload_id.id))
.execute(&mut conn)
.with_metrics("pict-rs.postgres.uploads.claim")
.with_metrics(crate::init_metrics::POSTGRES_UPLOADS_CLAIM)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?
@ -1951,7 +1951,7 @@ impl UploadRepo for PostgresRepo {
.filter(id.eq(upload_id.id))
.set(result.eq(upload_result))
.execute(&mut conn)
.with_metrics("pict-rs.postgres.uploads.complete")
.with_metrics(crate::init_metrics::POSTGRES_UPLOADS_COMPLETE)
.with_timeout(Duration::from_secs(5))
.await
.map_err(|_| PostgresError::DbTimeout)?

28
src/store/object_store.rs
View file

@ -206,7 +206,7 @@ impl Store for ObjectStore {
.head_bucket_request()
.await?
.send()
.with_metrics("pict-rs.object-storage.head-bucket-request")
.with_metrics(crate::init_metrics::OBJECT_STORAGE_HEAD_BUCKET_REQUEST)
.await
.map_err(ObjectError::from)?;
@ -248,7 +248,7 @@ impl Store for ObjectStore {
let response = req
.body(Body::wrap_stream(first_chunk))
.send()
.with_metrics("pict-rs.object-store.put-object-request")
.with_metrics(crate::init_metrics::OBJECT_STORAGE_PUT_OBJECT_REQUEST)
.await
.map_err(ObjectError::from)?;
@ -264,7 +264,7 @@ impl Store for ObjectStore {
let (req, object_id) = self.create_multipart_request(content_type).await?;
let response = req
.send()
.with_metrics("pict-rs.object-store.create-multipart-request")
.with_metrics(crate::init_metrics::OBJECT_STORAGE_CREATE_MULTIPART_REQUEST)
.await
.map_err(ObjectError::from)?;
@ -314,7 +314,9 @@ impl Store for ObjectStore {
.await?
.body(Body::wrap_stream(buf))
.send()
.with_metrics("pict-rs.object-storage.create-upload-part-request")
.with_metrics(
crate::init_metrics::OBJECT_STORAGE_CREATE_UPLOAD_PART_REQUEST,
)
.await
.map_err(ObjectError::from)?;
@ -370,7 +372,7 @@ impl Store for ObjectStore {
if let Err(e) = res {
self.create_abort_multipart_request(&object_id, upload_id)
.send()
.with_metrics("pict-rs.object-storage.abort-multipart-request")
.with_metrics(crate::init_metrics::OBJECT_STORAGE_ABORT_MULTIPART_REQUEST)
.await
.map_err(ObjectError::from)?;
return Err(e);
@ -390,7 +392,7 @@ impl Store for ObjectStore {
let response = req
.body(bytes)
.send()
.with_metrics("pict-rs.object-storage.put-object-request")
.with_metrics(crate::init_metrics::OBJECT_STORAGE_PUT_OBJECT_REQUEST)
.await
.map_err(ObjectError::from)?;
@ -422,7 +424,7 @@ impl Store for ObjectStore {
let response = self
.get_object_request(identifier, from_start, len)
.send()
.with_metrics("pict-rs.object-storage.get-object-request")
.with_metrics(crate::init_metrics::OBJECT_STORAGE_GET_OBJECT_REQUEST)
.await
.map_err(ObjectError::from)?;
@ -431,7 +433,7 @@ impl Store for ObjectStore {
}
Ok(Box::pin(crate::stream::metrics(
"pict-rs.object-storage.get-object-request.stream",
crate::init_metrics::OBJECT_STORAGE_GET_OBJECT_REQUEST_STREAM,
crate::stream::map_err(response.bytes_stream(), payload_to_io_error),
)))
}
@ -448,7 +450,7 @@ impl Store for ObjectStore {
let response = self
.get_object_request(identifier, None, None)
.send()
.with_metrics("pict-rs.object-storage.get-object-request")
.with_metrics(crate::init_metrics::OBJECT_STORAGE_GET_OBJECT_REQUEST)
.await
.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, ObjectError::from(e)))?;
@ -460,7 +462,7 @@ impl Store for ObjectStore {
}
let stream = std::pin::pin!(crate::stream::metrics(
"pict-rs.object-storage.get-object-request.stream",
crate::init_metrics::OBJECT_STORAGE_GET_OBJECT_REQUEST_STREAM,
response.bytes_stream()
));
let mut stream = stream.into_streamer();
@ -481,7 +483,7 @@ impl Store for ObjectStore {
let response = self
.head_object_request(identifier)
.send()
.with_metrics("pict-rs.object-storage.head-object-request")
.with_metrics(crate::init_metrics::OBJECT_STORAGE_HEAD_OBJECT_REQUEST)
.await
.map_err(ObjectError::from)?;
@ -506,7 +508,7 @@ impl Store for ObjectStore {
let response = self
.delete_object_request(identifier)
.send()
.with_metrics("pict-rs.object-storage.delete-object-request")
.with_metrics(crate::init_metrics::OBJECT_STORAGE_DELETE_OBJECT_REQUEST)
.await
.map_err(ObjectError::from)?;
@ -662,7 +664,7 @@ impl ObjectStore {
req.header(CONTENT_LENGTH, body.len())
.body(body)
.send()
.with_metrics("pict-rs.object-storage.complete-multipart-request")
.with_metrics(crate::init_metrics::OBJECT_STORAGE_COMPLETE_MULTIPART_REQUEST)
.await
}