OTA-1974: Teach Cincinnati to include product information when serving /graph-data

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(),

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(),

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"

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<String, String>,
+    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<Path>, data_path: Box<Path>) -> Result<(), Error> {
     let tar_gz = File::create(output_path)?;

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" }

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<CliOptions> for AppSettings {
@@ -41,6 +44,7 @@ impl MergeOptions<CliOptions> 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(())
     }

graph-builder/src/config/file.rs

@@ -24,6 +24,9 @@ pub struct FileOptions {
     /// Status service options.
     pub status: Option<options::StatusOptions>,
 
+    /// Product lifecycle options.
+    pub product_lifecycle: Option<options::ProductLifecycleOptions>,
+
     /// Plugin settings.
     pub plugin_settings: Option<Vec<toml::Value>>,
 }
@@ -58,6 +61,7 @@ impl MergeOptions<Option<FileOptions>> 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(())

graph-builder/src/config/options.rs

@@ -102,6 +102,34 @@ pub struct DockerRegistryOptions {
     pub fetch_concurrency: Option<usize>,
 }
 
+/// 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<bool>,
+
+    /// URL for the Red Hat product lifecycle API
+    #[structopt(long = "product.api_url")]
+    pub api_url: Option<String>,
+
+    /// 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<Duration>,
+
+    /// 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<Duration>,
+}
+
 impl MergeOptions<Option<ServiceOptions>> for AppSettings {
     fn try_merge(&mut self, opts: Option<ServiceOptions>) -> Fallible<()> {
         if let Some(service) = opts {
@@ -143,6 +171,18 @@ impl MergeOptions<Option<DockerRegistryOptions>> for AppSettings {
     }
 }
 
+impl MergeOptions<Option<ProductLifecycleOptions>> for AppSettings {
+    fn try_merge(&mut self, opts: Option<ProductLifecycleOptions>) -> 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<Option<std::time::Duration>, D::Error>
 where
     D: serde::Deserializer<'de>,

graph-builder/src/config/settings.rs

@@ -83,6 +83,22 @@ pub struct AppSettings {
 
     /// Jaeger host and port for tracing support
     pub tracing_endpoint: Option<String>,
+
+    /// 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)
     }
 

graph-builder/src/graph.rs

@@ -152,6 +152,8 @@ pub struct State {
     plugins: &'static [BoxedPlugin],
     registry: &'static prometheus::Registry,
     secondary_metadata: Arc<RwLock<String>>,
+    /// Path to temporary products.json file
+    pub product_data_path: Arc<RwLock<Option<std::path::PathBuf>>>,
 }
 
 impl State {
@@ -164,6 +166,7 @@ impl State {
         plugins: &'static [BoxedPlugin],
         registry: &'static prometheus::Registry,
         secondary_metadata: Arc<RwLock<String>>,
+        product_data_path: Arc<RwLock<Option<std::path::PathBuf>>>,
     ) -> 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);

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)]

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,
         )
     }