Add metrics to track metadata_json_download success and failure (#207)

Metadata download status metrics

---------

Co-authored-by: Kyle Espinola <[email protected]>

Api.Dockerfile

@@ -1,5 +1,5 @@
 FROM das-api/builder AS files
-FROM rust:1.75-slim-bullseye
+FROM rust:1.79-slim-bullseye
 ARG APP=/usr/src/app
 RUN apt update \
     && apt install -y curl ca-certificates tzdata \

Builder.Dockerfile

@@ -1,4 +1,4 @@
-FROM rust:1.75-bullseye AS builder
+FROM rust:1.79-bullseye AS builder
 RUN apt-get update -y && \
   apt-get install -y build-essential make git
 
@@ -21,6 +21,6 @@ WORKDIR /rust
 RUN --mount=type=cache,target=/rust/target,id=das-rust \
   cargo build --release --bins && cp `find /rust/target/release -maxdepth 1 -type f | sed 's/^\.\///' | grep -v "\." ` /rust/bins
 
-FROM rust:1.75-slim-bullseye as final
+FROM rust:1.79-slim-bullseye as final
 COPY --from=builder /rust/bins /das/
 CMD echo "Built the DAS API bins!"

Ingest.Dockerfile

@@ -1,5 +1,5 @@
 FROM das-api/builder AS files
-FROM rust:1.75-slim-bullseye
+FROM rust:1.79-slim-bullseye
 ARG APP=/usr/src/app
 RUN apt update \
     && apt install -y curl ca-certificates tzdata \

Load.Dockerfile

@@ -1,5 +1,5 @@
 FROM das-api/builder AS files
-FROM rust:1.75-slim-bullseye
+FROM rust:1.79-slim-bullseye
 ARG APP=/usr/src/app
 RUN apt update \
     && apt install -y curl ca-certificates tzdata \

Migrator.Dockerfile

@@ -1,6 +1,6 @@
 FROM das-api/builder AS files
 
-FROM rust:1.75-bullseye
+FROM rust:1.79-bullseye
 COPY init.sql /init.sql
 ENV INIT_FILE_PATH=/init.sql
 COPY --from=files /das/migration /bins/migration

Proxy.Dockerfile

@@ -1,4 +1,4 @@
-FROM rust:1.75-bullseye AS builder
+FROM rust:1.79-bullseye AS builder
 RUN cargo install wasm-pack
 
 RUN mkdir /rust

digital_asset_types/src/dapi/change_logs.rs

@@ -206,7 +206,7 @@ fn build_asset_proof(
     let mut final_node_list = vec![SimpleChangeLog::default(); req_indexes.len()];
     for node in required_nodes.iter() {
         if node.level < final_node_list.len().try_into().unwrap() {
-            final_node_list[node.level as usize] = node.to_owned();
+            node.clone_into(&mut final_node_list[node.level as usize])
         }
     }
     for (i, (n, nin)) in final_node_list

digital_asset_types/tests/common.rs

@@ -31,8 +31,10 @@ pub struct MockMetadataArgs {
     /// Since we cannot easily change Metadata, we add the new DataV2 fields here at the end.
     pub token_standard: Option<TokenStandard>,
     /// Collection
+    #[allow(dead_code)]
     pub collection: Option<Collection>,
     /// Uses
+    #[allow(dead_code)]
     pub uses: Option<Uses>,
     pub creators: Vec<Creator>,
 }

grpc-ingest/src/config.rs

@@ -111,19 +111,6 @@ impl ConfigIngestStream {
     }
 }
 
-#[derive(Debug, Clone, Deserialize, Default)]
-#[serde(default)]
-pub struct ConfigTopograph {
-    #[serde(default = "ConfigTopograph::default_num_threads")]
-    pub num_threads: usize,
-}
-
-impl ConfigTopograph {
-    pub const fn default_num_threads() -> usize {
-        5
-    }
-}
-
 #[derive(Debug, Clone, Default, Deserialize)]
 #[serde(default)]
 pub struct ConfigPrometheus {
@@ -297,12 +284,12 @@ pub struct ConfigIngesterDownloadMetadata {
         default = "ConfigIngesterDownloadMetadata::default_num_threads",
         deserialize_with = "deserialize_usize_str"
     )]
-    pub num_threads: usize,
+    pub _num_threads: usize,
     #[serde(
         default = "ConfigIngesterDownloadMetadata::default_max_attempts",
         deserialize_with = "deserialize_usize_str"
     )]
-    pub max_attempts: usize,
+    pub _max_attempts: usize,
     #[serde(
         default = "ConfigIngesterDownloadMetadata::default_request_timeout",
         deserialize_with = "deserialize_duration_str",
@@ -318,13 +305,13 @@ pub struct ConfigIngesterDownloadMetadata {
         default = "ConfigIngesterDownloadMetadata::default_stream_max_size",
         deserialize_with = "deserialize_usize_str"
     )]
-    pub pipeline_max_size: usize,
+    pub _pipeline_max_size: usize,
     #[serde(
         default = "ConfigIngesterDownloadMetadata::default_pipeline_max_idle",
         deserialize_with = "deserialize_duration_str",
         rename = "pipeline_max_idle_ms"
     )]
-    pub pipeline_max_idle: Duration,
+    pub _pipeline_max_idle: Duration,
 }
 
 impl ConfigIngesterDownloadMetadata {
@@ -366,7 +353,7 @@ pub struct ConfigBubblegumVerify {
         default = "ConfigBubblegumVerify::default_report_interval",
         deserialize_with = "deserialize_duration_str"
     )]
-    pub report_interval: Duration,
+    pub _report_interval: Duration,
     #[serde(default)]
     pub only_trees: Option<Vec<String>>,
     #[serde(

grpc-ingest/src/prom.rs

@@ -9,8 +9,7 @@ use {
     },
     program_transformers::error::ProgramTransformerError,
     prometheus::{
-        HistogramOpts, HistogramVec, IntCounter, IntCounterVec, IntGaugeVec, Opts, Registry,
-        TextEncoder,
+        HistogramOpts, HistogramVec, IntCounterVec, IntGaugeVec, Opts, Registry, TextEncoder,
     },
     std::{net::SocketAddr, sync::Once},
     tracing::{error, info},
@@ -59,8 +58,9 @@ lazy_static::lazy_static! {
         &["stream", "consumer"]
     ).unwrap();
 
-    static ref DOWNLOAD_METADATA_INSERTED_COUNT: IntCounter = IntCounter::new(
-        "download_metadata_inserted_count", "Total number of inserted tasks for download metadata"
+    static ref DOWNLOAD_METADATA_FETCHED_COUNT: IntGaugeVec = IntGaugeVec::new(
+        Opts::new("download_metadata_fetched_count", "Status of download metadata task"),
+        &["status"]
     ).unwrap();
 
     static ref INGEST_TASKS: IntGaugeVec = IntGaugeVec::new(
@@ -123,7 +123,7 @@ pub fn run_server(address: SocketAddr) -> anyhow::Result<()> {
         register!(PGPOOL_CONNECTIONS);
         register!(PROGRAM_TRANSFORMER_TASK_STATUS_COUNT);
         register!(INGEST_JOB_TIME);
-        register!(DOWNLOAD_METADATA_INSERTED_COUNT);
+        register!(DOWNLOAD_METADATA_FETCHED_COUNT);
         register!(INGEST_TASKS);
         register!(ACK_TASKS);
         register!(GRPC_TASKS);
@@ -175,7 +175,10 @@ fn metrics_handler() -> Response<Body> {
             error!("could not encode custom metrics: {}", error);
             String::new()
         });
-    Response::builder().body(Body::from(metrics)).unwrap()
+    Response::builder()
+        .header("content-type", "text/plain")
+        .body(Body::from(metrics))
+        .unwrap()
 }
 
 fn not_found_handler() -> Response<Body> {
@@ -258,6 +261,12 @@ pub fn grpc_tasks_total_dec(label: &str, stream: &str) {
     GRPC_TASKS.with_label_values(&[label, stream]).dec()
 }
 
+pub fn download_metadata_json_task_status_count_inc(status: u16) {
+    DOWNLOAD_METADATA_FETCHED_COUNT
+        .with_label_values(&[&status.to_string()])
+        .inc();
+}
+
 #[derive(Debug, Clone, Copy)]
 pub enum ProgramTransformerTaskStatusKind {
     Success,

grpc-ingest/src/redis.rs

@@ -2,12 +2,16 @@ use {
     crate::{
         config::{ConfigIngestStream, REDIS_STREAM_DATA_KEY},
         prom::{
-            ack_tasks_total_dec, ack_tasks_total_inc, ingest_job_time_set, ingest_tasks_total_dec,
-            ingest_tasks_total_inc, program_transformer_task_status_inc, redis_xack_inc,
-            redis_xlen_set, redis_xread_inc, ProgramTransformerTaskStatusKind,
+            ack_tasks_total_dec, ack_tasks_total_inc, download_metadata_json_task_status_count_inc,
+            ingest_job_time_set, ingest_tasks_total_dec, ingest_tasks_total_inc,
+            program_transformer_task_status_inc, redis_xack_inc, redis_xlen_set, redis_xread_inc,
+            ProgramTransformerTaskStatusKind,
         },
     },
-    das_core::{DownloadMetadata, DownloadMetadataInfo},
+    das_core::{
+        DownloadMetadata, DownloadMetadataInfo, FetchMetadataJsonError, MetadataJsonTaskError,
+        StatusCode,
+    },
     futures::{future::BoxFuture, stream::FuturesUnordered, StreamExt},
     program_transformers::{AccountInfo, ProgramTransformer, TransactionInfo},
     redis::{
@@ -202,10 +206,21 @@ impl MessageHandler for DownloadMetadataJsonHandle {
 
         Box::pin(async move {
             let info = DownloadMetadataInfo::try_parse_msg(input)?;
-            download_metadata
-                .handle_download(&info)
-                .await
-                .map_err(Into::into)
+            let response = download_metadata.handle_download(&info).await;
+            let status =
+                if let Err(MetadataJsonTaskError::Fetch(FetchMetadataJsonError::Response {
+                    status: StatusCode::Code(code),
+                    ..
+                })) = response
+                {
+                    code.as_u16()
+                } else {
+                    200
+                };
+
+            download_metadata_json_task_status_count_inc(status);
+
+            response.map_err(IngestMessageError::DownloadMetadataJson)
         })
     }
 }

nft_ingester/src/backfiller.rs

@@ -66,6 +66,7 @@ const BLOCK_CACHE_SIZE: usize = 300_000;
 const MAX_CACHE_COST: i64 = 32;
 const BLOCK_CACHE_DURATION: u64 = 172800;
 
+#[allow(dead_code)]
 struct SlotSeq(u64, u64);
 /// Main public entry point for backfiller task.
 pub fn setup_backfiller<T: Messenger>(

program_transformers/src/bubblegum/db.rs

@@ -504,7 +504,7 @@ where
 pub async fn upsert_asset_creators<T>(
     txn: &T,
     id: Vec<u8>,
-    creators: &Vec<Creator>,
+    creators: &[Creator],
     slot_updated: i64,
     seq: i64,
 ) -> ProgramTransformerResult<()>

rust-toolchain.toml

@@ -1,5 +1,5 @@
 [toolchain]
-channel = "1.75.0"
+channel = "1.79.0"
 components = ["clippy", "rustfmt"]
 targets = []
 profile = "minimal"