Add Proxmox Backup Server integration

- Docker volume path resolution for named volumes
- Archive name generation from filesystem paths
- PBS backup creation with authentication support
- Temporary file handling for backup paths
- Complete backup workflow implementation
This commit is contained in:
Radosław Gierwiało
2024-08-18 15:45:00 +02:00
parent c9eecb1117
commit 971faa7323

View File

@@ -149,6 +149,138 @@ class Docker2PBS:
raise RuntimeError(f"Failed to start service: {result.stderr}") raise RuntimeError(f"Failed to start service: {result.stderr}")
print(f"Service {self.service_name} started successfully") print(f"Service {self.service_name} started successfully")
def get_docker_volume_path(self, volume_name: str) -> Optional[str]:
"""Get host path for Docker volume"""
try:
cmd = ['docker', 'volume', 'inspect', volume_name]
if self.dry_run:
print(f"[DRY-RUN] Would run: {' '.join(cmd)}")
return f"/var/lib/docker/volumes/{volume_name}/_data"
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode == 0:
volume_info = json.loads(result.stdout)
if volume_info and len(volume_info) > 0:
return volume_info[0].get('Mountpoint')
except Exception as e:
print(f"Warning: Could not get path for volume {volume_name}: {e}")
return None
def prepare_volume_backups(self) -> List[Dict[str, str]]:
"""Prepare volume backup information"""
volume_backups = []
for volume in self.volumes:
if volume.get('type') == 'bind' and not volume.get('external'):
source_path = volume.get('source')
if self.dry_run or (source_path and os.path.exists(source_path)):
# Create safe archive name from path
archive_name = self.path_to_archive_name(source_path, 'bind')
volume_backups.append({
'archive_name': archive_name,
'source_path': source_path,
'volume_type': 'bind_mount'
})
elif volume.get('type') == 'volume':
# For Docker volumes, we need to find their host path
volume_name = volume.get('source')
if volume_name:
volume_path = self.get_docker_volume_path(volume_name)
if volume_path:
archive_name = f"volume_{volume_name}.pxar"
volume_backups.append({
'archive_name': archive_name,
'source_path': volume_path,
'volume_type': 'named_volume'
})
return volume_backups
def path_to_archive_name(self, path: str, prefix: str) -> str:
"""Convert file system path to safe archive name"""
# Normalize path and make it safe for archive names
normalized = path.replace('/', '_').replace('.', 'dot').replace('..', 'dotdot')
# Remove leading underscores and clean up
normalized = normalized.strip('_')
if not normalized:
normalized = 'root'
return f"{prefix}_{normalized}.pxar"
def create_backup_paths_file(self) -> str:
"""Create temporary file with paths to backup"""
volume_backups = self.prepare_volume_backups()
if not volume_backups:
raise ValueError("No backup paths found for the service")
backup_paths = [vb['source_path'] for vb in volume_backups]
# Create temporary file with paths
paths_file = f"/tmp/backup_paths_{self.service_name}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt"
if self.dry_run:
print(f"[DRY-RUN] Would create paths file: {paths_file}")
print(f"[DRY-RUN] Paths to backup: {backup_paths}")
return paths_file
with open(paths_file, 'w') as f:
for path in backup_paths:
f.write(f"{path}\n")
return paths_file
def create_pbs_backup(self) -> None:
"""Create backup on Proxmox Backup Server"""
timestamp = datetime.now().strftime('%Y-%m-%d_%H-%M-%S')
backup_id = f"{self.service_name}_{timestamp}"
paths_file = self.create_backup_paths_file()
try:
print(f"Creating backup {backup_id} on Proxmox Backup Server")
# Build proxmox-backup-client command
cmd = [
'proxmox-backup-client',
'backup',
f"--repository={self.pbs_config['repository']}",
f"--backup-id={backup_id}",
f"--backup-type=host"
]
# Add authentication if provided
if 'username' in self.pbs_config:
cmd.append(f"--userid={self.pbs_config['username']}")
if 'password' in self.pbs_config:
cmd.append(f"--password={self.pbs_config['password']}")
if 'fingerprint' in self.pbs_config:
cmd.append(f"--fingerprint={self.pbs_config['fingerprint']}")
# Add paths from file
cmd.append(f"root.pxar:@{paths_file}")
if self.dry_run:
print(f"[DRY-RUN] Would run: {' '.join(cmd)}")
print(f"[DRY-RUN] Backup {backup_id} would be created on PBS")
return
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
raise RuntimeError(f"Backup failed: {result.stderr}")
print(f"Backup {backup_id} created successfully")
finally:
# Clean up temporary file
if not self.dry_run and os.path.exists(paths_file):
os.remove(paths_file)
elif self.dry_run:
print(f"[DRY-RUN] Would remove temporary file: {paths_file}")
def main(): def main():