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

Start describing metrics

Browse source
This commit is contained in:
asonix 2024-02-04 14:26:18 -06:00
parent a6c4acd6f3
commit d8c4a22ba9
6 changed files with 505 additions and 82 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
src/generate.rs
View file

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

419
src/init_metrics.rs Normal file
View file

@ -0,0 +1,419 @@
pub(super) fn init_metrics() {
describe_postgres();
describe_middleware();
describe_generate();
describe_object_storage();
}
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_VARIANST_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_historgram!(
OBJECT_STORAGE_HEAD_BUCKET_REQUEST,
"Timings for HEAD requests for the pict-rs Bucket in object storage"
);
metrics::describe_historgram!(
OBJECT_STORAGE_PUT_OBJECT_REQUEST,
"Timings for PUT requests for uploading media to object storage"
);
metrics::describe_historgram!(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_historgram!(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_historgram!(
OBJECT_STORAGE_ABORT_MULTIPART_REQUEST,
"Timings for aborting a multipart upload to object storage"
);
metrics::describe_historgram!(
OBJECT_STORAGE_GET_OBJECT_REQUEST,
"Timings for requesting media from object storage"
);
metrics::describe_historgram!(
OBJECT_STORAGE_GET_OBJECT_REQUEST_STREAM,
"Timings for streaming an object from object storage"
);
metrics::describe_historgram!(
OBJECT_STORAGE_HEAD_OBJECT_REQUEST,
"Timings for requesting metadata for media from object storage"
);
metrics::describe_historgram!(
OBJECT_STORAGE_DELETE_OBJECT_REQUEST,
"Timings for requesting media in object storage be deleted"
);
metrics::describe_historgram!(
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";

1
src/lib.rs
View file

@ -14,6 +14,7 @@ mod formats;
mod future; mod future;
mod generate; mod generate;
mod ingest; mod ingest;
mod init_metrics;
mod init_tracing; mod init_tracing;
mod magick; mod magick;
mod middleware; mod middleware;

11
src/middleware/metrics.rs
View file

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

120
src/repo/postgres.rs
View file

@ -253,11 +253,11 @@ fn build_pool(
.create_timeout(Some(Duration::from_secs(2))) .create_timeout(Some(Duration::from_secs(2)))
.recycle_timeout(Some(Duration::from_secs(2))) .recycle_timeout(Some(Duration::from_secs(2)))
.post_create(Hook::sync_fn(|_, _| { .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(()) Ok(())
})) }))
.post_recycle(Hook::sync_fn(|_, _| { .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(()) Ok(())
})) }))
.max_size(max_size) .max_size(max_size)
@ -355,9 +355,9 @@ impl GetConnectionMetricsGuard {
impl Drop for GetConnectionMetricsGuard { impl Drop for GetConnectionMetricsGuard {
fn drop(&mut self) { 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); .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, => (!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) .or_insert_with(crate::sync::notify)
.notify_one(); .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()) .and_then(|weak| weak.upgrade())
{ {
notifier.notify_waiters(); 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 { while let Ok(notification) = receiver.recv_async().await {
tracing::trace!("delegate_notifications: looping"); 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() { match notification.channel() {
"queue_status_channel" => { "queue_status_channel" => {
@ -611,7 +611,7 @@ impl HashRepo for PostgresRepo {
let count = hashes let count = hashes
.count() .count()
.get_result::<i64>(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -630,7 +630,7 @@ impl HashRepo for PostgresRepo {
.select(created_at) .select(created_at)
.filter(hash.eq(&input_hash)) .filter(hash.eq(&input_hash))
.get_result(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -661,7 +661,7 @@ impl HashRepo for PostgresRepo {
.filter(created_at.lt(timestamp)) .filter(created_at.lt(timestamp))
.order(created_at.desc()) .order(created_at.desc())
.get_result::<(time::PrimitiveDateTime, Hash)>(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -700,7 +700,7 @@ impl HashRepo for PostgresRepo {
.then_order_by(hash.desc()) .then_order_by(hash.desc())
.limit(limit as i64 + 1) .limit(limit as i64 + 1)
.get_results::<Hash>(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -714,7 +714,7 @@ impl HashRepo for PostgresRepo {
.then_order_by(hash) .then_order_by(hash)
.limit(limit as i64) .limit(limit as i64)
.get_results::<Hash>(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -729,7 +729,7 @@ impl HashRepo for PostgresRepo {
.then_order_by(hash.desc()) .then_order_by(hash.desc())
.limit(limit as i64 + 1) .limit(limit as i64 + 1)
.get_results::<Hash>(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -768,7 +768,7 @@ impl HashRepo for PostgresRepo {
created_at.eq(&timestamp), created_at.eq(&timestamp),
)) ))
.execute(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)?; .map_err(|_| PostgresError::DbTimeout)?;
@ -797,7 +797,7 @@ impl HashRepo for PostgresRepo {
.filter(hash.eq(&input_hash)) .filter(hash.eq(&input_hash))
.set(identifier.eq(input_identifier.as_ref())) .set(identifier.eq(input_identifier.as_ref()))
.execute(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -816,7 +816,7 @@ impl HashRepo for PostgresRepo {
.select(identifier) .select(identifier)
.filter(hash.eq(&input_hash)) .filter(hash.eq(&input_hash))
.get_result::<String>(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -844,7 +844,7 @@ impl HashRepo for PostgresRepo {
identifier.eq(input_identifier.as_ref()), identifier.eq(input_identifier.as_ref()),
)) ))
.execute(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)?; .map_err(|_| PostgresError::DbTimeout)?;
@ -874,7 +874,7 @@ impl HashRepo for PostgresRepo {
.filter(hash.eq(&input_hash)) .filter(hash.eq(&input_hash))
.filter(variant.eq(&input_variant)) .filter(variant.eq(&input_variant))
.get_result::<String>(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -895,7 +895,7 @@ impl HashRepo for PostgresRepo {
.select((variant, identifier)) .select((variant, identifier))
.filter(hash.eq(&input_hash)) .filter(hash.eq(&input_hash))
.get_results::<(String, String)>(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -921,7 +921,7 @@ impl HashRepo for PostgresRepo {
.filter(hash.eq(&input_hash)) .filter(hash.eq(&input_hash))
.filter(variant.eq(&input_variant)) .filter(variant.eq(&input_variant))
.execute(&mut conn) .execute(&mut conn)
.with_metrics("pict-rs.postgres.variants.remove") .with_metrics(crate::init_metrics::POSTGRES_VARIANTS_REMOVE)
.with_timeout(Duration::from_secs(5)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -944,7 +944,7 @@ impl HashRepo for PostgresRepo {
.filter(hash.eq(&input_hash)) .filter(hash.eq(&input_hash))
.set(motion_identifier.eq(input_identifier.as_ref())) .set(motion_identifier.eq(input_identifier.as_ref()))
.execute(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -963,7 +963,7 @@ impl HashRepo for PostgresRepo {
.select(motion_identifier) .select(motion_identifier)
.filter(hash.eq(&input_hash)) .filter(hash.eq(&input_hash))
.get_result::<Option<String>>(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -984,13 +984,13 @@ impl HashRepo for PostgresRepo {
diesel::delete(schema::variants::dsl::variants) diesel::delete(schema::variants::dsl::variants)
.filter(schema::variants::dsl::hash.eq(&input_hash)) .filter(schema::variants::dsl::hash.eq(&input_hash))
.execute(conn) .execute(conn)
.with_metrics("pict-rs.postgres.variants.cleanup") .with_metrics(crate::init_metrics::POSTGRES_VARIANTS_CLEANUP)
.await?; .await?;
diesel::delete(schema::hashes::dsl::hashes) diesel::delete(schema::hashes::dsl::hashes)
.filter(schema::hashes::dsl::hash.eq(&input_hash)) .filter(schema::hashes::dsl::hash.eq(&input_hash))
.execute(conn) .execute(conn)
.with_metrics("pict-rs.postgres.hashes.cleanup") .with_metrics(crate::init_metrics::POSTGRES_HASHES_CLEANUP)
.await .await
}) })
}) })
@ -1021,7 +1021,7 @@ impl AliasRepo for PostgresRepo {
token.eq(delete_token), token.eq(delete_token),
)) ))
.execute(&mut conn) .execute(&mut conn)
.with_metrics("pict-rs.postgres.aliases.create") .with_metrics(crate::init_metrics::POSTGRES_ALIASES_CREATE)
.with_timeout(Duration::from_secs(5)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)?; .map_err(|_| PostgresError::DbTimeout)?;
@ -1046,7 +1046,7 @@ impl AliasRepo for PostgresRepo {
.select(token) .select(token)
.filter(alias.eq(input_alias)) .filter(alias.eq(input_alias))
.get_result(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -1066,7 +1066,7 @@ impl AliasRepo for PostgresRepo {
.select(hash) .select(hash)
.filter(alias.eq(input_alias)) .filter(alias.eq(input_alias))
.get_result(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -1086,7 +1086,7 @@ impl AliasRepo for PostgresRepo {
.select(alias) .select(alias)
.filter(hash.eq(&input_hash)) .filter(hash.eq(&input_hash))
.get_results(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -1104,7 +1104,7 @@ impl AliasRepo for PostgresRepo {
diesel::delete(aliases) diesel::delete(aliases)
.filter(alias.eq(input_alias)) .filter(alias.eq(input_alias))
.execute(&mut conn) .execute(&mut conn)
.with_metrics("pict-rs.postgres.aliases.cleanup") .with_metrics(crate::init_metrics::POSTGRES_ALIASES_CLEANUP)
.with_timeout(Duration::from_secs(5)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -1130,7 +1130,7 @@ impl SettingsRepo for PostgresRepo {
.do_update() .do_update()
.set(value.eq(&input_value)) .set(value.eq(&input_value))
.execute(&mut conn) .execute(&mut conn)
.with_metrics("pict-rs.postgres.settings.set") .with_metrics(crate::init_metrics::POSTGRES_SETTINGS_SET)
.with_timeout(Duration::from_secs(5)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -1149,7 +1149,7 @@ impl SettingsRepo for PostgresRepo {
.select(value) .select(value)
.filter(key.eq(input_key)) .filter(key.eq(input_key))
.get_result::<String>(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -1172,7 +1172,7 @@ impl SettingsRepo for PostgresRepo {
diesel::delete(settings) diesel::delete(settings)
.filter(key.eq(input_key)) .filter(key.eq(input_key))
.execute(&mut conn) .execute(&mut conn)
.with_metrics("pict-rs.postgres.settings.remove") .with_metrics(crate::init_metrics::POSTGRES_SETTINGS_REMOVE)
.with_timeout(Duration::from_secs(5)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -1200,7 +1200,7 @@ impl DetailsRepo for PostgresRepo {
diesel::insert_into(details) diesel::insert_into(details)
.values((identifier.eq(input_identifier.as_ref()), json.eq(&value))) .values((identifier.eq(input_identifier.as_ref()), json.eq(&value)))
.execute(&mut conn) .execute(&mut conn)
.with_metrics("pict-rs.postgres.details.relate") .with_metrics(crate::init_metrics::POSTGRES_DETAILS_RELATE)
.with_timeout(Duration::from_secs(5)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -1219,7 +1219,7 @@ impl DetailsRepo for PostgresRepo {
.select(json) .select(json)
.filter(identifier.eq(input_identifier.as_ref())) .filter(identifier.eq(input_identifier.as_ref()))
.get_result::<serde_json::Value>(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -1242,7 +1242,7 @@ impl DetailsRepo for PostgresRepo {
diesel::delete(details) diesel::delete(details)
.filter(identifier.eq(input_identifier.as_ref())) .filter(identifier.eq(input_identifier.as_ref()))
.execute(&mut conn) .execute(&mut conn)
.with_metrics("pict-rs.postgres.details.cleanup") .with_metrics(crate::init_metrics::POSTGRES_DETAILS_CLEANUP)
.with_timeout(Duration::from_secs(5)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -1262,7 +1262,7 @@ impl QueueRepo for PostgresRepo {
let count = job_queue let count = job_queue
.count() .count()
.get_result::<i64>(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -1292,7 +1292,7 @@ impl QueueRepo for PostgresRepo {
)) ))
.returning(id) .returning(id)
.get_result::<Uuid>(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -1337,7 +1337,7 @@ impl QueueRepo for PostgresRepo {
diesel::sql_query("LISTEN queue_status_channel;") diesel::sql_query("LISTEN queue_status_channel;")
.execute(&mut notifier_conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -1357,7 +1357,7 @@ impl QueueRepo for PostgresRepo {
worker.eq(Option::<Uuid>::None), worker.eq(Option::<Uuid>::None),
)) ))
.execute(&mut conn) .execute(&mut conn)
.with_metrics("pict-rs.postgres.queue.requeue") .with_metrics(crate::init_metrics::POSTGRES_QUEUE_REQUEUE)
.with_timeout(Duration::from_secs(5)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -1392,7 +1392,7 @@ impl QueueRepo for PostgresRepo {
)) ))
.returning((id, job)) .returning((id, job))
.get_result(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -1439,7 +1439,7 @@ impl QueueRepo for PostgresRepo {
) )
.set(heartbeat.eq(timestamp)) .set(heartbeat.eq(timestamp))
.execute(&mut conn) .execute(&mut conn)
.with_metrics("pict-rs.postgres.queue.heartbeat") .with_metrics(crate::init_metrics::POSGRES_QUEUE_HEARTBEAT)
.with_timeout(Duration::from_secs(5)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -1466,7 +1466,7 @@ impl QueueRepo for PostgresRepo {
.and(worker.eq(worker_id)), .and(worker.eq(worker_id)),
) )
.execute(&mut conn) .execute(&mut conn)
.with_metrics("pict-rs.postgres.queue.complete") .with_metrics(crate::init_metrics::POSTGRES_QUEUE_COMPLETE)
.with_timeout(Duration::from_secs(5)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -1487,7 +1487,7 @@ impl StoreMigrationRepo for PostgresRepo {
let count = store_migrations let count = store_migrations
.count() .count()
.get_result::<i64>(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -1514,7 +1514,7 @@ impl StoreMigrationRepo for PostgresRepo {
.on_conflict(old_identifier) .on_conflict(old_identifier)
.do_nothing() .do_nothing()
.execute(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -1533,7 +1533,7 @@ impl StoreMigrationRepo for PostgresRepo {
store_migrations.filter(old_identifier.eq(input_old_identifier.as_ref())), store_migrations.filter(old_identifier.eq(input_old_identifier.as_ref())),
)) ))
.get_result(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -1550,7 +1550,7 @@ impl StoreMigrationRepo for PostgresRepo {
diesel::delete(store_migrations) diesel::delete(store_migrations)
.execute(&mut conn) .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)) .with_timeout(Duration::from_secs(20))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -1571,7 +1571,7 @@ impl ProxyRepo for PostgresRepo {
diesel::insert_into(proxies) diesel::insert_into(proxies)
.values((url.eq(input_url.as_str()), alias.eq(&input_alias))) .values((url.eq(input_url.as_str()), alias.eq(&input_alias)))
.execute(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -1590,7 +1590,7 @@ impl ProxyRepo for PostgresRepo {
.select(alias) .select(alias)
.filter(url.eq(input_url.as_str())) .filter(url.eq(input_url.as_str()))
.get_result(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -1609,7 +1609,7 @@ impl ProxyRepo for PostgresRepo {
diesel::delete(proxies) diesel::delete(proxies)
.filter(alias.eq(&input_alias)) .filter(alias.eq(&input_alias))
.execute(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -1637,7 +1637,7 @@ impl AliasAccessRepo for PostgresRepo {
.filter(alias.eq(&input_alias)) .filter(alias.eq(&input_alias))
.set(accessed.eq(timestamp)) .set(accessed.eq(timestamp))
.execute(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -1659,7 +1659,7 @@ impl AliasAccessRepo for PostgresRepo {
.select(accessed) .select(accessed)
.filter(alias.eq(&input_alias)) .filter(alias.eq(&input_alias))
.get_result::<time::PrimitiveDateTime>(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -1689,7 +1689,7 @@ impl AliasAccessRepo for PostgresRepo {
.order(accessed.desc()) .order(accessed.desc())
.limit(100) .limit(100)
.get_results(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -1725,7 +1725,7 @@ impl VariantAccessRepo for PostgresRepo {
.filter(hash.eq(&input_hash).and(variant.eq(&input_variant))) .filter(hash.eq(&input_hash).and(variant.eq(&input_variant)))
.set(accessed.eq(timestamp)) .set(accessed.eq(timestamp))
.execute(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -1748,7 +1748,7 @@ impl VariantAccessRepo for PostgresRepo {
.select(accessed) .select(accessed)
.filter(hash.eq(&input_hash).and(variant.eq(&input_variant))) .filter(hash.eq(&input_hash).and(variant.eq(&input_variant)))
.get_result(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -1778,7 +1778,7 @@ impl VariantAccessRepo for PostgresRepo {
.order(accessed.desc()) .order(accessed.desc())
.limit(100) .limit(100)
.get_results(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -1843,7 +1843,7 @@ impl UploadRepo for PostgresRepo {
.default_values() .default_values()
.returning(id) .returning(id)
.get_result(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -1868,7 +1868,7 @@ impl UploadRepo for PostgresRepo {
diesel::sql_query("LISTEN upload_completion_channel;") diesel::sql_query("LISTEN upload_completion_channel;")
.execute(&mut notifier_conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -1882,7 +1882,7 @@ impl UploadRepo for PostgresRepo {
.select(result) .select(result)
.filter(id.eq(upload_id.id)) .filter(id.eq(upload_id.id))
.get_result(&mut conn) .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)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -1924,7 +1924,7 @@ impl UploadRepo for PostgresRepo {
diesel::delete(uploads) diesel::delete(uploads)
.filter(id.eq(upload_id.id)) .filter(id.eq(upload_id.id))
.execute(&mut conn) .execute(&mut conn)
.with_metrics("pict-rs.postgres.uploads.claim") .with_metrics(crate::init_metrics::POSTGRES_UPLOADS_CLAIM)
.with_timeout(Duration::from_secs(5)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?
@ -1951,7 +1951,7 @@ impl UploadRepo for PostgresRepo {
.filter(id.eq(upload_id.id)) .filter(id.eq(upload_id.id))
.set(result.eq(upload_result)) .set(result.eq(upload_result))
.execute(&mut conn) .execute(&mut conn)
.with_metrics("pict-rs.postgres.uploads.complete") .with_metrics(crate::init_metrics::POSTGRES_UPLOADS_COMPLETE)
.with_timeout(Duration::from_secs(5)) .with_timeout(Duration::from_secs(5))
.await .await
.map_err(|_| PostgresError::DbTimeout)? .map_err(|_| PostgresError::DbTimeout)?

28
src/store/object_store.rs
View file

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