*: split dumnati to commons, graph-builder, and policy-engine

Cargo.toml

@@ -1,7 +1,8 @@
 [workspace]
 
 members = [
+	"commons",
     "dumnati",
-#    "graph-builder",
-#    "policy-engine",
+   "fcos-graph-builder",
+   "fcos-policy-engine",
 ]

commons/.gitignore

@@ -0,0 +1,2 @@
+/target
+**/*.rs.bk

commons/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "commons"
+version = "0.1.0"
+authors = ["Allen Bai <abai@redhat.com>"]
+edition = "2018"
+publish = false
+
+[dependencies]
+actix-web = "^2.0.0"
+chrono = "^0.4.7"
+failure = "^0.1.1"
+maplit = "^1.0"
+prometheus = "^0.9"
+serde = "^1.0.70"
+serde_derive = "^1.0.70"

commons/src/graph.rs

@@ -0,0 +1,203 @@
+use crate::metadata;
+use failure::Fallible;
+use serde_derive::{Deserialize, Serialize};
+use std::collections::HashMap;
+
+/// Single release entry in the Cincinnati update-graph.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct CincinnatiPayload {
+    pub version: String,
+    pub metadata: HashMap<String, String>,
+    pub payload: String,
+}
+
+/// Cincinnati update-graph, a DAG with releases (nodes) and update paths (edges).
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct Graph {
+    pub nodes: Vec<CincinnatiPayload>,
+    pub edges: Vec<(u64, u64)>,
+}
+
+impl Default for Graph {
+    fn default() -> Self {
+        Self {
+            nodes: vec![],
+            edges: vec![],
+        }
+    }
+}
+
+impl Graph {
+    /// Assemble a graph from release-index and updates metadata.
+    pub fn from_metadata(
+        releases: Vec<metadata::Release>,
+        updates: metadata::UpdatesJSON,
+    ) -> Fallible<Self> {
+        let nodes: Vec<CincinnatiPayload> = releases
+            .into_iter()
+            .enumerate()
+            .map(|(age_index, entry)| {
+                let mut current = CincinnatiPayload {
+                    version: entry.version,
+                    payload: "".to_string(),
+                    metadata: maplit::hashmap! {
+                        metadata::AGE_INDEX.to_string() => age_index.to_string(),
+                    },
+                };
+                for commit in entry.commits {
+                    if commit.architecture.is_empty() || commit.checksum.is_empty() {
+                        continue;
+                    }
+                    let key = format!("{}.{}", metadata::ARCH_PREFIX, commit.architecture);
+                    let value = commit.checksum;
+                    current.metadata.insert(key, value);
+                }
+
+                // Augment with dead-ends metadata.
+                Self::inject_deadend_reason(&updates, &mut current);
+
+                // Augment with barriers metadata.
+                Self::inject_barrier_reason(&updates, &mut current);
+
+                // Augment with rollouts metadata.
+                Self::inject_throttling_params(&updates, &mut current);
+
+                current
+            })
+            .collect();
+
+        // Compute the update graph.
+        let edges = Self::compute_edges(&nodes)?;
+
+        let graph = Graph { nodes, edges };
+        Ok(graph)
+    }
+
+    /// Compute edges based on graph metadata.
+    fn compute_edges(nodes: &[CincinnatiPayload]) -> Fallible<Vec<(u64, u64)>> {
+        use std::collections::BTreeSet;
+        use std::ops::Bound;
+
+        // Collect all rollouts and barriers.
+        let mut rollouts = BTreeSet::<u64>::new();
+        let mut barriers = BTreeSet::<u64>::new();
+        for (index, release) in nodes.iter().enumerate() {
+            if release.metadata.contains_key(metadata::ROLLOUT) {
+                rollouts.insert(index as u64);
+            }
+            if release.metadata.contains_key(metadata::BARRIER) {
+                barriers.insert(index as u64);
+            }
+        }
+
+        // Add edges targeting rollouts, back till the previous barrier.
+        let mut edges = vec![];
+        for (index, _release) in nodes.iter().enumerate().rev() {
+            let age = index as u64;
+            if !rollouts.contains(&age) {
+                continue;
+            }
+
+            let previous_barrier = barriers
+                .range((Bound::Unbounded, Bound::Excluded(age)))
+                .next_back()
+                .cloned()
+                .unwrap_or(0);
+
+            for i in previous_barrier..age {
+                edges.push((i, age))
+            }
+        }
+
+        // Add edges targeting barriers, back till the previous barrier.
+        let mut start = 0;
+        for target in barriers {
+            if rollouts.contains(&target) {
+                // This is an in-progress barrier. Rollout logic already took care
+                // of it, nothing to do here.
+            } else {
+                for i in start..target {
+                    edges.push((i, target))
+                }
+            }
+            start = target;
+        }
+
+        Ok(edges)
+    }
+
+    fn inject_barrier_reason(updates: &metadata::UpdatesJSON, release: &mut CincinnatiPayload) {
+        for entry in &updates.releases {
+            if entry.version != release.version {
+                continue;
+            }
+
+            if let Some(barrier) = &entry.metadata.barrier {
+                let reason = if barrier.reason.is_empty() {
+                    "generic"
+                } else {
+                    &barrier.reason
+                };
+
+                release
+                    .metadata
+                    .insert(metadata::BARRIER.to_string(), true.to_string());
+                release
+                    .metadata
+                    .insert(metadata::BARRIER_REASON.to_string(), reason.to_string());
+            }
+        }
+    }
+
+    fn inject_deadend_reason(updates: &metadata::UpdatesJSON, release: &mut CincinnatiPayload) {
+        for entry in &updates.releases {
+            if entry.version != release.version {
+                continue;
+            }
+
+            if let Some(deadend) = &entry.metadata.deadend {
+                let reason = if deadend.reason.is_empty() {
+                    "generic"
+                } else {
+                    &deadend.reason
+                };
+
+                release
+                    .metadata
+                    .insert(metadata::DEADEND.to_string(), true.to_string());
+                release
+                    .metadata
+                    .insert(metadata::DEADEND_REASON.to_string(), reason.to_string());
+            }
+        }
+    }
+
+    fn inject_throttling_params(updates: &metadata::UpdatesJSON, release: &mut CincinnatiPayload) {
+        for entry in &updates.releases {
+            if entry.version != release.version {
+                continue;
+            }
+
+            if let Some(rollout) = &entry.metadata.rollout {
+                release
+                    .metadata
+                    .insert(metadata::ROLLOUT.to_string(), true.to_string());
+                if let Some(val) = rollout.start_epoch {
+                    release
+                        .metadata
+                        .insert(metadata::START_EPOCH.to_string(), val.to_string());
+                }
+                if let Some(val) = rollout.start_percentage {
+                    release
+                        .metadata
+                        .insert(metadata::START_VALUE.to_string(), val.to_string());
+                }
+                if let Some(minutes) = &rollout.duration_minutes {
+                    release
+                        .metadata
+                        .insert(metadata::DURATION.to_string(), minutes.to_string());
+                }
+            }
+        }
+    }
+}

commons/src/lib.rs

@@ -0,0 +1,4 @@
+pub mod graph;
+pub mod metadata;
+pub mod metrics;
+pub mod policy;

commons/src/metadata.rs

@@ -0,0 +1,80 @@
+//! Fedora CoreOS metadata.
+
+use serde_derive::Deserialize;
+
+/// Templated URL for release index.
+pub static RELEASES_JSON: &str =
+    "https://builds.coreos.fedoraproject.org/prod/streams/${stream}/releases.json";
+
+/// Templated URL for stream metadata.
+pub static STREAM_JSON: &str = "https://builds.coreos.fedoraproject.org/updates/${stream}.json";
+
+pub static SCHEME: &str = "org.fedoraproject.coreos.scheme";
+
+pub static AGE_INDEX: &str = "org.fedoraproject.coreos.releases.age_index";
+pub static ARCH_PREFIX: &str = "org.fedoraproject.coreos.releases.arch";
+
+pub static BARRIER: &str = "org.fedoraproject.coreos.updates.barrier";
+pub static BARRIER_REASON: &str = "org.fedoraproject.coreos.updates.barrier_reason";
+pub static DEADEND: &str = "org.fedoraproject.coreos.updates.deadend";
+pub static DEADEND_REASON: &str = "org.fedoraproject.coreos.updates.deadend_reason";
+pub static ROLLOUT: &str = "org.fedoraproject.coreos.updates.rollout";
+pub static DURATION: &str = "org.fedoraproject.coreos.updates.duration_minutes";
+pub static START_EPOCH: &str = "org.fedoraproject.coreos.updates.start_epoch";
+pub static START_VALUE: &str = "org.fedoraproject.coreos.updates.start_value";
+
+/// Fedora CoreOS release index.
+#[derive(Debug, Deserialize)]
+pub struct ReleasesJSON {
+    pub releases: Vec<Release>,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct Release {
+    pub commits: Vec<ReleaseCommit>,
+    pub version: String,
+    pub metadata: String,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct ReleaseCommit {
+    pub architecture: String,
+    pub checksum: String,
+}
+
+/// Fedora CoreOS updates metadata
+#[derive(Debug, Deserialize)]
+pub struct UpdatesJSON {
+    pub stream: String,
+    pub releases: Vec<ReleaseUpdate>,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct ReleaseUpdate {
+    pub version: String,
+    pub metadata: UpdateMetadata,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct UpdateMetadata {
+    pub barrier: Option<UpdateBarrier>,
+    pub deadend: Option<UpdateDeadend>,
+    pub rollout: Option<UpdateRollout>,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct UpdateBarrier {
+    pub reason: String,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct UpdateDeadend {
+    pub reason: String,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct UpdateRollout {
+    pub start_epoch: Option<i64>,
+    pub start_percentage: Option<f64>,
+    pub duration_minutes: Option<u64>,
+}