Clean up
Some checks failed
Build and Publish Docker Image / test (push) Failing after 7s
Build and Publish Docker Image / build-and-push (push) Has been skipped
Build and Publish Docker Image / security-scan (push) Has been skipped

This commit is contained in:
Andras Schmelczer 2026-06-06 19:41:55 +01:00
commit b414c05400
8 changed files with 868 additions and 948 deletions

View file

@ -2,8 +2,13 @@ use crate::registry::ImageRef;
use anyhow::Result;
use regex::Regex;
use std::fs;
use std::sync::LazyLock;
use tracing::{info, warn};
static IMAGE_REGEX: LazyLock<Regex> = LazyLock::new(|| {
Regex::new(r#"^\s*image:\s*(?:["']([^"']+)["']|([^\s#]+))\s*(#.*)?$"#).unwrap()
});
#[derive(Debug, Clone)]
pub struct ServiceImage {
pub service_name: String,
@ -27,7 +32,6 @@ impl ComposeParser {
pub fn parse_file(&self, file_path: &str) -> Result<ComposeFile> {
let content = fs::read_to_string(file_path)?;
let services = self.extract_services(&content)?;
Ok(ComposeFile { content, services })
}
@ -36,41 +40,77 @@ impl ComposeParser {
let lines: Vec<&str> = content.lines().collect();
let mut in_services = false;
let mut current_service: Option<String> = None;
let image_regex = Regex::new(r#"^\s*image:\s*(?:["']([^"']+)["']|([^\s#]+))\s*(#.*)?$"#)?;
let mut service_indent: Option<usize> = None;
let mut service_child_indent: Option<usize> = None;
for (line_number, line) in lines.iter().enumerate() {
if line.trim_start().starts_with("services:") {
in_services = true;
let trimmed = line.trim_start();
if trimmed.is_empty() || trimmed.starts_with('#') {
continue;
}
if in_services
&& line.chars().next().is_some_and(|c| c.is_alphabetic())
&& !line.starts_with(" ")
{
let indent = line.len() - trimmed.len();
if indent == 0 && trimmed.starts_with("services:") {
in_services = true;
current_service = None;
service_indent = None;
service_child_indent = None;
continue;
}
if in_services && indent == 0 {
in_services = false;
current_service = None;
service_indent = None;
service_child_indent = None;
continue;
}
if in_services {
if let Some(service_name) = self.extract_service_name(line) {
current_service = Some(service_name);
continue;
}
if !in_services {
continue;
}
if let Some(ref service_name) = current_service {
if let Some(image_ref) = self.extract_image_from_line(line, &image_regex)? {
info!(
"Found service '{}' with image '{}'",
service_name, image_ref
);
services.push(ServiceImage {
service_name: service_name.clone(),
image_ref,
original_line: line.to_string(),
line_number,
});
if (service_indent.is_none() || indent == service_indent.unwrap())
&& trimmed.ends_with(':')
&& !trimmed.contains(' ')
&& trimmed
.trim_end_matches(':')
.chars()
.all(|c| c.is_alphanumeric() || c == '_' || c == '-' || c == '.')
{
let name = trimmed.trim_end_matches(':');
if !name.is_empty() {
service_indent = Some(indent);
service_child_indent = None;
current_service = Some(name.to_string());
}
continue;
}
if let Some(ref service_name) = current_service {
if let Some(si) = service_indent {
if indent > si {
// The first key under a service fixes the direct-child
// indent; only keys at that exact level are the service's
// own properties. This avoids picking up an `image:` nested
// under `build:` or other deeper mappings.
let child_indent = *service_child_indent.get_or_insert(indent);
if indent == child_indent {
if let Some(image_ref) = self.extract_image_from_line(line)? {
info!(
"Found service '{}' with image '{}'",
service_name, image_ref
);
services.push(ServiceImage {
service_name: service_name.clone(),
image_ref,
original_line: line.to_string(),
line_number,
});
}
}
}
}
}
@ -79,62 +119,37 @@ impl ComposeParser {
Ok(services)
}
fn extract_service_name(&self, line: &str) -> Option<String> {
fn extract_image_from_line(&self, line: &str) -> Result<Option<ImageRef>> {
let trimmed = line.trim_start();
let indent_level = line.len() - trimmed.len();
if indent_level > 0 && indent_level <= 8 && trimmed.ends_with(':') && !trimmed.contains(' ')
{
let potential_service = trimmed.trim_end_matches(':');
if !potential_service.is_empty()
&& potential_service
.chars()
.all(|c| c.is_alphanumeric() || c == '_' || c == '-')
{
return Some(potential_service.to_string());
}
if !trimmed.starts_with("image:") {
return Ok(None);
}
None
}
let Some(captures) = IMAGE_REGEX.captures(line) else {
return Ok(None);
};
fn extract_image_from_line(&self, line: &str, image_regex: &Regex) -> Result<Option<ImageRef>> {
let trimmed = line.trim_start();
let indent_level = line.len() - trimmed.len();
let image_str = captures
.get(1)
.or_else(|| captures.get(2))
.unwrap()
.as_str();
if indent_level > 2 && trimmed.starts_with("image:") {
if let Some(captures) = image_regex.captures(line) {
let image_str = captures
.get(1)
.or_else(|| captures.get(2))
.unwrap()
.as_str();
if image_str.contains('@') {
info!("Skipping digest-pinned image: {}", image_str);
return Ok(None);
}
if image_str.find('@').is_some() {
info!("Found digest in image string, skipping: {}", image_str);
return Ok(None);
}
if image_str.trim().is_empty() {
return Ok(None);
}
if image_str.trim().is_empty()
|| image_str.trim() == "\"\""
|| image_str.trim() == "''"
{
info!("Empty image string found in line: {}, skipping", line);
return Ok(None);
}
match ImageRef::parse(image_str) {
Ok(image_ref) => Ok(Some(image_ref)),
Err(e) => {
warn!("Failed to parse image '{}': {}", image_str, e);
Ok(None)
}
}
} else {
match ImageRef::parse(image_str) {
Ok(image_ref) => Ok(Some(image_ref)),
Err(e) => {
warn!("Failed to parse image '{}': {}", image_str, e);
Ok(None)
}
} else {
Ok(None)
}
}
}
@ -160,12 +175,12 @@ services:
image: nginx:1.21.0 # Web server
ports:
- "80:80"
db:
image: postgres:13.7
environment:
POSTGRES_DB: myapp
redis:
image: redis:6.2-alpine
"#;
@ -183,4 +198,52 @@ services:
assert_eq!(result.services[0].image_ref.name, "nginx");
assert_eq!(result.services[0].image_ref.tag, "1.21.0");
}
#[test]
fn test_service_names_with_dots() {
let compose_content = r#"
services:
my.service:
image: nginx:1.21.0
another-service:
image: redis:7.0
"#;
let mut temp_file = NamedTempFile::new().unwrap();
temp_file.write_all(compose_content.as_bytes()).unwrap();
let parser = ComposeParser::new();
let result = parser
.parse_file(temp_file.path().to_str().unwrap())
.unwrap();
assert_eq!(result.services.len(), 2);
assert_eq!(result.services[0].service_name, "my.service");
assert_eq!(result.services[1].service_name, "another-service");
}
#[test]
fn test_ignores_image_nested_under_build() {
let compose_content = r#"
services:
web:
build:
context: .
image: should-not-match:9.9.9
image: nginx:1.21.0
"#;
let mut temp_file = NamedTempFile::new().unwrap();
temp_file.write_all(compose_content.as_bytes()).unwrap();
let parser = ComposeParser::new();
let result = parser
.parse_file(temp_file.path().to_str().unwrap())
.unwrap();
assert_eq!(result.services.len(), 1);
assert_eq!(result.services[0].service_name, "web");
assert_eq!(result.services[0].image_ref.name, "nginx");
assert_eq!(result.services[0].image_ref.tag, "1.21.0");
}
}

View file

@ -1,38 +1,49 @@
use super::parser::{ComposeFile, ComposeParser, ServiceImage};
use crate::config::Config;
use crate::registry::Client as RegistryClient;
use crate::config::{Config, UpdateStrategy};
use crate::registry::{Client as RegistryClient, ImageRef};
use crate::strategy::create_selector;
use crate::version::parse_version_tag;
use crate::version::{parse_version_tag, VersionInfo};
use anyhow::{anyhow, Result};
use regex::Regex;
use semver::Version;
use std::collections::HashSet;
use std::fs;
use std::path::{Path, PathBuf};
use std::sync::LazyLock;
use tracing::{debug, info, warn};
static IMAGE_LINE_REGEX: LazyLock<Regex> = LazyLock::new(|| {
Regex::new(r#"^(\s*image:\s*)(?:["']([^"']+)["']|([^\s#]+))(\s*(?:#.*)?)$"#).unwrap()
});
pub struct ComposeUpdater {
config: Config,
registry_client: RegistryClient,
parser: ComposeParser,
}
pub struct UpdateReport {
pub files_found: usize,
pub updated_files: Vec<String>,
}
impl ComposeUpdater {
pub fn new(config: Config) -> Self {
let registry_client = RegistryClient::new(config.clone());
let parser = ComposeParser::new();
Self {
config,
registry_client,
parser,
parser: ComposeParser::new(),
}
}
pub async fn update_all_compose_files(&self) -> Result<Vec<String>> {
pub async fn update_all_compose_files(&self) -> Result<UpdateReport> {
let mut updated_files = Vec::new();
let mut files_found = 0;
for compose_path in &self.config.compose_paths {
let compose_files = self.find_compose_files(compose_path)?;
files_found += compose_files.len();
for file_path in compose_files {
let updated = self.update_compose_file(&file_path).await?;
@ -42,7 +53,10 @@ impl ComposeUpdater {
}
}
Ok(updated_files)
Ok(UpdateReport {
files_found,
updated_files,
})
}
pub fn parse_compose_file(&self, file_path: &str) -> Result<ComposeFile> {
@ -58,7 +72,7 @@ impl ComposeUpdater {
for service in &compose_file.services {
if self.config.is_image_ignored(&service.image_ref.to_string()) {
info!("Skipping ignored image: {}", service.image_ref.to_string());
debug!("Skipping ignored image: {}", service.image_ref);
continue;
}
@ -69,16 +83,13 @@ impl ComposeUpdater {
updated = true;
info!(
"Updated {}: {} -> {}",
service.service_name,
service.image_ref.to_string(),
new_image
service.service_name, service.image_ref, new_image
);
}
Ok(None) => {
debug!(
"No update needed for {}: {}",
service.service_name,
service.image_ref.to_string()
service.service_name, service.image_ref
);
}
Err(e) => {
@ -97,57 +108,45 @@ impl ComposeUpdater {
}
fn write_updated_content(&self, file_path: &str, content: String) -> Result<()> {
if !self.config.dry_run {
if self.config.dry_run {
info!("Dry run: would update {}", file_path);
} else {
fs::write(file_path, content)?;
info!("Updated compose file: {}", file_path);
} else {
info!("Dry run: Would update compose file: {}", file_path);
}
Ok(())
}
async fn update_service_image(&self, service: &ServiceImage) -> Result<Option<String>> {
info!(
"Checking for updates for service: {} (current image: {})",
service.service_name,
service.image_ref.to_string()
);
let (current_version, current_prefix, current_suffix, _) =
parse_version_tag(&service.image_ref.tag);
let Some(current_version) = current_version else {
debug!(
"Skipping non-semver tag '{}' for service {}",
service.image_ref.tag, service.service_name
);
return Ok(None);
};
let available_versions = self
.registry_client
.get_available_versions(&service.image_ref)
.await?;
if available_versions.is_empty() {
warn!("No versions available for selection");
warn!("No versions available for {}", service.image_ref);
return Ok(None);
}
let selector = create_selector(&self.config.update_strategy);
if let Some(target_version_info) =
selector.select_target_version(&available_versions, current_prefix, current_suffix)
{
// Only update if the target version is different AND higher than the current version
// This prevents downgrades
if Some(target_version_info.version.clone()) != current_version {
if let Some(ref current_ver) = current_version {
if target_version_info.version < *current_ver {
info!(
"Skipping downgrade from {} to {}",
current_ver, target_version_info.version
);
return Ok(None);
}
}
let mut new_image_ref = service.image_ref.clone();
new_image_ref.tag = target_version_info.to_string();
return Ok(Some(new_image_ref.to_string()));
}
}
Ok(None)
Ok(choose_new_tag(
&service.image_ref,
&current_version,
&available_versions,
current_prefix,
current_suffix,
&self.config.update_strategy,
))
}
pub fn replace_image_in_content(
@ -156,49 +155,44 @@ impl ComposeUpdater {
service: &ServiceImage,
new_image: &str,
) -> Result<String> {
let image_regex =
Regex::new(r#"^(\s*image:\s*)(?:["']([^"']+)["']|([^\s#]+))(\s*(?:#.*)?)$"#)?;
if let Some(captures) = image_regex.captures(&service.original_line) {
let prefix = captures.get(1).unwrap().as_str();
let _old_image = captures
.get(2)
.or_else(|| captures.get(3))
.unwrap()
.as_str();
let suffix = captures.get(4).unwrap().as_str();
let image_part = if captures.get(2).is_some() {
format!("\"{new_image}\"")
} else {
new_image.to_string()
};
let new_line = format!("{prefix}{image_part}{suffix}");
let lines: Vec<&str> = content.lines().collect();
if service.line_number < lines.len()
&& lines[service.line_number] == service.original_line
{
let mut result_lines = lines;
result_lines[service.line_number] = &new_line;
let mut result = result_lines.join("\n");
if content.ends_with('\n') {
result.push('\n');
}
Ok(result)
} else {
Err(anyhow!(
"Line number mismatch or content changed for service: {}",
service.service_name
))
}
} else {
Err(anyhow!(
let Some(captures) = IMAGE_LINE_REGEX.captures(&service.original_line) else {
return Err(anyhow!(
"Could not parse image line: {}",
service.original_line
));
};
let prefix = captures.get(1).unwrap().as_str();
let suffix = captures.get(4).unwrap().as_str();
let was_quoted = captures.get(2).is_some();
let image_part = if was_quoted {
format!("\"{new_image}\"")
} else {
new_image.to_string()
};
let new_line = format!("{prefix}{image_part}{suffix}");
let lines: Vec<&str> = content.lines().collect();
if service.line_number < lines.len() && lines[service.line_number] == service.original_line
{
let mut result_lines = lines;
result_lines[service.line_number] = &new_line;
let mut result = result_lines.join("\n");
if content.ends_with('\n') {
result.push('\n');
}
Ok(result)
} else {
Err(anyhow!(
"Line mismatch for service '{}': line {} expected '{}', got '{}'",
service.service_name,
service.line_number,
service.original_line,
lines.get(service.line_number).unwrap_or(&"<out of bounds>")
))
}
}
@ -256,38 +250,111 @@ impl ComposeUpdater {
}
}
/// Chooses the new image string for `image_ref`, or `None` when no suitable
/// upgrade exists. The selected version must share the current tag's prefix and
/// suffix and be strictly higher than `current_version`, which prevents both
/// downgrades and no-op rewrites to the same version.
fn choose_new_tag(
image_ref: &ImageRef,
current_version: &Version,
available_versions: &[VersionInfo],
current_prefix: Option<String>,
current_suffix: Option<String>,
strategy: &UpdateStrategy,
) -> Option<String> {
let selector = create_selector(strategy);
let target =
selector.select_target_version(available_versions, current_prefix, current_suffix)?;
if target.version <= *current_version {
return None;
}
let mut new_image_ref = image_ref.clone();
new_image_ref.tag = target.to_string();
Some(new_image_ref.to_string())
}
#[cfg(test)]
mod tests {
use super::*;
use crate::config::{Config, UpdateStrategy};
use std::io::Write;
use tempfile::NamedTempFile;
#[tokio::test]
async fn test_prevents_downgrade() {
let mut config = Config::default();
config.update_strategy = UpdateStrategy::Latest;
config.dry_run = true;
fn version_infos(tags: &[&str]) -> Vec<VersionInfo> {
tags.iter()
.map(|t| VersionInfo::from_tag(t).unwrap())
.collect()
}
let updater = ComposeUpdater::new(config);
#[test]
fn test_choose_new_tag_upgrades_to_higher_version() {
let image = ImageRef::parse("nginx:1.25.0").unwrap();
let current = Version::parse("1.25.0").unwrap();
let available = version_infos(&["1.25.0", "1.26.0", "1.24.0"]);
// Create a temporary compose file with a higher version
let mut temp_file = NamedTempFile::new().unwrap();
writeln!(temp_file, "services:\n web:\n image: nginx:1.25.0").unwrap();
let result = choose_new_tag(
&image,
&current,
&available,
None,
None,
&UpdateStrategy::Latest,
);
assert_eq!(result, Some("nginx:1.26.0".to_string()));
}
let file_path = temp_file.path().to_str().unwrap();
let compose_file = updater.parse_compose_file(file_path).unwrap();
#[test]
fn test_choose_new_tag_prevents_downgrade() {
let image = ImageRef::parse("nginx:1.25.0").unwrap();
let current = Version::parse("1.25.0").unwrap();
// Registry only offers older versions.
let available = version_infos(&["1.24.0", "1.23.5"]);
assert_eq!(compose_file.services.len(), 1);
let service = &compose_file.services[0];
let result = choose_new_tag(
&image,
&current,
&available,
None,
None,
&UpdateStrategy::Latest,
);
assert!(result.is_none(), "should never downgrade");
}
// Mock a scenario where the strategy selects a lower version
// This would happen if available versions only include older versions
let (current_version, _, _, _) = parse_version_tag(&service.image_ref.tag);
assert!(current_version.is_some());
#[test]
fn test_choose_new_tag_skips_equal_version() {
let image = ImageRef::parse("nginx:1.25.0").unwrap();
let current = Version::parse("1.25.0").unwrap();
let available = version_infos(&["1.25.0"]);
// The actual test would need mocked registry responses, but we can verify
// the logic by checking that current version is properly extracted
assert_eq!(current_version.unwrap().to_string(), "1.25.0");
let result = choose_new_tag(
&image,
&current,
&available,
None,
None,
&UpdateStrategy::Latest,
);
assert!(result.is_none(), "no rewrite when already on the latest");
}
#[test]
fn test_choose_new_tag_respects_prefix_and_suffix() {
let image = ImageRef::parse("nginx:v1.25.0-alpine").unwrap();
let current = Version::parse("1.25.0").unwrap();
let available = vec![
VersionInfo::from_tag("v1.26.0-alpine").unwrap(),
VersionInfo::from_tag("1.27.0").unwrap(), // no matching prefix/suffix
VersionInfo::from_tag("v1.26.0").unwrap(), // missing suffix
];
let result = choose_new_tag(
&image,
&current,
&available,
Some("v".to_string()),
Some("-alpine".to_string()),
&UpdateStrategy::Latest,
);
assert_eq!(result, Some("nginx:v1.26.0-alpine".to_string()));
}
}

View file

@ -1,22 +1,22 @@
use anyhow::Result;
use regex::Regex;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::path::PathBuf;
use tracing::info;
use std::sync::LazyLock;
use tracing::{info, warn};
static ENV_VAR_REGEX: LazyLock<Regex> =
LazyLock::new(|| Regex::new(r"\$\{([^}]+)\}|\$([A-Za-z_][A-Za-z0-9_]*)").unwrap());
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default)]
pub struct Config {
#[serde(default)]
pub compose_paths: Vec<PathBuf>,
#[serde(default)]
pub schedule: String,
#[serde(default)]
pub registries: HashMap<String, RegistryConfig>,
#[serde(default)]
pub update_strategy: UpdateStrategy,
#[serde(default)]
pub ignore_images: Vec<String>,
#[serde(default)]
pub dry_run: bool,
}
@ -53,7 +53,7 @@ impl Default for Config {
Self {
compose_paths: vec![PathBuf::from(".")],
schedule: "0 0 2 * * *".to_string(), // Daily at 2 AM
schedule: "0 0 2 * * *".to_string(),
registries,
update_strategy: UpdateStrategy::LatestPatchOfPreviousMinor,
ignore_images: vec![],
@ -68,31 +68,32 @@ impl Config {
let content = std::fs::read_to_string(config_path)?;
let expanded_content = Self::expand_env_vars(&content);
let mut config: Self = serde_yaml::from_str(&expanded_content)?;
config.resolve_env_tokens();
config.normalize_auth_tokens();
Ok(config)
}
/// Expand environment variable placeholders like ${VAR} in the config content
pub fn expand_env_vars(content: &str) -> String {
let env_var_pattern = regex::Regex::new(r"\$\{([^}]+)\}").unwrap();
env_var_pattern
ENV_VAR_REGEX
.replace_all(content, |caps: &regex::Captures| {
let var_name = &caps[1];
std::env::var(var_name).unwrap_or_else(|_| format!("${{{var_name}}}"))
let var_name = caps.get(1).or_else(|| caps.get(2)).unwrap().as_str();
std::env::var(var_name).unwrap_or_else(|_| {
warn!(
"Environment variable `{}` referenced in config is not set; substituting an empty string",
var_name
);
String::new()
})
})
.to_string()
.into_owned()
}
/// Resolve environment variable tokens after deserialization
fn resolve_env_tokens(&mut self) {
for registry_config in self.registries.values_mut() {
if let Some(token) = &registry_config.auth_token {
if token.starts_with("$") {
// Handle direct env var references like "$GITHUB_TOKEN"
let env_var_name = token.trim_start_matches('$');
registry_config.auth_token = std::env::var(env_var_name).ok();
}
/// Treat a registry `auth_token` that expanded to an empty string (e.g. an
/// unset `${TOKEN}`) as no token at all, so we fall back to unauthenticated
/// access with a clear error instead of attempting auth with an empty token.
fn normalize_auth_tokens(&mut self) {
for registry in self.registries.values_mut() {
if registry.auth_token.as_deref() == Some("") {
registry.auth_token = None;
}
}
}
@ -100,6 +101,6 @@ impl Config {
pub fn is_image_ignored(&self, image: &str) -> bool {
self.ignore_images
.iter()
.any(|ignored| image.contains(ignored))
.any(|pattern| image.contains(pattern))
}
}

View file

@ -14,22 +14,18 @@ pub struct HealthHandle {
last_update_success: Arc<Mutex<bool>>,
}
impl Default for HealthServer {
fn default() -> Self {
Self::new().0
}
}
impl HealthServer {
pub fn new() -> (Self, HealthHandle) {
let last_update_success = Arc::new(Mutex::new(true));
let server = Self {
let handle = HealthHandle {
last_update_success: last_update_success.clone(),
};
let handle = HealthHandle {
last_update_success,
};
(server, handle)
(
Self {
last_update_success,
},
handle,
)
}
pub async fn start(self) -> anyhow::Result<()> {
@ -37,64 +33,56 @@ impl HealthServer {
info!("Health server listening on port 8080");
loop {
let (mut socket, _) = listener.accept().await?;
let (socket, _) = match listener.accept().await {
Ok(conn) => conn,
Err(e) => {
warn!("Failed to accept connection: {}", e);
continue;
}
};
let health_status = self.last_update_success.clone();
tokio::spawn(async move {
// Set timeout for the entire request handling (5 seconds)
let handle_request = async {
let mut buffer = [0; 1024];
// Read with timeout to prevent hanging connections
match timeout(Duration::from_secs(5), socket.read(&mut buffer)).await {
Ok(Ok(_)) => {
// Successfully read request (we don't need to parse it for health check)
}
Ok(Err(_)) | Err(_) => {
warn!("Health check request read timeout or error");
return;
}
}
// Safely access health status without panicking
let is_healthy = match health_status.lock() {
Ok(status) => *status,
Err(_) => {
warn!("Health status mutex poisoned, defaulting to unhealthy");
false
}
};
let (status_line, json_body) = if is_healthy {
("HTTP/1.1 200 OK", "{\"status\":\"healthy\"}")
} else {
(
"HTTP/1.1 503 Service Unavailable",
"{\"status\":\"unhealthy\"}",
)
};
let response = format!(
"{}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
status_line,
json_body.len(),
json_body
);
let _ = timeout(
Duration::from_secs(5),
socket.write_all(response.as_bytes()),
)
.await;
};
// Overall timeout for the entire connection handling
let _ = timeout(Duration::from_secs(10), handle_request).await;
let _ = timeout(Duration::from_secs(10), async {
handle_connection(socket, health_status).await;
})
.await;
});
}
}
}
async fn handle_connection(mut socket: tokio::net::TcpStream, health_status: Arc<Mutex<bool>>) {
let mut buffer = [0; 1024];
match timeout(Duration::from_secs(5), socket.read(&mut buffer)).await {
Ok(Ok(0)) | Ok(Err(_)) | Err(_) => return,
Ok(Ok(_)) => {}
}
let is_healthy = health_status.lock().map(|s| *s).unwrap_or(false);
let (status_line, json_body) = if is_healthy {
("HTTP/1.1 200 OK", r#"{"status":"healthy"}"#)
} else {
(
"HTTP/1.1 503 Service Unavailable",
r#"{"status":"unhealthy"}"#,
)
};
let response = format!(
"{status_line}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{json_body}",
json_body.len(),
);
let _ = timeout(
Duration::from_secs(5),
socket.write_all(response.as_bytes()),
)
.await;
}
impl HealthHandle {
pub fn set_health_status(&self, is_healthy: bool) {
if let Ok(mut status) = self.last_update_success.lock() {
@ -103,12 +91,10 @@ impl HealthHandle {
}
pub fn report_update_success(&self) {
info!("Update succeeded - marking health as healthy");
self.set_health_status(true);
}
pub fn report_update_failure(&self) {
info!("Update failed - marking health as unhealthy");
self.set_health_status(false);
}
}

View file

@ -4,13 +4,18 @@ use anyhow::{anyhow, Result};
use chrono::{DateTime, Utc};
use reqwest::{Client as HttpClient, Response, StatusCode};
use serde::Deserialize;
use std::sync::LazyLock;
use std::time::Duration;
use tokio::time::sleep;
use tracing::{debug, warn};
const PAGE_SIZE: usize = 500;
const MAX_PAGES: usize = 100;
const MAX_RETRY_ATTEMPTS: u32 = 5;
const MAX_AUTH_ATTEMPTS: u32 = 2;
const INITIAL_RETRY_DELAY_SECS: u64 = 1;
const HTTP_TIMEOUT_SECS: u64 = 30;
const HTTP_CONNECT_TIMEOUT_SECS: u64 = 10;
#[derive(Debug, Clone)]
pub struct ImageRef {
@ -49,7 +54,6 @@ impl ImageRef {
}
}
_ => {
// 3 or more parts: first is registry, last is name, everything in between is namespace
let registry = registry_parts[0].to_string();
let name = registry_parts[registry_parts.len() - 1].to_string();
let namespace = Some(registry_parts[1..registry_parts.len() - 1].join("/"));
@ -68,23 +72,20 @@ impl ImageRef {
impl std::fmt::Display for ImageRef {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let base = match &self.namespace {
Some(ns) => format!("{}/{}/{}", self.registry, ns, self.name),
None => format!("{}/{}", self.registry, self.name),
};
if self.registry == "docker.io" {
match &self.namespace {
Some(ns) => write!(f, "{}/{}:{}", ns, self.name, self.tag),
None => write!(f, "{}:{}", self.name, self.tag),
}
} else {
write!(f, "{}:{}", base, self.tag)
match &self.namespace {
Some(ns) => write!(f, "{}/{}/{}:{}", self.registry, ns, self.name, self.tag),
None => write!(f, "{}/{}:{}", self.registry, self.name, self.tag),
}
}
}
}
/// Docker Hub API response for tag list
#[derive(Deserialize)]
struct DockerHubTagsResponse {
results: Vec<DockerHubTag>,
@ -96,8 +97,6 @@ struct DockerHubTag {
name: String,
}
/// Unified registry client with hybrid approach
/// Uses Docker Hub's public API for Docker Hub, standard v2 API for others
pub struct Client {
http_client: HttpClient,
config: Config,
@ -105,21 +104,23 @@ pub struct Client {
impl Client {
pub fn new(config: Config) -> Self {
let http_client = HttpClient::builder()
.timeout(Duration::from_secs(HTTP_TIMEOUT_SECS))
.connect_timeout(Duration::from_secs(HTTP_CONNECT_TIMEOUT_SECS))
.build()
.expect("failed to build HTTP client");
Self {
http_client: HttpClient::new(),
http_client,
config,
}
}
/// Parse Retry-After header to determine wait duration
/// Supports both delay-seconds (integer) and HTTP-date formats
fn parse_retry_after(&self, retry_after: &str) -> Option<Duration> {
// Try parsing as seconds first
if let Ok(seconds) = retry_after.parse::<u64>() {
return Some(Duration::from_secs(seconds));
}
// Try parsing as HTTP date (RFC 2822 or RFC 3339)
if let Ok(date) = DateTime::parse_from_rfc2822(retry_after) {
let now = Utc::now();
let wait_time = date.signed_duration_since(now);
@ -131,7 +132,6 @@ impl Client {
None
}
/// Perform HTTP request with retry logic for rate limiting (429)
async fn request_with_retry<F, Fut>(
&self,
request_fn: F,
@ -158,7 +158,6 @@ impl Client {
));
}
// Check for Retry-After header
let wait_duration = if let Some(retry_after) = response
.headers()
.get("retry-after")
@ -176,7 +175,6 @@ impl Client {
sleep(wait_duration).await;
// Exponential backoff for next attempt if no Retry-After header
delay = delay.saturating_mul(2);
} else {
return Ok(response);
@ -194,15 +192,10 @@ impl Client {
fn build_repository_path(&self, image_ref: &ImageRef) -> String {
match &image_ref.namespace {
Some(namespace) => format!("{}/{}", namespace, image_ref.name),
None => {
if image_ref.registry == "docker.io" {
// For Docker Hub official images, use 'library' namespace in public API
format!("library/{}", image_ref.name)
} else {
// For other registries, no namespace means just the image name
image_ref.name.clone()
}
None if image_ref.registry == "docker.io" => {
format!("library/{}", image_ref.name)
}
None => image_ref.name.clone(),
}
}
@ -255,7 +248,7 @@ impl Client {
async fn get_dockerhub_versions(&self, image_ref: &ImageRef) -> Result<Vec<VersionInfo>> {
let repo_path = self.build_repository_path(image_ref);
let mut results = Vec::new();
// Use Docker Hub's public REST API which doesn't require authentication for public repos
let mut page_count: usize = 0;
let mut url = format!(
"https://hub.docker.com/v2/repositories/{repo_path}/tags/?page_size={PAGE_SIZE}"
);
@ -286,6 +279,15 @@ impl Client {
results.extend(new_tags);
page_count += 1;
if page_count >= MAX_PAGES {
warn!(
"Reached maximum page count ({}) for {}, stopping pagination",
MAX_PAGES, image_ref
);
break;
}
if let Some(next) = next_page {
url = next;
} else {
@ -304,12 +306,16 @@ impl Client {
let mut last_tag: Option<String> = None;
let mut next_url: Option<String> = None;
let mut bearer_token: Option<String> = None;
let mut auth_attempts: u32 = 0;
let mut page_count: usize = 0;
// Reference: https://github.com/opencontainers/distribution-spec/blob/main/spec.md#endpoints
loop {
// Use next URL from Link header if available, otherwise build URL with last parameter
let url = if let Some(next) = next_url.take() {
format!("{}{next}", registry_config.url)
if next.starts_with("http://") || next.starts_with("https://") {
next
} else {
format!("{}{next}", registry_config.url)
}
} else {
format!(
"{}/v2/{repo_path}/tags/list?n={PAGE_SIZE}{}",
@ -324,7 +330,6 @@ impl Client {
debug!("Registry API URL: {}", url);
// Try unauthenticated request first to potentially get auth challenge
let url_clone = url.clone();
let bearer_token_clone = bearer_token.clone();
let response = self
@ -341,13 +346,21 @@ impl Client {
.await?;
let (new_tags, link_next) = if response.status() == reqwest::StatusCode::UNAUTHORIZED {
if auth_attempts >= MAX_AUTH_ATTEMPTS {
return Err(anyhow!(
"Authentication failed after {} attempts for registry {}",
auth_attempts,
image_ref.registry
));
}
if let Some(token) = &registry_config.auth_token {
// Try Docker Registry v2 auth flow if we get auth challenge
if let Some(auth_header) = response.headers().get("www-authenticate") {
bearer_token = self
.try_registry_v2_auth(auth_header.to_str().unwrap(), token)
.await?;
continue; // Retry with new token
let auth_str = auth_header.to_str().map_err(|e| {
anyhow!("Invalid WWW-Authenticate header encoding: {}", e)
})?;
bearer_token = self.try_registry_v2_auth(auth_str, token).await?;
auth_attempts += 1;
continue;
} else {
return Err(anyhow!(
"Unauthorized request but no WWW-Authenticate header found"
@ -357,7 +370,6 @@ impl Client {
return Err(anyhow!("Unauthorized request but no auth token configured"));
}
} else if response.status().is_success() {
// Check for Link header for pagination
let link_next = response
.headers()
.get("link")
@ -377,14 +389,26 @@ impl Client {
));
};
page_count += 1;
if page_count >= MAX_PAGES {
warn!(
"Reached maximum page count ({}) for {}, stopping pagination",
MAX_PAGES, image_ref
);
results.extend(new_tags);
break;
}
let maybe_last_tag = new_tags.last().map(|v| v.original.clone());
results.extend(new_tags);
// Use Link header next URL if available, otherwise fall back to last parameter pagination
if let Some(next) = link_next {
next_url = Some(next);
} else if let Some(last) = maybe_last_tag {
last_tag = Some(last);
} else if let Some(ref last) = maybe_last_tag {
if last_tag.as_ref() == Some(last) {
break;
}
last_tag = maybe_last_tag;
} else {
break;
}
@ -394,13 +418,15 @@ impl Client {
}
async fn try_registry_v2_auth(&self, auth_str: &str, token: &str) -> Result<Option<String>> {
// Parse WWW-Authenticate header to extract auth parameters
let realm = self.extract_auth_param(auth_str, "realm")?;
let service = self.extract_auth_param(auth_str, "service")?;
let scope = self.extract_auth_param(auth_str, "scope")?;
let realm = extract_auth_param(auth_str, "realm")?;
let service = extract_auth_param(auth_str, "service")?;
let scope = extract_auth_param(auth_str, "scope")?;
// Request token from auth endpoint
let auth_url = format!("{realm}?service={service}&scope={scope}");
let auth_url = format!(
"{realm}?service={}&scope={}",
urlencoding::encode(&service),
urlencoding::encode(&scope),
);
debug!("Getting registry token from: {}", auth_url);
let auth_url_clone = auth_url.clone();
@ -432,29 +458,16 @@ impl Client {
return Ok(Some(registry_token.to_string()));
}
Err(anyhow!("Auth flow didn't work, caller can try fallback"))
}
fn extract_auth_param(&self, auth_str: &str, param: &str) -> Result<String> {
let pattern = format!(r#"{param}="([^"]+)""#);
let re = regex::Regex::new(&pattern)?;
re.captures(auth_str)
.and_then(|caps| caps.get(1))
.map(|m| m.as_str().to_string())
.ok_or_else(|| anyhow!("Missing {} in auth challenge", param))
Err(anyhow!("Auth response missing 'token' field"))
}
fn parse_link_header(&self, link_header: &str) -> Option<String> {
// Parse RFC 5988 Link header to find "next" relation
// Format: <https://example.com/page2>; rel="next", <https://example.com/last>; rel="last"
for link in link_header.split(',') {
let parts: Vec<&str> = link.trim().split(';').collect();
if parts.len() >= 2 {
let url = parts[0].trim();
if url.starts_with('<') && url.ends_with('>') {
let url = &url[1..url.len() - 1]; // Remove < and >
let url = &url[1..url.len() - 1];
for param in &parts[1..] {
let param = param.trim();
if param == "rel=\"next\"" || param == "rel=next" {
@ -468,6 +481,22 @@ impl Client {
}
}
static AUTH_PARAM_REGEX: LazyLock<regex::Regex> =
LazyLock::new(|| regex::Regex::new(r#"(\w+)="([^"]+)""#).unwrap());
fn extract_auth_param(auth_str: &str, param: &str) -> Result<String> {
for caps in AUTH_PARAM_REGEX.captures_iter(auth_str) {
if caps.get(1).map(|m| m.as_str()) == Some(param) {
return Ok(caps[2].to_string());
}
}
Err(anyhow!(
"Missing '{}' in auth challenge: {}",
param,
auth_str
))
}
#[cfg(test)]
mod tests {
use super::*;

View file

@ -5,8 +5,7 @@ use cron::Schedule;
use std::str::FromStr;
use std::time::Duration;
use tokio::time::{sleep, Instant};
use tracing::error;
use tracing::info;
use tracing::{error, info};
pub struct Scheduler {
config: Config,
@ -38,24 +37,26 @@ impl Scheduler {
loop {
if let Err(err) = self.run_update().await.context("Failed to run update") {
error!("{:?}", err)
error!("{:#}", err);
}
if let Some(next_run) = self.schedule.upcoming(chrono::Utc).take(1).next() {
let now = chrono::Utc::now();
let duration_until_next = next_run.signed_duration_since(now);
let sleep_duration =
if let Some(next_run) = self.schedule.upcoming(chrono::Utc).take(1).next() {
let now = chrono::Utc::now();
let duration_until_next = next_run.signed_duration_since(now);
let millis = duration_until_next.num_milliseconds().max(0) as u64;
let duration = Duration::from_millis(millis).max(Duration::from_secs(1));
if duration_until_next.num_seconds() > 0 {
info!(
"Next update scheduled for: {}",
next_run.format("%Y-%m-%d %H:%M:%S UTC")
);
duration
} else {
Duration::from_secs(60)
};
let sleep_duration =
Duration::from_secs(duration_until_next.num_seconds() as u64);
sleep(sleep_duration).await;
}
}
sleep(sleep_duration).await;
}
}
@ -66,48 +67,55 @@ impl Scheduler {
async fn run_update(&self) -> Result<()> {
let start_time = Instant::now();
info!("Starting Docker Compose update cycle");
info!("Starting update cycle");
match self.updater.update_all_compose_files().await {
Ok(updated_files) => {
Ok(report) => {
let duration = start_time.elapsed();
if updated_files.is_empty() {
if report.files_found == 0 {
if let Some(health) = &self.health_handle {
health.report_update_failure();
}
return Err(anyhow!(
"No compose files found under configured compose_paths: {:?}",
self.config.compose_paths
));
}
if report.updated_files.is_empty() {
info!(
"Update cycle completed in {:?} - no files updated",
duration
"Update cycle completed in {:?} - scanned {} files, none updated",
duration, report.files_found
);
} else {
let verb = if self.config.dry_run {
"would update"
} else {
"updated"
};
info!(
"Update cycle completed in {:?} - {} {} files:",
"Update cycle completed in {:?} - scanned {} files, {} {} files:",
duration,
if self.config.dry_run {
"would update"
} else {
"updated"
},
updated_files.len()
report.files_found,
verb,
report.updated_files.len()
);
for file in &updated_files {
for file in &report.updated_files {
info!(" - {}", file);
}
}
// Report success to health monitor
if let Some(ref health) = self.health_handle {
if let Some(health) = &self.health_handle {
health.report_update_success();
}
Ok(())
}
Err(e) => {
let error = e.context("Failed to update Docker Compose files");
// Report failure to health monitor
if let Some(ref health) = self.health_handle {
if let Some(health) = &self.health_handle {
health.report_update_failure();
}
Err(error)
Err(e.context("Failed to update Docker Compose files"))
}
}
}
@ -120,18 +128,20 @@ mod tests {
use std::collections::HashMap;
use std::path::PathBuf;
fn test_config(schedule: &str) -> Config {
Config {
compose_paths: vec![PathBuf::from("./test")],
schedule: schedule.to_string(),
registries: HashMap::new(),
update_strategy: UpdateStrategy::LatestPatchOfPreviousMinor,
ignore_images: vec![],
dry_run: true,
}
}
#[test]
fn test_scheduler_creation() {
let config = Config {
compose_paths: vec![PathBuf::from("./test")],
schedule: "0 0 2 * * *".to_string(),
registries: HashMap::new(),
update_strategy: UpdateStrategy::LatestPatchOfPreviousMinor,
ignore_images: vec![],
dry_run: true,
};
let scheduler = Scheduler::new(config, None).unwrap();
let scheduler = Scheduler::new(test_config("0 0 2 * * *"), None).unwrap();
assert!(scheduler
.schedule
.upcoming(chrono::Utc)
@ -141,19 +151,8 @@ mod tests {
}
#[test]
fn test_cron_parsing() {
let config = Config {
compose_paths: vec![PathBuf::from("./test")],
schedule: "0 30 1 * * *".to_string(), // 1:30 AM daily
registries: HashMap::new(),
update_strategy: UpdateStrategy::LatestPatchOfPreviousMinor,
ignore_images: vec![],
dry_run: true,
};
let scheduler = Scheduler::new(config, None).unwrap();
// Just verify the scheduler can be created
fn test_scheduler_with_different_cron() {
let scheduler = Scheduler::new(test_config("0 30 1 * * *"), None).unwrap();
assert!(scheduler
.schedule
.upcoming(chrono::Utc)
@ -161,4 +160,17 @@ mod tests {
.next()
.is_some());
}
#[tokio::test]
async fn test_run_once_errors_when_no_compose_files_found() {
let mut config = test_config("0 0 2 * * *");
config.compose_paths = vec![];
let scheduler = Scheduler::new(config, None).unwrap();
let result = scheduler.run_once().await;
assert!(
result.is_err(),
"an update cycle that finds no compose files should report an error"
);
}
}

View file

@ -1,6 +1,6 @@
use crate::config::UpdateStrategy;
use crate::version::VersionInfo;
use tracing::{info, warn};
use tracing::{debug, warn};
pub fn create_selector(strategy: &UpdateStrategy) -> Box<dyn VersionSelector> {
match strategy {
@ -27,17 +27,15 @@ impl VersionSelector for LatestVersionSelector {
current_prefix: Option<String>,
current_suffix: Option<String>,
) -> Option<VersionInfo> {
info!("Using update strategy: LatestVersionSelector");
let versions =
get_filtered_and_soreted_matching_versions(available, current_prefix, current_suffix);
get_filtered_and_sorted_matching_versions(available, current_prefix, current_suffix);
let latest = versions.first().cloned();
if let Some(ref selected) = latest {
info!("Selected {} as latest version", selected.version);
debug!("Latest version selected: {}", selected.version);
} else {
warn!("No versions available for selection");
warn!("No matching versions available");
}
latest
@ -53,25 +51,19 @@ impl VersionSelector for SmartPreviousMinorSelector {
current_prefix: Option<String>,
current_suffix: Option<String>,
) -> Option<VersionInfo> {
info!("Using update strategy: SmartPreviousMinorSelector");
let versions =
get_filtered_and_soreted_matching_versions(available, current_prefix, current_suffix);
get_filtered_and_sorted_matching_versions(available, current_prefix, current_suffix);
let latest = &versions.first()?.version;
info!("Latest version available: {}", latest);
debug!("Latest version available: {}", latest);
let (target_major, max_minor) = if latest.minor == 0 {
// If current minor is 0, look for the latest minor of the previous major
if latest.major == 0 {
info!("Cannot go to previous version of 0.0.x, skipping update");
debug!("Cannot go to previous version of 0.0.x");
return None;
}
info!("The latest version has a minor version of 0, looking for the latest minor of the previous major");
(latest.major - 1, None)
} else {
info!("The latest version has a minor version of {}, looking for the latest patch of the previous minor", latest.minor);
(latest.major, Some(latest.minor - 1))
};
@ -83,46 +75,35 @@ impl VersionSelector for SmartPreviousMinorSelector {
.cloned();
if let Some(ref selected) = selected {
info!("Selected {} as latest version", selected.version);
debug!("Selected version: {}", selected.version);
} else {
warn!("No versions available for selection");
warn!("No matching versions available for previous minor strategy");
}
selected
}
}
fn get_filtered_and_soreted_matching_versions(
fn get_filtered_and_sorted_matching_versions(
available: &[VersionInfo],
current_prefix: Option<String>,
current_suffix: Option<String>,
) -> Vec<VersionInfo> {
let mut sorted_versions = available.to_vec();
sorted_versions.sort_by(|a, b| b.version.cmp(&a.version));
info!(
"Available versions for selection: {}",
sorted_versions
.iter()
.map(|v| v.to_string())
.collect::<Vec<_>>()
.join(", ")
);
let filtered_versions: Vec<VersionInfo> = sorted_versions
.into_iter()
let mut versions: Vec<VersionInfo> = available
.iter()
.filter(|v| v.prefix == current_prefix && v.suffix == current_suffix)
.cloned()
.collect();
versions.sort_by(|a, b| b.version.cmp(&a.version));
info!(
"Filtered versions with matching prefix and suffix: {}",
filtered_versions
.iter()
.map(|v| v.to_string())
.collect::<Vec<_>>()
.join(", ")
debug!(
"Filtered versions (prefix={:?}, suffix={:?}): {}",
current_prefix,
current_suffix,
versions.len()
);
filtered_versions
versions
}
#[cfg(test)]

View file

@ -1,6 +1,11 @@
use regex::Regex;
use semver::Version;
use std::fmt;
use std::sync::LazyLock;
static VERSION_REGEX: LazyLock<Regex> = LazyLock::new(|| {
Regex::new(r"^(?P<prefix>.*?)(?P<version>(?:\d+\.)+\d+)(?P<suffix>.*?)$").unwrap()
});
#[derive(Debug, Clone, PartialEq)]
pub struct VersionInfo {
@ -24,9 +29,7 @@ impl VersionInfo {
/// assert_eq!(version_info.suffix, Some("-alpine-slim".to_string()));
/// ```
pub fn from_tag(tag: &str) -> Option<Self> {
let re = Regex::new(r"^(?P<prefix>.*?)(?P<version>(?:\d+\.)+\d+)(?P<suffix>.*?)$").unwrap();
if let Some(captures) = re.captures(tag) {
if let Some(captures) = VERSION_REGEX.captures(tag) {
let prefix_part = captures.name("prefix").map_or("", |m| m.as_str());
let version_part = captures.name("version").map_or("", |m| m.as_str());
let suffix_part = captures.name("suffix").map_or("", |m| m.as_str());