From e82a7eff2b56223efc3dc8fd5ee582b92e1731a4 Mon Sep 17 00:00:00 2001 From: Rachel Ryan Date: Fri, 10 Jul 2026 13:45:23 +0100 Subject: [PATCH] Teach Cincinnati to include product information when serving /graph-data --- .../plugin.rs | 2 + .../plugin.rs | 2 + commons/Cargo.toml | 2 +- commons/src/lib.rs | 19 ++ graph-builder/Cargo.toml | 1 - graph-builder/src/config/cli.rs | 4 + graph-builder/src/config/file.rs | 4 + graph-builder/src/config/options.rs | 40 ++++ graph-builder/src/config/settings.rs | 25 +++ graph-builder/src/graph.rs | 25 ++- graph-builder/src/lib.rs | 1 + graph-builder/src/main.rs | 27 +++ graph-builder/src/product_lifecycle.rs | 195 ++++++++++++++++++ 13 files changed, 342 insertions(+), 5 deletions(-) create mode 100644 graph-builder/src/product_lifecycle.rs diff --git a/cincinnati/src/plugins/internal/graph_builder/dkrv2_openshift_secondary_metadata_scraper/plugin.rs b/cincinnati/src/plugins/internal/graph_builder/dkrv2_openshift_secondary_metadata_scraper/plugin.rs index d5f7365b8..aa8c4b952 100644 --- a/cincinnati/src/plugins/internal/graph_builder/dkrv2_openshift_secondary_metadata_scraper/plugin.rs +++ b/cincinnati/src/plugins/internal/graph_builder/dkrv2_openshift_secondary_metadata_scraper/plugin.rs @@ -362,6 +362,8 @@ impl InternalPlugin for DkrV2OpenshiftSecondaryMetadataScraperPlugin { tokio::fs::symlink(signatures_path, signatures_symlink).await?; } + commons::copy_product_data_if_available(&io.parameters, &graph_data_path).await; + commons::create_tar( graph_data_tar_path.clone().into_boxed_path(), graph_data_path.clone().into(), diff --git a/cincinnati/src/plugins/internal/graph_builder/github_openshift_secondary_metadata_scraper/plugin.rs b/cincinnati/src/plugins/internal/graph_builder/github_openshift_secondary_metadata_scraper/plugin.rs index 87aadf08f..2cdea9f07 100644 --- a/cincinnati/src/plugins/internal/graph_builder/github_openshift_secondary_metadata_scraper/plugin.rs +++ b/cincinnati/src/plugins/internal/graph_builder/github_openshift_secondary_metadata_scraper/plugin.rs @@ -523,6 +523,8 @@ impl InternalPlugin for GithubOpenshiftSecondaryMetadataScraperPlugin { tokio::fs::symlink(signatures_path, signatures_symlink).await?; } + commons::copy_product_data_if_available(&io.parameters, graph_data_dir.as_path()).await; + commons::create_tar( graph_data_tar_path.clone().into_boxed_path(), graph_data_dir.into_boxed_path(), diff --git a/commons/Cargo.toml b/commons/Cargo.toml index 8a6c60fe6..91d8c08e9 100644 --- a/commons/Cargo.toml +++ b/commons/Cargo.toml @@ -15,7 +15,7 @@ prometheus = "0.13" serde = "^1.0.189" serde_json = "^1.0.109" serde_derive = "^1.0.123" -tokio = { version = "1.33", features = [ "rt-multi-thread" ] } +tokio = { version = "1.33", features = [ "rt-multi-thread", "fs" ] } url = "^2.5" futures = "^0.3" flate2 = "^1.0.34" diff --git a/commons/src/lib.rs b/commons/src/lib.rs index 2e336becb..df915ca3c 100644 --- a/commons/src/lib.rs +++ b/commons/src/lib.rs @@ -38,6 +38,8 @@ use url::form_urlencoded; pub static GRAPH_DATA_DIR_PARAM_KEY: &str = "io.openshift.upgrades.secondary_metadata.directory"; /// Defines the key for placing the graph_data tar path in the IO parameters pub static SECONDARY_METADATA_PARAM_KEY: &str = "io.openshift.upgrades.secondary_metadata.tar"; +/// Defines the key for placing the product lifecycle data path in the IO parameters +pub static PRODUCT_DATA_PARAM_KEY: &str = "io.openshift.upgrades.product_lifecycle.path"; /// Defines the path of default root certificate that graph_data will use pub static DEFAULT_ROOT_CERT_DIR: &str = "/etc/pki/ca-trust/extracted/"; @@ -174,6 +176,23 @@ pub fn validate_content_type( } } +/// Copy product lifecycle data into the given directory if the path is present in parameters. +pub async fn copy_product_data_if_available( + parameters: &std::collections::HashMap, + dest_dir: &Path, +) { + if let Some(product_path) = parameters.get(PRODUCT_DATA_PARAM_KEY) { + let source_path = Path::new(product_path); + if source_path.exists() { + let dest_path = dest_dir.join("products.json"); + match tokio::fs::copy(source_path, &dest_path).await { + Ok(_) => log::info!("Copied product lifecycle data to graph-data tarball"), + Err(e) => log::warn!("Failed to copy product lifecycle data: {}", e), + } + } + } +} + /// creates the tar file in the output directory from data_path pub async fn create_tar(output_path: Box, data_path: Box) -> Result<(), Error> { let tar_gz = File::create(output_path)?; diff --git a/graph-builder/Cargo.toml b/graph-builder/Cargo.toml index 174c8b216..8c1d6b544 100644 --- a/graph-builder/Cargo.toml +++ b/graph-builder/Cargo.toml @@ -8,7 +8,6 @@ build = "src/build.rs" [dependencies] actix = "0.13.2" actix-web = "^4.4.1" -chrono = "^0.4.38" actix-files = "^0.6.5" cincinnati = { path = "../cincinnati" } commons = { path = "../commons" } diff --git a/graph-builder/src/config/cli.rs b/graph-builder/src/config/cli.rs index d33a89735..32c0e7fa9 100644 --- a/graph-builder/src/config/cli.rs +++ b/graph-builder/src/config/cli.rs @@ -28,6 +28,9 @@ pub struct CliOptions { #[structopt(flatten)] pub upstream_registry: options::DockerRegistryOptions, + + #[structopt(flatten)] + pub product_lifecycle: options::ProductLifecycleOptions, } impl MergeOptions for AppSettings { @@ -41,6 +44,7 @@ impl MergeOptions for AppSettings { self.try_merge(Some(opts.service))?; self.try_merge(Some(opts.status))?; self.try_merge(Some(opts.upstream_registry))?; + self.try_merge(Some(opts.product_lifecycle))?; Ok(()) } diff --git a/graph-builder/src/config/file.rs b/graph-builder/src/config/file.rs index 275e2e904..c97c8c3a2 100644 --- a/graph-builder/src/config/file.rs +++ b/graph-builder/src/config/file.rs @@ -24,6 +24,9 @@ pub struct FileOptions { /// Status service options. pub status: Option, + /// Product lifecycle options. + pub product_lifecycle: Option, + /// Plugin settings. pub plugin_settings: Option>, } @@ -58,6 +61,7 @@ impl MergeOptions> for AppSettings { self.try_merge(file.upstream)?; self.try_merge(file.service)?; self.try_merge(file.status)?; + self.try_merge(file.product_lifecycle)?; self.try_merge(file.plugin_settings)?; } Ok(()) diff --git a/graph-builder/src/config/options.rs b/graph-builder/src/config/options.rs index 2cba63467..826a3c3f9 100644 --- a/graph-builder/src/config/options.rs +++ b/graph-builder/src/config/options.rs @@ -102,6 +102,34 @@ pub struct DockerRegistryOptions { pub fetch_concurrency: Option, } +/// Options for the product lifecycle API fetcher. +#[derive(Debug, Deserialize, Serialize, StructOpt)] +pub struct ProductLifecycleOptions { + /// Enable product lifecycle data fetching + #[structopt(long = "product.enabled")] + pub enabled: Option, + + /// URL for the Red Hat product lifecycle API + #[structopt(long = "product.api_url")] + pub api_url: Option, + + /// Polling interval in seconds for product lifecycle data + #[structopt( + long = "product.poll_interval_secs", + parse(try_from_str = duration_from_secs) + )] + #[serde(default = "Option::default", deserialize_with = "de_duration_secs")] + pub poll_interval_secs: Option, + + /// HTTP timeout in seconds for product lifecycle API requests + #[structopt( + long = "product.timeout_secs", + parse(try_from_str = duration_from_secs) + )] + #[serde(default = "Option::default", deserialize_with = "de_duration_secs")] + pub timeout_secs: Option, +} + impl MergeOptions> for AppSettings { fn try_merge(&mut self, opts: Option) -> Fallible<()> { if let Some(service) = opts { @@ -143,6 +171,18 @@ impl MergeOptions> for AppSettings { } } +impl MergeOptions> for AppSettings { + fn try_merge(&mut self, opts: Option) -> Fallible<()> { + if let Some(product) = opts { + assign_if_some!(self.product_enabled, product.enabled); + assign_if_some!(self.product_api_url, product.api_url); + assign_if_some!(self.product_poll_interval_secs, product.poll_interval_secs); + assign_if_some!(self.product_timeout_secs, product.timeout_secs); + } + Ok(()) + } +} + pub fn de_duration_secs<'de, D>(deserializer: D) -> Result, D::Error> where D: serde::Deserializer<'de>, diff --git a/graph-builder/src/config/settings.rs b/graph-builder/src/config/settings.rs index ce03607c1..c640e6d20 100644 --- a/graph-builder/src/config/settings.rs +++ b/graph-builder/src/config/settings.rs @@ -83,6 +83,22 @@ pub struct AppSettings { /// Jaeger host and port for tracing support pub tracing_endpoint: Option, + + /// Enable product lifecycle data fetching + #[default(false)] + pub product_enabled: bool, + + /// URL for the Red Hat product lifecycle API + #[default("https://access.redhat.com/product-life-cycles/api/v2/products".to_string())] + pub product_api_url: String, + + /// Polling interval (in seconds) for product lifecycle data + #[default(time::Duration::from_secs(3600))] + pub product_poll_interval_secs: time::Duration, + + /// HTTP timeout (in seconds) for product lifecycle API requests + #[default(time::Duration::from_secs(30))] + pub product_timeout_secs: time::Duration, } impl AppSettings { @@ -128,6 +144,15 @@ impl AppSettings { bail!("unexpected 0s pause"); } + if self.product_enabled { + if self.product_poll_interval_secs.as_secs() == 0 { + bail!("unexpected 0s product poll interval"); + } + if self.product_timeout_secs.as_secs() == 0 { + bail!("unexpected 0s product timeout"); + } + } + Ok(self) } diff --git a/graph-builder/src/graph.rs b/graph-builder/src/graph.rs index 8a9536e3d..74199b4d1 100644 --- a/graph-builder/src/graph.rs +++ b/graph-builder/src/graph.rs @@ -152,6 +152,8 @@ pub struct State { plugins: &'static [BoxedPlugin], registry: &'static prometheus::Registry, secondary_metadata: Arc>, + /// Path to temporary products.json file + pub product_data_path: Arc>>, } impl State { @@ -164,6 +166,7 @@ impl State { plugins: &'static [BoxedPlugin], registry: &'static prometheus::Registry, secondary_metadata: Arc>, + product_data_path: Arc>>, ) -> State { State { json, @@ -173,6 +176,7 @@ impl State { plugins, registry, secondary_metadata, + product_data_path, } } @@ -226,13 +230,23 @@ pub fn run(settings: &config::AppSettings, state: &State) -> ! { info!("graph update triggered"); let scrape_timer = UPSTREAM_SCRAPES_DURATION.start_timer(); + // Prepare plugin parameters including product data path + let mut parameters = std::collections::HashMap::new(); + if let Some(ref path) = *state.product_data_path.read() { + if let Some(path_str) = path.to_str() { + parameters.insert( + commons::PRODUCT_DATA_PARAM_KEY.to_string(), + path_str.to_string(), + ); + } + } + let scrape = cincinnati::plugins::process_blocking( state.plugins.iter(), cincinnati::plugins::PluginIO::InternalIO(cincinnati::plugins::InternalIO { // the first plugin will produce the initial graph graph: Default::default(), - // the plugins used in the graph-builder don't expect any parameters yet - parameters: Default::default(), + parameters, }), settings.scrape_timeout_secs, ); @@ -284,7 +298,12 @@ pub fn run(settings: &config::AppSettings, state: &State) -> ! { UPSTREAM_SCRAPES_DURATION.observe(scrape_value); } - GRAPH_LAST_SUCCESSFUL_REFRESH.set(chrono::Utc::now().timestamp() as i64); + GRAPH_LAST_SUCCESSFUL_REFRESH.set( + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() as i64, + ); GRAPH_FINAL_RELEASES.set(nodes_count); info!("graph update completed, {} valid releases", nodes_count); diff --git a/graph-builder/src/lib.rs b/graph-builder/src/lib.rs index b5bfac933..0cd8bab0b 100644 --- a/graph-builder/src/lib.rs +++ b/graph-builder/src/lib.rs @@ -15,6 +15,7 @@ extern crate cincinnati; pub mod config; pub mod graph; +pub mod product_lifecycle; pub mod status; #[allow(dead_code)] diff --git a/graph-builder/src/main.rs b/graph-builder/src/main.rs index 3d569ddad..d311baec0 100644 --- a/graph-builder/src/main.rs +++ b/graph-builder/src/main.rs @@ -65,6 +65,7 @@ async fn main() -> Result<(), Error> { let live = Arc::new(RwLock::new(false)); let ready = Arc::new(RwLock::new(false)); let secondary_metadata = Arc::new(RwLock::new(String::new())); + let product_data_path = Arc::new(RwLock::new(None)); graph::State::new( json_graph, settings.mandatory_client_parameters.clone(), @@ -73,9 +74,16 @@ async fn main() -> Result<(), Error> { Box::leak(Box::new(plugins)), Box::leak(Box::new(registry)), secondary_metadata, + product_data_path, ) }; + // Extract product lifecycle settings before moving settings + let product_enabled = settings.product_enabled; + let product_api_url = settings.product_api_url.clone(); + let product_poll_interval_secs = settings.product_poll_interval_secs; + let product_timeout_secs = settings.product_timeout_secs; + // Graph scraper { let graph_state = state.clone(); @@ -84,8 +92,25 @@ async fn main() -> Result<(), Error> { }); } + // Product lifecycle fetcher + if product_enabled { + let product_state = Arc::new(state.clone()); + tokio::spawn(async move { + graph_builder::product_lifecycle::run( + product_api_url, + product_poll_interval_secs, + product_timeout_secs, + product_state, + ) + .await + }); + } else { + info!("Product lifecycle fetching is disabled"); + } + // Status service. graph::register_metrics(state.registry())?; + graph_builder::product_lifecycle::register_metrics(state.registry())?; let status_state = state.clone(); let metrics_server = HttpServer::new(move || { @@ -212,6 +237,7 @@ mod tests { metrics::new_registry(Some(config::METRICS_PREFIX.to_string())).unwrap(), )); let secondary_metadata = Arc::new(RwLock::new(String::new())); + let product_data_path = Arc::new(RwLock::new(None)); State::new( json_graph, @@ -221,6 +247,7 @@ mod tests { plugins, registry, secondary_metadata, + product_data_path, ) } diff --git a/graph-builder/src/product_lifecycle.rs b/graph-builder/src/product_lifecycle.rs new file mode 100644 index 000000000..2d9334b3f --- /dev/null +++ b/graph-builder/src/product_lifecycle.rs @@ -0,0 +1,195 @@ +//! Product lifecycle API fetcher. +//! +//! This module fetches product lifecycle information from the Red Hat product API +//! and makes it available for inclusion in the graph-data tarball. + +use crate::graph::State; +use commons::prelude_errors::*; +use log::{error, info}; +use prometheus::{Counter, Gauge, Opts}; +use std::io::Write; +use std::path::PathBuf; +use std::sync::Arc; +use std::time::Duration; +use tokio::time::sleep; + +lazy_static::lazy_static! { + static ref PRODUCT_LIFECYCLE_FETCHES: Counter = Counter::with_opts( + Opts::new( + "product_lifecycle_fetches_total", + "Total number of product lifecycle API fetch attempts" + ) + ) + .unwrap(); + + static ref PRODUCT_LIFECYCLE_ERRORS: Counter = Counter::with_opts( + Opts::new( + "product_lifecycle_errors_total", + "Total number of product lifecycle API fetch errors" + ) + ) + .unwrap(); + + static ref PRODUCT_LIFECYCLE_NOT_MODIFIED: Counter = Counter::with_opts( + Opts::new( + "product_lifecycle_not_modified_total", + "Total number of 304 Not Modified responses from product lifecycle API" + ) + ) + .unwrap(); + + static ref PRODUCT_LIFECYCLE_LAST_FETCH: Gauge = Gauge::with_opts( + Opts::new( + "product_lifecycle_last_successful_fetch_timestamp", + "UTC timestamp of last successful product lifecycle fetch" + ) + ) + .unwrap(); +} + +/// Register product lifecycle metrics +pub fn register_metrics(registry: &prometheus::Registry) -> Fallible<()> { + registry.register(Box::new(PRODUCT_LIFECYCLE_FETCHES.clone()))?; + registry.register(Box::new(PRODUCT_LIFECYCLE_ERRORS.clone()))?; + registry.register(Box::new(PRODUCT_LIFECYCLE_NOT_MODIFIED.clone()))?; + registry.register(Box::new(PRODUCT_LIFECYCLE_LAST_FETCH.clone()))?; + Ok(()) +} + +/// Fetch product data from the API +async fn fetch_products( + client: &reqwest::Client, + api_url: &str, + last_etag: Option<&str>, +) -> Fallible<(Option, Option)> { + let mut request = client.get(api_url).header("Accept", "application/json"); + + if let Some(etag) = last_etag { + request = request.header("If-None-Match", etag); + } + + let response = request.send().await?; + let status = response.status(); + let new_etag = response + .headers() + .get("etag") + .and_then(|v| v.to_str().ok()) + .map(String::from); + + if status == reqwest::StatusCode::NOT_MODIFIED { + // 304 Not Modified - data hasn't changed + PRODUCT_LIFECYCLE_NOT_MODIFIED.inc(); + info!("Product lifecycle data not modified (304)"); + return Ok((None, new_etag)); + } + + if !status.is_success() { + bail!("Product lifecycle API returned status {}", status); + } + + let body = response.text().await?; + Ok((Some(body), new_etag)) +} + +/// Save product JSON to a temporary file +fn save_to_file(json: &str) -> Fallible { + let temp_dir = std::env::temp_dir(); + let final_path = temp_dir.join("products.json"); + + let mut temp_file = + tempfile::NamedTempFile::new_in(&temp_dir).context("Failed to create temporary file")?; + + temp_file + .write_all(json.as_bytes()) + .context("Failed to write to temporary file")?; + temp_file + .as_file_mut() + .sync_all() + .context("Failed to sync temporary file")?; + + temp_file + .persist(&final_path) + .context("Failed to persist temporary file")?; + + Ok(final_path) +} + +/// Main polling loop for product lifecycle data +pub async fn run(api_url: String, poll_interval: Duration, timeout: Duration, state: Arc) { + info!( + "Starting product lifecycle fetcher: api_url={}, poll_interval={}s, timeout={}s", + api_url, + poll_interval.as_secs(), + timeout.as_secs() + ); + + // Build HTTP client + let client = reqwest::ClientBuilder::new() + .gzip(true) + .timeout(timeout) + .build() + .expect("Failed to build HTTP client for product lifecycle"); + + let mut first_iteration = true; + let mut last_etag: Option = None; + + loop { + if first_iteration { + first_iteration = false; + } else { + sleep(poll_interval).await; + } + + PRODUCT_LIFECYCLE_FETCHES.inc(); + + // Fetch from API + match fetch_products(&client, &api_url, last_etag.as_deref()).await { + Ok((Some(json), new_etag)) => { + // New data received + info!( + "Fetched product lifecycle data ({} bytes), etag={:?}", + json.len(), + new_etag + ); + + // Validate it's valid JSON + if let Err(e) = serde_json::from_str::(&json) { + error!("Product lifecycle API returned invalid JSON: {}", e); + PRODUCT_LIFECYCLE_ERRORS.inc(); + continue; + } + + // Save to file + match save_to_file(&json) { + Ok(path) => { + last_etag = new_etag; + *state.product_data_path.write() = Some(path.clone()); + + PRODUCT_LIFECYCLE_LAST_FETCH.set( + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() as f64, + ); + info!("Saved product lifecycle data to {:?}", path); + } + Err(e) => { + error!("Failed to save product lifecycle data to file: {}", e); + PRODUCT_LIFECYCLE_ERRORS.inc(); + } + } + } + Ok((None, new_etag)) => { + // 304 Not Modified - update ETag but keep existing data + if let Some(etag) = new_etag { + last_etag = Some(etag); + } + } + Err(e) => { + error!("Failed to fetch product lifecycle data: {}", e); + PRODUCT_LIFECYCLE_ERRORS.inc(); + // Continue loop - will retry on next interval + } + } + } +}