logos-blockchain/logos-blockchain-testing
1
0
Fork 0
mirror of https://github.com/logos-blockchain/logos-blockchain-testing.git synced 2026-03-31 16:23:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

Rename cfgsync artifact surface for external use

Browse Source
This commit is contained in:
andrussal 2026-03-10 11:08:17 +01:00
parent 5b69519ab1
commit 9ebf029f2a
9 changed files with 97 additions and 73 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
cfgsync/adapter/src/lib.rs
View File

@ -2,7 +2,7 @@ use std::{collections::HashMap, error::Error, sync::Mutex};
use cfgsync_artifacts::ArtifactFile;
use cfgsync_core::{
CfgSyncErrorResponse, CfgSyncPayload, ConfigResolveResponse, NodeConfigSource,
CfgSyncErrorResponse, ConfigResolveResponse, NodeArtifactsPayload, NodeConfigSource,
NodeRegistration, RegisterNodeResponse,
};
use serde::{Deserialize, Serialize};
@ -281,9 +281,9 @@ where
};
match catalog.resolve(&registration.identifier) {
Some(config) => {
ConfigResolveResponse::Config(CfgSyncPayload::from_files(config.files.clone()))
}
Some(config) => ConfigResolveResponse::Config(NodeArtifactsPayload::from_files(
config.files.clone(),
)),
None => ConfigResolveResponse::Error(CfgSyncErrorResponse::missing_config(
&registration.identifier,
)),
@ -317,7 +317,7 @@ where
let registrations = self.registration_set();
match self.materializer.materialize(&registration, &registrations) {
Ok(Some(artifacts)) => ConfigResolveResponse::Config(CfgSyncPayload::from_files(
Ok(Some(artifacts)) => ConfigResolveResponse::Config(NodeArtifactsPayload::from_files(
artifacts.files().to_vec(),
)),
Ok(None) => ConfigResolveResponse::Error(CfgSyncErrorResponse::not_ready(

20
cfgsync/core/src/bundle.rs
View File

@ -1,26 +1,32 @@
use serde::{Deserialize, Serialize};
use crate::CfgSyncFile;
use crate::NodeArtifactFile;
/// Top-level cfgsync bundle containing per-node file payloads.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CfgSyncBundle {
pub nodes: Vec<CfgSyncBundleNode>,
pub struct NodeArtifactsBundle {
pub nodes: Vec<NodeArtifactsBundleEntry>,
}
impl CfgSyncBundle {
impl NodeArtifactsBundle {
#[must_use]
pub fn new(nodes: Vec<CfgSyncBundleNode>) -> Self {
pub fn new(nodes: Vec<NodeArtifactsBundleEntry>) -> Self {
Self { nodes }
}
}
/// Artifact set for a single node resolved by identifier.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CfgSyncBundleNode {
pub struct NodeArtifactsBundleEntry {
/// Stable node identifier used by cfgsync lookup.
pub identifier: String,
/// Files that should be materialized for the node.
#[serde(default)]
pub files: Vec<CfgSyncFile>,
pub files: Vec<NodeArtifactFile>,
}
#[doc(hidden)]
pub type CfgSyncBundle = NodeArtifactsBundle;
#[doc(hidden)]
pub type CfgSyncBundleNode = NodeArtifactsBundleEntry;

8
cfgsync/core/src/client.rs
View File

@ -1,7 +1,7 @@
use serde::Serialize;
use thiserror::Error;
use crate::repo::{CfgSyncErrorResponse, CfgSyncPayload, NodeRegistration};
use crate::repo::{CfgSyncErrorResponse, NodeArtifactsPayload, NodeRegistration};
/// cfgsync client-side request/response failures.
#[derive(Debug, Error)]
@ -58,7 +58,7 @@ impl CfgSyncClient {
pub async fn fetch_node_config(
&self,
payload: &NodeRegistration,
) -> Result<CfgSyncPayload, ClientError> {
) -> Result<NodeArtifactsPayload, ClientError> {
self.post_json("/node", payload).await
}
@ -66,7 +66,7 @@ impl CfgSyncClient {
pub async fn fetch_init_with_node_config(
&self,
payload: &NodeRegistration,
) -> Result<CfgSyncPayload, ClientError> {
) -> Result<NodeArtifactsPayload, ClientError> {
self.post_json("/init-with-node", payload).await
}
@ -93,7 +93,7 @@ impl CfgSyncClient {
&self,
path: &str,
payload: &P,
) -> Result<CfgSyncPayload, ClientError> {
) -> Result<NodeArtifactsPayload, ClientError> {
let url = self.endpoint_url(path);
let response = self.http.post(url).json(payload).send().await?;

11
cfgsync/core/src/lib.rs
View File

@ -4,7 +4,9 @@ pub mod render;
pub mod repo;
pub mod server;
#[doc(hidden)]
pub use bundle::{CfgSyncBundle, CfgSyncBundleNode};
pub use bundle::{NodeArtifactsBundle, NodeArtifactsBundleEntry};
pub use client::{CfgSyncClient, ClientError, ConfigFetchStatus};
pub use render::{
CfgsyncConfigOverrides, CfgsyncOutputPaths, RenderedCfgsync, apply_cfgsync_overrides,
@ -13,13 +15,14 @@ pub use render::{
};
pub use repo::{
BundleConfigSource, BundleConfigSourceError, CFGSYNC_SCHEMA_VERSION, CfgSyncErrorCode,
CfgSyncErrorResponse, CfgSyncFile, CfgSyncPayload, ConfigResolveResponse, NodeConfigSource,
NodeRegistration, RegisterNodeResponse, RegistrationPayload, StaticConfigSource,
CfgSyncErrorResponse, ConfigResolveResponse, NodeArtifactFile, NodeArtifactsPayload,
NodeConfigSource, NodeRegistration, RegisterNodeResponse, RegistrationPayload,
StaticConfigSource,
};
#[doc(hidden)]
pub use repo::{
ConfigProvider, ConfigRepo, FileConfigProvider, FileConfigProviderError, RegistrationResponse,
RepoResponse,
CfgSyncFile, CfgSyncPayload, ConfigProvider, ConfigRepo, FileConfigProvider,
FileConfigProviderError, RegistrationResponse, RepoResponse,
};
#[doc(hidden)]
pub use server::CfgSyncState;

47
cfgsync/core/src/repo.rs
View File

@ -5,22 +5,22 @@ use serde::{Deserialize, Deserializer, Serialize, Serializer, de::DeserializeOwn
use serde_json::Value;
use thiserror::Error;
use crate::{CfgSyncBundle, CfgSyncBundleNode};
use crate::{NodeArtifactsBundle, NodeArtifactsBundleEntry};
/// Schema version served by cfgsync payload responses.
pub const CFGSYNC_SCHEMA_VERSION: u16 = 1;
/// Canonical cfgsync file type used in payloads and bundles.
pub type CfgSyncFile = ArtifactFile;
pub type NodeArtifactFile = ArtifactFile;
/// Payload returned by cfgsync server for one node.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CfgSyncPayload {
pub struct NodeArtifactsPayload {
/// Payload schema version for compatibility checks.
pub schema_version: u16,
/// Files that must be written on the target node.
#[serde(default)]
pub files: Vec<CfgSyncFile>,
pub files: Vec<NodeArtifactFile>,
}
/// Adapter-owned registration payload stored alongside a generic node identity.
@ -137,9 +137,9 @@ impl NodeRegistration {
}
}
impl CfgSyncPayload {
impl NodeArtifactsPayload {
#[must_use]
pub fn from_files(files: Vec<CfgSyncFile>) -> Self {
pub fn from_files(files: Vec<NodeArtifactFile>) -> Self {
Self {
schema_version: CFGSYNC_SCHEMA_VERSION,
files,
@ -147,7 +147,7 @@ impl CfgSyncPayload {
}
#[must_use]
pub fn files(&self) -> &[CfgSyncFile] {
pub fn files(&self) -> &[NodeArtifactFile] {
&self.files
}
@ -201,7 +201,7 @@ impl CfgSyncErrorResponse {
/// Resolution outcome for a requested node identifier.
pub enum ConfigResolveResponse {
Config(CfgSyncPayload),
Config(NodeArtifactsPayload),
Error(CfgSyncErrorResponse),
}
@ -220,12 +220,12 @@ pub trait NodeConfigSource: Send + Sync {
/// In-memory map-backed source used by cfgsync server state.
pub struct StaticConfigSource {
configs: HashMap<String, CfgSyncPayload>,
configs: HashMap<String, NodeArtifactsPayload>,
}
impl StaticConfigSource {
#[must_use]
pub fn from_bundle(configs: HashMap<String, CfgSyncPayload>) -> Arc<Self> {
pub fn from_bundle(configs: HashMap<String, NodeArtifactsPayload>) -> Arc<Self> {
Arc::new(Self { configs })
}
}
@ -273,19 +273,19 @@ pub enum BundleLoadError {
}
#[must_use]
pub fn bundle_to_payload_map(bundle: CfgSyncBundle) -> HashMap<String, CfgSyncPayload> {
pub fn bundle_to_payload_map(bundle: NodeArtifactsBundle) -> HashMap<String, NodeArtifactsPayload> {
bundle
.nodes
.into_iter()
.map(|node| {
let CfgSyncBundleNode { identifier, files } = node;
let NodeArtifactsBundleEntry { identifier, files } = node;
(identifier, CfgSyncPayload::from_files(files))
(identifier, NodeArtifactsPayload::from_files(files))
})
.collect()
}
pub fn load_bundle(path: &Path) -> Result<CfgSyncBundle, BundleLoadError> {
pub fn load_bundle(path: &Path) -> Result<NodeArtifactsBundle, BundleLoadError> {
let path_string = path.display().to_string();
let raw = fs::read_to_string(path).map_err(|source| BundleLoadError::ReadBundle {
path: path_string.clone(),
@ -337,8 +337,8 @@ mod tests {
assert_eq!(typed.service, "blend");
}
fn sample_payload() -> CfgSyncPayload {
CfgSyncPayload::from_files(vec![CfgSyncFile::new("/config.yaml", "key: value")])
fn sample_payload() -> NodeArtifactsPayload {
NodeArtifactsPayload::from_files(vec![NodeArtifactFile::new("/config.yaml", "key: value")])
}
#[test]
@ -455,7 +455,7 @@ impl BundleConfigSource {
source,
})?;
let bundle: CfgSyncBundle =
let bundle: NodeArtifactsBundle =
serde_yaml::from_str(&raw).map_err(|source| BundleConfigSourceError::Parse {
path: path.display().to_string(),
source,
@ -483,8 +483,11 @@ impl NodeConfigSource for BundleConfigSource {
}
}
fn payload_from_bundle_node(node: CfgSyncBundleNode) -> (String, CfgSyncPayload) {
(node.identifier, CfgSyncPayload::from_files(node.files))
fn payload_from_bundle_node(node: NodeArtifactsBundleEntry) -> (String, NodeArtifactsPayload) {
(
node.identifier,
NodeArtifactsPayload::from_files(node.files),
)
}
#[doc(hidden)]
@ -506,3 +509,9 @@ pub type FileConfigProvider = BundleConfigSource;
#[doc(hidden)]
pub type FileConfigProviderError = BundleConfigSourceError;
#[doc(hidden)]
pub type CfgSyncFile = NodeArtifactFile;
#[doc(hidden)]
pub type CfgSyncPayload = NodeArtifactsPayload;

14
cfgsync/core/src/server.rs
View File

@ -119,12 +119,12 @@ mod tests {
use super::{CfgsyncServerState, NodeRegistration, node_config, register_node};
use crate::repo::{
CFGSYNC_SCHEMA_VERSION, CfgSyncErrorCode, CfgSyncErrorResponse, CfgSyncFile,
CfgSyncPayload, ConfigResolveResponse, NodeConfigSource, RegisterNodeResponse,
CFGSYNC_SCHEMA_VERSION, CfgSyncErrorCode, CfgSyncErrorResponse, ConfigResolveResponse,
NodeArtifactFile, NodeArtifactsPayload, NodeConfigSource, RegisterNodeResponse,
};
struct StaticProvider {
data: HashMap<String, CfgSyncPayload>,
data: HashMap<String, NodeArtifactsPayload>,
}
impl NodeConfigSource for StaticProvider {
@ -154,7 +154,7 @@ mod tests {
}
struct RegistrationAwareProvider {
data: HashMap<String, CfgSyncPayload>,
data: HashMap<String, NodeArtifactsPayload>,
registrations: std::sync::Mutex<HashMap<String, NodeRegistration>>,
}
@ -201,10 +201,10 @@ mod tests {
}
}
fn sample_payload() -> CfgSyncPayload {
CfgSyncPayload {
fn sample_payload() -> NodeArtifactsPayload {
NodeArtifactsPayload {
schema_version: CFGSYNC_SCHEMA_VERSION,
files: vec![CfgSyncFile::new("/app-config.yaml", "app: test")],
files: vec![NodeArtifactFile::new("/app-config.yaml", "app: test")],
}
}

36
cfgsync/runtime/src/client.rs
View File

@ -6,8 +6,8 @@ use std::{
use anyhow::{Context as _, Result, bail};
use cfgsync_core::{
CFGSYNC_SCHEMA_VERSION, CfgSyncClient, CfgSyncFile, CfgSyncPayload, NodeRegistration,
RegistrationPayload,
CFGSYNC_SCHEMA_VERSION, CfgSyncClient, NodeArtifactFile, NodeArtifactsPayload,
NodeRegistration, RegistrationPayload,
};
use thiserror::Error;
use tokio::time::{Duration, sleep};
@ -22,7 +22,10 @@ enum ClientEnvError {
InvalidIp { value: String },
}
async fn fetch_with_retry(payload: &NodeRegistration, server_addr: &str) -> Result<CfgSyncPayload> {
async fn fetch_with_retry(
payload: &NodeRegistration,
server_addr: &str,
) -> Result<NodeArtifactsPayload> {
let client = CfgSyncClient::new(server_addr);
for attempt in 1..=FETCH_ATTEMPTS {
@ -43,7 +46,10 @@ async fn fetch_with_retry(payload: &NodeRegistration, server_addr: &str) -> Resu
unreachable!("cfgsync fetch loop always returns before exhausting attempts");
}
async fn fetch_once(client: &CfgSyncClient, payload: &NodeRegistration) -> Result<CfgSyncPayload> {
async fn fetch_once(
client: &CfgSyncClient,
payload: &NodeRegistration,
) -> Result<NodeArtifactsPayload> {
let response = client.fetch_node_config(payload).await?;
Ok(response)
@ -92,7 +98,7 @@ async fn register_node(payload: &NodeRegistration, server_addr: &str) -> Result<
unreachable!("cfgsync register loop always returns before exhausting attempts");
}
fn ensure_schema_version(config: &CfgSyncPayload) -> Result<()> {
fn ensure_schema_version(config: &NodeArtifactsPayload) -> Result<()> {
if config.schema_version != CFGSYNC_SCHEMA_VERSION {
bail!(
"unsupported cfgsync payload schema version {}, expected {}",
@ -104,7 +110,7 @@ fn ensure_schema_version(config: &CfgSyncPayload) -> Result<()> {
Ok(())
}
fn collect_payload_files(config: &CfgSyncPayload) -> Result<&[CfgSyncFile]> {
fn collect_payload_files(config: &NodeArtifactsPayload) -> Result<&[NodeArtifactFile]> {
if config.is_empty() {
bail!("cfgsync payload contains no files");
}
@ -112,7 +118,7 @@ fn collect_payload_files(config: &CfgSyncPayload) -> Result<&[CfgSyncFile]> {
Ok(config.files())
}
fn write_cfgsync_file(file: &CfgSyncFile) -> Result<()> {
fn write_cfgsync_file(file: &NodeArtifactFile) -> Result<()> {
let path = PathBuf::from(&file.path);
ensure_parent_dir(&path)?;
@ -181,8 +187,8 @@ mod tests {
use std::collections::HashMap;
use cfgsync_core::{
CfgSyncBundle, CfgSyncBundleNode, CfgSyncPayload, CfgsyncServerState, StaticConfigSource,
run_cfgsync,
CfgsyncServerState, NodeArtifactsBundle, NodeArtifactsBundleEntry, NodeArtifactsPayload,
StaticConfigSource, run_cfgsync,
};
use tempfile::tempdir;
@ -194,11 +200,11 @@ mod tests {
let app_config_path = dir.path().join("config.yaml");
let deployment_path = dir.path().join("deployment.yaml");
let bundle = CfgSyncBundle::new(vec![CfgSyncBundleNode {
let bundle = NodeArtifactsBundle::new(vec![NodeArtifactsBundleEntry {
identifier: "node-1".to_owned(),
files: vec![
CfgSyncFile::new(app_config_path.to_string_lossy(), "app_key: app_value"),
CfgSyncFile::new(deployment_path.to_string_lossy(), "mode: local"),
NodeArtifactFile::new(app_config_path.to_string_lossy(), "app_key: app_value"),
NodeArtifactFile::new(deployment_path.to_string_lossy(), "mode: local"),
],
}]);
@ -227,14 +233,14 @@ mod tests {
assert_eq!(deployment, "mode: local");
}
fn bundle_to_payload_map(bundle: CfgSyncBundle) -> HashMap<String, CfgSyncPayload> {
fn bundle_to_payload_map(bundle: NodeArtifactsBundle) -> HashMap<String, NodeArtifactsPayload> {
bundle
.nodes
.into_iter()
.map(|node| {
let CfgSyncBundleNode { identifier, files } = node;
let NodeArtifactsBundleEntry { identifier, files } = node;
(identifier, CfgSyncPayload::from_files(files))
(identifier, NodeArtifactsPayload::from_files(files))
})
.collect()
}

6
cfgsync/runtime/src/server.rs
View File

@ -3,7 +3,7 @@ use std::{fs, path::Path, sync::Arc};
use anyhow::Context as _;
use cfgsync_adapter::{NodeArtifacts, NodeArtifactsCatalog, RegistrationConfigProvider};
use cfgsync_core::{
BundleConfigSource, CfgSyncBundle, CfgsyncServerState, NodeConfigSource, run_cfgsync,
BundleConfigSource, CfgsyncServerState, NodeArtifactsBundle, NodeConfigSource, run_cfgsync,
};
use serde::Deserialize;
use thiserror::Error;
@ -120,7 +120,7 @@ fn load_materializing_provider(bundle_path: &Path) -> anyhow::Result<Arc<dyn Nod
Ok(Arc::new(provider))
}
fn load_bundle_yaml(bundle_path: &Path) -> anyhow::Result<CfgSyncBundle> {
fn load_bundle_yaml(bundle_path: &Path) -> anyhow::Result<NodeArtifactsBundle> {
let raw = fs::read_to_string(bundle_path)
.with_context(|| format!("reading cfgsync bundle from {}", bundle_path.display()))?;
@ -128,7 +128,7 @@ fn load_bundle_yaml(bundle_path: &Path) -> anyhow::Result<CfgSyncBundle> {
.with_context(|| format!("parsing cfgsync bundle from {}", bundle_path.display()))
}
fn build_node_catalog(bundle: CfgSyncBundle) -> NodeArtifactsCatalog {
fn build_node_catalog(bundle: NodeArtifactsBundle) -> NodeArtifactsCatalog {
let nodes = bundle
.nodes
.into_iter()

18
logos/runtime/ext/src/cfgsync/mod.rs
View File

@ -2,7 +2,7 @@ use anyhow::Result;
use cfgsync_adapter::{CfgsyncEnv, build_cfgsync_node_catalog};
pub(crate) use cfgsync_core::render::CfgsyncOutputPaths;
use cfgsync_core::{
CfgSyncBundle, CfgSyncBundleNode,
NodeArtifactsBundle, NodeArtifactsBundleEntry,
render::{
CfgsyncConfigOverrides, RenderedCfgsync, ensure_bundle_path,
render_cfgsync_yaml_from_template, write_rendered_cfgsync,
@ -48,20 +48,20 @@ pub(crate) fn render_cfgsync_from_template<E: CfgsyncEnv>(
fn build_cfgsync_bundle<E: CfgsyncEnv>(
topology: &E::Deployment,
hostnames: &[String],
) -> Result<CfgSyncBundle> {
) -> Result<NodeArtifactsBundle> {
let nodes = build_cfgsync_node_catalog::<E>(topology, hostnames)?.into_configs();
let nodes = nodes
.into_iter()
.map(|node| CfgSyncBundleNode {
.map(|node| NodeArtifactsBundleEntry {
identifier: node.identifier,
files: node.files,
})
.collect();
Ok(CfgSyncBundle::new(nodes))
Ok(NodeArtifactsBundle::new(nodes))
}
fn append_deployment_files(bundle: &mut CfgSyncBundle) -> Result<()> {
fn append_deployment_files(bundle: &mut NodeArtifactsBundle) -> Result<()> {
for node in &mut bundle.nodes {
if has_file_path(node, "/deployment.yaml") {
continue;
@ -80,18 +80,18 @@ fn append_deployment_files(bundle: &mut CfgSyncBundle) -> Result<()> {
Ok(())
}
fn has_file_path(node: &CfgSyncBundleNode, path: &str) -> bool {
fn has_file_path(node: &NodeArtifactsBundleEntry, path: &str) -> bool {
node.files.iter().any(|file| file.path == path)
}
fn config_file_content(node: &CfgSyncBundleNode) -> Option<String> {
fn config_file_content(node: &NodeArtifactsBundleEntry) -> Option<String> {
node.files
.iter()
.find_map(|file| (file.path == "/config.yaml").then_some(file.content.clone()))
}
fn build_bundle_file(path: &str, content: String) -> cfgsync_core::CfgSyncFile {
cfgsync_core::CfgSyncFile {
fn build_bundle_file(path: &str, content: String) -> cfgsync_core::NodeArtifactFile {
cfgsync_core::NodeArtifactFile {
path: path.to_owned(),
content,
}