179 lines
7.4 KiB
Python
179 lines
7.4 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import docker
|
|
import yaml
|
|
import os
|
|
import json
|
|
import time
|
|
import argparse
|
|
from datetime import datetime
|
|
import logging
|
|
from typing import Dict, List, Optional
|
|
|
|
logging.basicConfig(level=logging.INFO,
|
|
format='%(asctime)s - %(levelname)s - %(message)s')
|
|
|
|
class ContainerManager:
|
|
def __init__(self, manifest_dir: str = "/app/manifest"):
|
|
self.client = docker.from_env()
|
|
self.manifest_dir = manifest_dir
|
|
self.manifest_file = os.path.join(manifest_dir, "container_versions.yaml")
|
|
os.makedirs(manifest_dir, exist_ok=True)
|
|
|
|
def get_current_versions(self) -> Dict[str, Dict]:
|
|
"""Get current versions of all running containers."""
|
|
versions = {}
|
|
for container in self.client.containers.list():
|
|
try:
|
|
image = container.image
|
|
image_tags = image.tags
|
|
image_id = image.id
|
|
versions[container.name] = {
|
|
"image": image_tags[0] if image_tags else image_id,
|
|
"id": image_id,
|
|
"created": container.attrs['Created'],
|
|
"status": container.status
|
|
}
|
|
except Exception as e:
|
|
logging.error(f"Error processing container {container.name}: {e}")
|
|
return versions
|
|
|
|
def load_manifest(self) -> Dict:
|
|
"""Load the existing version manifest."""
|
|
if os.path.exists(self.manifest_file):
|
|
with open(self.manifest_file, 'r') as f:
|
|
return yaml.safe_load(f) or {}
|
|
return {}
|
|
|
|
def save_manifest(self, versions: Dict):
|
|
"""Save the current versions to the manifest file."""
|
|
versions['last_updated'] = datetime.now().isoformat()
|
|
with open(self.manifest_file, 'w') as f:
|
|
yaml.dump(versions, f, default_flow_style=False)
|
|
logging.info(f"Manifest updated: {self.manifest_file}")
|
|
|
|
def check_for_updates(self) -> List[str]:
|
|
"""Check which containers have updates available."""
|
|
updates_available = []
|
|
current_versions = self.get_current_versions()
|
|
|
|
for name, info in current_versions.items():
|
|
try:
|
|
# Skip if image is referenced by ID only (sha256:...)
|
|
if info['image'].startswith('sha256:') or ':' not in info['image']:
|
|
logging.info(f"Skipping image referenced by ID for container {name}: {info['image'][:20]}...")
|
|
continue
|
|
|
|
repo, tag = info['image'].split(':', 1)
|
|
|
|
# Skip locally built images (they typically start with the directory name)
|
|
if repo.startswith(('mydocker_', 'docker-')):
|
|
logging.info(f"Skipping locally built image: {name}")
|
|
continue
|
|
|
|
# Skip if the repository looks like it contains a hash or invalid characters
|
|
if '/' not in repo or any(char in repo for char in ['@', '#']):
|
|
logging.info(f"Skipping non-standard image format for {name}: {info['image']}")
|
|
continue
|
|
|
|
self.client.images.pull(repo, tag)
|
|
latest_image = self.client.images.get(f"{repo}:{tag}")
|
|
|
|
if latest_image.id != info['id']:
|
|
updates_available.append(name)
|
|
logging.info(f"Update available for {name}")
|
|
except Exception as e:
|
|
logging.error(f"Error checking updates for {name}: {e}")
|
|
|
|
return updates_available
|
|
|
|
def update_container(self, container_name: str) -> bool:
|
|
"""Update a specific container."""
|
|
try:
|
|
container = self.client.containers.get(container_name)
|
|
image_name = container.image.tags[0]
|
|
|
|
# Pull the latest image
|
|
self.client.images.pull(image_name)
|
|
|
|
# Stop and remove the old container
|
|
container.stop()
|
|
container.remove()
|
|
|
|
# The container will be automatically recreated by docker-compose
|
|
logging.info(f"Container {container_name} updated successfully")
|
|
return True
|
|
except Exception as e:
|
|
logging.error(f"Error updating container {container_name}: {e}")
|
|
return False
|
|
|
|
def update_all(self) -> Dict[str, bool]:
|
|
"""Update all containers that have updates available."""
|
|
results = {}
|
|
updates = self.check_for_updates()
|
|
|
|
for container in updates:
|
|
results[container] = self.update_container(container)
|
|
|
|
# Update the manifest after updates
|
|
self.save_manifest(self.get_current_versions())
|
|
return results
|
|
|
|
def run_daemon(self, check_interval: int = 3600):
|
|
"""Run in daemon mode, checking for updates periodically."""
|
|
while True:
|
|
try:
|
|
logging.info("Checking for container updates...")
|
|
current_versions = self.get_current_versions()
|
|
self.save_manifest(current_versions)
|
|
|
|
updates = self.check_for_updates()
|
|
if updates:
|
|
logging.info(f"Updates available for: {', '.join(updates)}")
|
|
results = self.update_all()
|
|
logging.info("Update results:")
|
|
for container, success in results.items():
|
|
logging.info(f"{container}: {'Success' if success else 'Failed'}")
|
|
else:
|
|
logging.info("All containers are up to date")
|
|
|
|
time.sleep(check_interval)
|
|
except Exception as e:
|
|
logging.error(f"Error in daemon mode: {e}")
|
|
time.sleep(60) # Wait a minute before retrying on error
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description='Docker container version manager')
|
|
parser.add_argument('--daemon', action='store_true', help='Run in daemon mode')
|
|
parser.add_argument('--check-interval', type=int, default=3600,
|
|
help='Interval between update checks in daemon mode (seconds)')
|
|
parser.add_argument('--check', action='store_true', help='Check for updates')
|
|
parser.add_argument('--update', action='store_true', help='Update all containers')
|
|
parser.add_argument('--update-container', type=str, help='Update specific container')
|
|
|
|
args = parser.parse_args()
|
|
manager = ContainerManager()
|
|
|
|
if args.daemon:
|
|
logging.info(f"Starting daemon mode with {args.check_interval}s check interval")
|
|
manager.run_daemon(args.check_interval)
|
|
elif args.check:
|
|
updates = manager.check_for_updates()
|
|
if updates:
|
|
print(f"Updates available for: {', '.join(updates)}")
|
|
else:
|
|
print("All containers are up to date")
|
|
elif args.update:
|
|
results = manager.update_all()
|
|
for container, success in results.items():
|
|
print(f"{container}: {'Success' if success else 'Failed'}")
|
|
elif args.update_container:
|
|
success = manager.update_container(args.update_container)
|
|
print(f"Update {'successful' if success else 'failed'} for {args.update_container}")
|
|
else:
|
|
# Default behavior: show current versions
|
|
versions = manager.get_current_versions()
|
|
print(yaml.dump(versions, default_flow_style=False))
|
|
|
|
if __name__ == "__main__":
|
|
main() |