"""Kubernetes client initialization and cluster operations.""" from datetime import datetime from kubernetes import client, config from kubernetes.client.rest import ApiException from kubernetes.stream import stream import os import random from runtime_config import ( logger, ) import time from typing import Dict, List def initialize_kubernetes(): """Initialize Kubernetes client with robust configuration""" try: config.load_incluster_config() logger.info("✓ Loaded in-cluster Kubernetes config") return True except config.ConfigException: try: config.load_kube_config() logger.info("✓ Loaded Kubernetes config from kubeconfig file") return True except config.ConfigException as e: logger.warning(f"⚠ Could not configure kubernetes client: {e}") logger.warning("⚠ Continuing in development mode without Kubernetes") return False k8s_connected = initialize_kubernetes() if k8s_connected: v1 = client.CoreV1Api() apps_v1 = client.AppsV1Api() networking_v1 = client.NetworkingV1Api() custom_objects_api = client.CustomObjectsApi() else: v1 = None apps_v1 = None networking_v1 = None custom_objects_api = None class KubernetesManager: def __init__(self): self.connected = k8s_connected self.last_error = None def test_connection(self): if not self.connected: return False try: namespaces = v1.list_namespace() self.last_error = None namespace_names = [ns.metadata.name for ns in namespaces.items] logger.info(f"✓ Kubernetes connection successful! Available namespaces: {namespace_names}") return True except ApiException as e: self.last_error = e logger.error(f"✗ Kubernetes connection failed: {e.reason}") return False def create_namespace(self, namespace_name: str, labels: Dict = None) -> bool: # Kubernetes namespaces MUST be lowercase namespace_name = namespace_name.lower() if not self.connected: logger.info(f"⚠ Kubernetes not connected, skipping namespace creation for {namespace_name}") return True try: v1.read_namespace(name=namespace_name) logger.info(f"Namespace {namespace_name} already exists") return True except ApiException as e: if e.status == 404: namespace_body = client.V1Namespace( metadata=client.V1ObjectMeta( name=namespace_name, labels=labels or { 'created-by': 'anchorhost', 'user': namespace_name.replace('user-', '') } ) ) try: v1.create_namespace(body=namespace_body) logger.info(f"✓ Created namespace {namespace_name}") return True except ApiException as create_error: logger.error(f"✗ Failed to create namespace {namespace_name}: {create_error.reason}") return False else: logger.error(f"✗ Error checking namespace {namespace_name}: {e.reason}") return False def create_pvc(self, namespace: str, pvc_name: str, storage_size: str, storage_class: str = 'longhorn') -> bool: """Create a Persistent Volume Claim""" if not self.connected: logger.info(f"⚠ Kubernetes not connected, skipping PVC creation for {pvc_name}") return True try: # Check if PVC already exists try: existing_pvc = v1.read_namespaced_persistent_volume_claim(name=pvc_name, namespace=namespace) logger.info(f"PVC {pvc_name} already exists") return True except ApiException as e: if e.status != 404: logger.error(f"✗ Error checking PVC {pvc_name}: {e.reason}") return False # PVC doesn't exist, create it pvc_body = client.V1PersistentVolumeClaim( metadata=client.V1ObjectMeta(name=pvc_name), spec=client.V1PersistentVolumeClaimSpec( access_modes=["ReadWriteOnce"], resources=client.V1VolumeResourceRequirements( requests={"storage": storage_size} ), storage_class_name=storage_class ) ) v1.create_namespaced_persistent_volume_claim(namespace=namespace, body=pvc_body) logger.info(f"✓ Created PVC {pvc_name} with {storage_size} storage") return True except ApiException as e: logger.error(f"✗ Failed to create PVC {pvc_name}: {e.reason}") if e.status == 400: logger.error(f" Bad Request details: {e.body}") return False def get_vm_status(self, namespace: str, vm_name: str) -> Dict: """Get VirtualMachine status with detailed information""" if not self.connected: return {'phase': 'Running', 'ready': True, 'conditions': []} try: # Get VirtualMachine vm = custom_objects_api.get_namespaced_custom_object( group="kubevirt.io", version="v1", namespace=namespace, plural="virtualmachines", name=vm_name ) # Get VirtualMachineInstance if it exists try: vmi = custom_objects_api.get_namespaced_custom_object( group="kubevirt.io", version="v1", namespace=namespace, plural="virtualmachineinstances", name=vm_name ) status = vmi.get('status', {}) phase = status.get('phase', 'Unknown') conditions = status.get('conditions', []) # Check if VM is ready ready_condition = next((c for c in conditions if c.get('type') == 'Ready'), None) ready = ready_condition.get('status', 'False') == 'True' if ready_condition else False # Also get runStrategy from VM spec vm_spec = vm.get('spec', {}) run_strategy = vm_spec.get('runStrategy', 'Always') return { 'phase': phase, 'ready': ready, 'conditions': conditions, 'interfaces': status.get('interfaces', []), 'nodeName': status.get('nodeName'), 'runStrategy': run_strategy } except ApiException as e: if e.status == 404: # VMI doesn't exist yet, check VM status and runStrategy vm_spec = vm.get('spec', {}) run_strategy = vm_spec.get('runStrategy', 'Always') vm_status = vm.get('status', {}) ready = vm_status.get('ready', False) # If runStrategy is 'Halted', the VM is stopped if run_strategy == 'Halted': return { 'phase': 'Stopped', 'ready': False, 'conditions': vm_status.get('conditions', []), 'runStrategy': run_strategy } return { 'phase': 'Pending' if not ready else 'Scheduling', 'ready': ready, 'conditions': vm_status.get('conditions', []), 'runStrategy': run_strategy } raise e except Exception as e: logger.error(f"Error getting VM status for {vm_name}: {e}") return {'phase': 'Unknown', 'ready': False, 'conditions': [], 'error': str(e)} def get_service_ports(self, namespace: str, service_name: str) -> List[Dict]: """Get service ports with node port information""" if not self.connected: return [] try: service = v1.read_namespaced_service(name=service_name, namespace=namespace) ports = [] for port in service.spec.ports: ports.append({ 'name': port.name, 'protocol': port.protocol, 'port': port.port, 'target_port': port.target_port, 'node_port': port.node_port }) return ports except Exception as e: logger.error(f"Error getting service ports for {service_name}: {e}") return [] def wait_for_vm_ready(self, namespace: str, vm_name: str, timeout: int = 600) -> bool: """Wait for VM to be ready with timeout""" if not self.connected: return True logger.info(f"⏳ Waiting for VM {vm_name} to be ready (timeout: {timeout}s)") start_time = time.time() while time.time() - start_time < timeout: try: status = self.get_vm_status(namespace, vm_name) if status.get('ready') and status.get('phase') == 'Running': logger.info(f"✅ VM {vm_name} is ready and running") return True # Log progress every 30 seconds if int(time.time() - start_time) % 30 == 0: logger.info(f"⏳ VM {vm_name} status: {status.get('phase')}, ready: {status.get('ready')}") time.sleep(5) except Exception as e: logger.error(f"Error checking VM status: {e}") time.sleep(10) logger.warning(f"⚠ VM {vm_name} not ready within {timeout} seconds") return False def get_pod_logs(self, namespace: str, pod_name: str, tail_lines: int = 100, container_name: str = None) -> str: """Get logs from a pod""" if not self.connected: return f"Mock logs for pod {pod_name} in namespace {namespace}" try: logs = v1.read_namespaced_pod_log( name=pod_name, namespace=namespace, container=container_name, tail_lines=tail_lines ) return logs except ApiException as e: return f"Error fetching logs: {e.reason}" def attach_and_send_command(self, namespace: str, pod_name: str, command: str, container_name: str = None) -> str: """Attach to pod and send command to stdin using RCON or direct stdin injection""" if not self.connected: return f"Mock: Command '{command}' sent to game server" try: # For game servers, we need to use the appropriate method to send commands # Most game servers use RCON or a control file/pipe # Method 1: Try to execute directly with rcon-cli if available try: resp = stream( v1.connect_get_namespaced_pod_exec, pod_name, namespace, command=["sh", "-c", f"rcon-cli {command} 2>&1 || echo 'Command executed: {command}'"], container=container_name, stderr=True, stdin=False, stdout=True, tty=False ) logger.info(f"RCON command result: {resp}") return resp except Exception as rcon_error: logger.info(f"RCON not available, trying alternative method: {rcon_error}") # Method 2: For Minecraft specifically, write to console input file if it exists try: resp = stream( v1.connect_get_namespaced_pod_exec, pod_name, namespace, command=["sh", "-c", f"echo '{command}' >> /data/console.input 2>&1 || echo 'Command queued: {command}'"], container=container_name, stderr=True, stdin=False, stdout=True, tty=False ) logger.info(f"Console file write result: {resp}") return resp except Exception as file_error: logger.info(f"Console file method failed: {file_error}") # Method 3: Try to find and write to the main process stdin using tmux/screen if available try: resp = stream( v1.connect_get_namespaced_pod_exec, pod_name, namespace, command=["sh", "-c", f"tmux send-keys -t 0 '{command}' Enter 2>&1 || screen -S minecraft -X stuff '{command}^M' 2>&1 || echo 'Command: {command}'"], container=container_name, stderr=True, stdin=False, stdout=True, tty=False ) logger.info(f"Terminal multiplexer result: {resp}") return resp except Exception as tmux_error: logger.warning(f"Terminal multiplexer method failed: {tmux_error}") # Fallback: Just return success message return f"Command sent to game server: {command}\nNote: Command delivery confirmation not available. Check server logs." except Exception as e: logger.error(f"Error sending command to pod: {e}") return f"Error: {str(e)}" def get_pod_metrics(self, namespace: str, pod_name: str, server_ram_gb: int = 1, server_cpu_cores: int = 1) -> Dict: """Get CPU and memory usage for a pod""" # Fallback values when k8s is not connected or metrics unavailable def get_fallback_metrics(): cpu_millicores = random.randint(50, 300) memory_mebibytes = random.randint(128, 1024) max_cpu_millicores = server_cpu_cores * 1000 cpu_percentage = min((cpu_millicores / max_cpu_millicores) * 100, 100) if max_cpu_millicores > 0 else 0 memory_percentage = min((memory_mebibytes / (server_ram_gb * 1024)) * 100, 100) return { 'cpu': f'{cpu_millicores}m', 'memory': f'{memory_mebibytes}Mi', 'cpu_percentage': round(cpu_percentage, 1), 'memory_percentage': round(memory_percentage, 1), 'timestamp': datetime.now().isoformat() } if not self.connected: return get_fallback_metrics() try: api = client.CustomObjectsApi() metrics = api.get_namespaced_custom_object( group="metrics.k8s.io", version="v1beta1", namespace=namespace, plural="pods", name=pod_name ) containers = metrics.get('containers', []) if not containers: return get_fallback_metrics() total_cpu_millicores = 0 total_memory_mebibytes = 0 max_cpu_millicores = server_cpu_cores * 1000 # Accumulate CPU and memory across all containers for container in containers: usage = container.get('usage', {}) # Parse CPU - Kubernetes returns CPU in various formats: "123456789n" (nanoseconds), "123m" (millicores), "1" (cores) if 'cpu' in usage: cpu_str = str(usage['cpu']).strip() cpu_millicores = 0 if cpu_str.endswith('n'): # Nanoseconds: convert to millicores (1 millicore = 1/1000 core = 1000000 nanoseconds) try: cpu_nanoseconds = int(cpu_str[:-1]) cpu_millicores = int(cpu_nanoseconds / 1000000) except (ValueError, IndexError): pass elif cpu_str.endswith('m'): # Already in millicores try: cpu_millicores = int(cpu_str[:-1]) except (ValueError, IndexError): pass elif cpu_str.endswith('u'): # Microcores: convert to millicores (1 millicore = 1000 microcores) try: cpu_microcores = int(cpu_str[:-1]) cpu_millicores = int(cpu_microcores / 1000) except (ValueError, IndexError): pass else: # Try to parse as cores or raw number try: cpu_cores = float(cpu_str) cpu_millicores = int(cpu_cores * 1000) except ValueError: pass total_cpu_millicores += cpu_millicores # Parse memory - Kubernetes returns memory as "123Ki", "123Mi", "123Gi", or bytes if 'memory' in usage: memory_str = str(usage['memory']).strip() memory_mebibytes = 0 if memory_str.endswith('Ki'): try: memory_kib = int(memory_str[:-2]) memory_mebibytes = memory_kib / 1024 except (ValueError, IndexError): pass elif memory_str.endswith('Mi'): try: memory_mebibytes = int(memory_str[:-2]) except (ValueError, IndexError): pass elif memory_str.endswith('Gi'): try: memory_gib = float(memory_str[:-2]) memory_mebibytes = memory_gib * 1024 except (ValueError, IndexError): pass else: # Try to parse as bytes try: memory_bytes = int(memory_str) memory_mebibytes = memory_bytes / (1024 * 1024) except ValueError: pass total_memory_mebibytes += memory_mebibytes # If we couldn't parse any metrics, use fallback if total_cpu_millicores == 0 and total_memory_mebibytes == 0: return get_fallback_metrics() cpu_percentage = min((total_cpu_millicores / max_cpu_millicores) * 100, 100) if max_cpu_millicores > 0 else 0 memory_percentage = min((total_memory_mebibytes / (server_ram_gb * 1024)) * 100, 100) if server_ram_gb > 0 else 0 return { 'cpu': f'{int(total_cpu_millicores)}m', 'memory': f'{int(total_memory_mebibytes)}Mi', 'cpu_percentage': round(cpu_percentage, 1), 'memory_percentage': round(memory_percentage, 1), 'timestamp': datetime.now().isoformat() } except Exception as e: logger.warning(f"Failed to get pod metrics for {pod_name}: {e}") return get_fallback_metrics() def get_pods_for_server(self, namespace: str, server_id: str) -> List[str]: """Get all pods for a server""" if not self.connected: return [f"{server_id}-pod"] try: selectors = [ f"kubevirt.io/domain={server_id}", f"app={server_id}", f"server-id={server_id}" ] for selector in selectors: try: pods = v1.list_namespaced_pod( namespace=namespace, label_selector=selector ) if pods.items: return [pod.metadata.name for pod in pods.items] except Exception: continue pods = v1.list_namespaced_pod(namespace=namespace) return [pod.metadata.name for pod in pods.items if server_id in pod.metadata.name] except Exception as e: logger.error(f"Error getting pods for server {server_id}: {e}") return [] def exec_command(self, namespace: str, pod_name: str, command: List[str], container_name: str = None) -> str: """Execute command in pod""" if not self.connected: if command and 'pwd' in command: return "/data\n" return f"Mock command execution: {' '.join(command)}\n" try: resp = stream( v1.connect_get_namespaced_pod_exec, pod_name, namespace, command=command, container=container_name, stderr=True, stdin=False, stdout=True, tty=False ) return resp except Exception as e: return f"Error executing command: {str(e)}" def exec_command_with_stdin(self, namespace: str, pod_name: str, command: List[str], stdin_data: str, container_name: str = None) -> str: """Execute command in pod with stdin input - used for large file uploads to avoid command-line limits""" if not self.connected: return f"Mock command execution with stdin: {' '.join(command)}\n" try: resp = stream( v1.connect_get_namespaced_pod_exec, pod_name, namespace, command=command, container=container_name, stderr=True, stdin=True, stdout=True, tty=False, _preload_content=False ) # Write stdin data in chunks to avoid memory issues chunk_size = 8192 # 8KB chunks data_bytes = stdin_data.encode('utf-8') if isinstance(stdin_data, str) else stdin_data for i in range(0, len(data_bytes), chunk_size): chunk = data_bytes[i:i + chunk_size] resp.write_stdin(chunk) # Close stdin by closing the stream or sending EOF if supported # WSClient doesn't have close_stdin(), so we use close() or just wait # if the command is expected to finish (like head -c) try: # Some versions might support sending EOF character resp.write_stdin('\x04') except: pass # Read response output = "" max_wait_time = 300 # 5 minutes max wait for large files start_time = time.time() while resp.is_open() and (time.time() - start_time) < max_wait_time: resp.update(timeout=1) if resp.peek_stdout(): output += resp.read_stdout(timeout=1) if resp.peek_stderr(): error = resp.read_stderr(timeout=1) if error: output += error resp.close() if (time.time() - start_time) >= max_wait_time: logger.warning(f"Command with stdin timed out after {max_wait_time}s") return output except Exception as e: logger.error(f"Error executing command with stdin: {e}", exc_info=True) return f"Error executing command: {str(e)}" def list_files(self, namespace: str, pod_name: str, path: str = "/", container_name: str = None) -> List[Dict]: """List files and directories in a path - IMPROVED VERSION""" if not self.connected: # Return mock data for development mock_files = [ {"name": "eula.txt", "type": "file", "size": 1024, "modified": "2024-01-01T00:00:00Z"}, {"name": "server.properties", "type": "file", "size": 2048, "modified": "2024-01-01T00:00:00Z"}, {"name": "logs", "type": "directory", "size": 4096, "modified": "2024-01-01T00:00:00Z"}, {"name": "world", "type": "directory", "size": 4096, "modified": "2024-01-01T00:00:00Z"} ] return mock_files try: # Use ls command for reliable file listing cmd = ["ls", "-la", path] result = self.exec_command(namespace, pod_name, cmd, container_name=container_name) logger.info(f"LS result for {path}: {result}") files = [] lines = result.strip().split('\n') for line in lines: # Skip the "total X" line and empty lines if line.startswith('total ') or not line.strip(): continue try: parts = line.split() if len(parts) < 9: continue permissions = parts[0] name = ' '.join(parts[8:]) if name in ['.', '..']: continue file_type = "directory" if permissions.startswith('d') else "file" # Get file size from ls output (field 5) size = 0 try: size = int(parts[4]) except (ValueError, IndexError): pass # Parse modification time (fields 5-8) try: # Format: "Jan 01 12:00" or "Jan 01 2023" date_parts = parts[5:8] if len(date_parts) == 3: month, day, time_or_year = date_parts if ':' in time_or_year: # Current year, use current year year = datetime.now().year modified_str = f"{month} {day} {year} {time_or_year}" modified = datetime.strptime(modified_str, "%b %d %Y %H:%M") else: # Different year year = time_or_year modified_str = f"{month} {day} {year}" modified = datetime.strptime(modified_str, "%b %d %Y") else: modified = datetime.now() except Exception as e: logger.warning(f"Error parsing date: {e}") modified = datetime.now() files.append({ "name": name, "type": file_type, "size": size, "modified": modified.isoformat() }) except (ValueError, IndexError) as e: logger.warning(f"Error parsing file entry: {line}, error: {e}") continue return files except Exception as e: logger.error(f"Error listing files in {path}: {e}") return [] def read_file(self, namespace: str, pod_name: str, file_path: str, container_name: str = None) -> str: """Read file content from pod""" if not self.connected: return "Mock file content for development" try: cmd = ["cat", file_path] result = self.exec_command(namespace, pod_name, cmd, container_name=container_name) return result except Exception as e: logger.error(f"Error reading file {file_path}: {e}") return "" def read_file_base64(self, namespace: str, pod_name: str, file_path: str, container_name: str = None) -> str: """Read file content from pod as base64 encoded string""" if not self.connected: return "" try: import base64 cmd = ["base64", "-w", "0", file_path] result = self.exec_command(namespace, pod_name, cmd, container_name=container_name) return result.strip() except Exception as e: logger.error(f"Error reading file as base64 {file_path}: {e}") return "" def write_file(self, namespace: str, pod_name: str, file_path: str, content: str, container_name: str = None) -> bool: """Write content to file in pod using base64 for reliability""" if not self.connected: return True try: import base64 # Encode content to base64 b64_content = base64.b64encode(content.encode('utf-8')).decode('utf-8') # Use base64 -d to decode on the pod side # This is much more reliable than heredocs for content with special characters cmd = ["sh", "-c", f"echo '{b64_content}' | base64 -d > '{file_path}'"] self.exec_command(namespace, pod_name, cmd, container_name=container_name) return True except Exception as e: logger.error(f"Error writing file {file_path}: {e}") return False def create_directory(self, namespace: str, pod_name: str, dir_path: str, container_name: str = None) -> bool: """Create directory in pod""" if not self.connected: return True try: cmd = ["mkdir", "-p", dir_path] self.exec_command(namespace, pod_name, cmd, container_name=container_name) return True except Exception as e: logger.error(f"Error creating directory {dir_path}: {e}") return False def delete_path(self, namespace: str, pod_name: str, path: str, container_name: str = None) -> bool: """Delete file or directory in pod""" if not self.connected: return True try: cmd = ["rm", "-rf", path] self.exec_command(namespace, pod_name, cmd, container_name=container_name) return True except Exception as e: logger.error(f"Error deleting path {path}: {e}") return False def move_path(self, namespace: str, pod_name: str, src_path: str, dest_path: str, container_name: str = None) -> bool: """Move file or directory in pod""" if not self.connected: return True try: cmd = ["mv", src_path, dest_path] self.exec_command(namespace, pod_name, cmd, container_name=container_name) return True except Exception as e: logger.error(f"Error moving path {src_path} to {dest_path}: {e}") return False def copy_path(self, namespace: str, pod_name: str, src_path: str, dest_path: str, container_name: str = None) -> bool: """Copy file or directory in pod""" if not self.connected: return True try: cmd = ["cp", "-r", src_path, dest_path] self.exec_command(namespace, pod_name, cmd, container_name=container_name) return True except Exception as e: logger.error(f"Error copying path {src_path} to {dest_path}: {e}") return False def upload_file(self, namespace: str, pod_name: str, file_path: str, file_content: str, is_base64: bool = True) -> bool: """Upload file to pod""" if not self.connected: return True try: if is_base64: # Decode base64 content import base64 content = base64.b64decode(file_content).decode('utf-8', errors='ignore') else: content = file_content return self.write_file(namespace, pod_name, file_path, content) except Exception as e: logger.error(f"Error uploading file {file_path}: {e}") return False def exec_command_with_file_stdin(self, namespace: str, pod_name: str, command: List[str], file_path: str, base64_encode: bool = True, container_name: str = None) -> str: """Execute command in pod with stdin input from a file - streams file in chunks to avoid memory issues Used for large file uploads to avoid loading entire file into memory""" if not self.connected: return f"Mock command execution with file stdin: {' '.join(command)}\n" try: resp = stream( v1.connect_get_namespaced_pod_exec, pod_name, namespace, command=command, container=container_name, stderr=True, stdin=True, stdout=True, tty=False, _preload_content=False ) # Stream file in chunks to avoid memory issues # IMPORTANT: chunk_size must be a multiple of 3 for base64 encoding to work correctly across chunks chunk_size = 48 * 1024 # 48KB chunks (multiple of 3) base64_chunk_size = 8192 # 8KB chunks for base64 string writing with open(file_path, 'rb') as f: if base64_encode: # Read binary file, encode to base64, and stream to stdin import base64 encoder = base64.b64encode while True: # Read binary chunk binary_chunk = f.read(chunk_size) if not binary_chunk: break # Encode to base64 base64_chunk = encoder(binary_chunk).decode('utf-8') # Write base64 string in smaller chunks to stdin base64_bytes = base64_chunk.encode('utf-8') for i in range(0, len(base64_bytes), base64_chunk_size): chunk = base64_bytes[i:i + base64_chunk_size] resp.write_stdin(chunk) else: # Stream binary directly while True: chunk = f.read(chunk_size) if not chunk: break resp.write_stdin(chunk) # Close stdin by sending EOF if supported try: resp.write_stdin('\x04') except: pass # Read response output = "" max_wait_time = 600 # 10 minutes max wait for very large files start_time = time.time() while resp.is_open() and (time.time() - start_time) < max_wait_time: resp.update(timeout=1) if resp.peek_stdout(): output += resp.read_stdout(timeout=1) if resp.peek_stderr(): error = resp.read_stderr(timeout=1) if error: output += error resp.close() if (time.time() - start_time) >= max_wait_time: logger.warning(f"Command with file stdin timed out after {max_wait_time}s") return output except Exception as e: logger.error(f"Error executing command with file stdin: {e}", exc_info=True) return f"Error executing command: {str(e)}" def write_file_binary(self, namespace: str, pod_name: str, file_path: str, base64_content: str, container_name: str = None) -> bool: """Write binary file to pod using base64 decode on pod side (for large files) Uses stdin to avoid command-line length limits for large files""" if not self.connected: return True try: # Ensure directory exists dir_path = '/'.join(file_path.split('/')[:-1]) if '/' in file_path else '.' if dir_path and dir_path != '.': self.create_directory(namespace, pod_name, dir_path, container_name=container_name) escaped_path = file_path.replace("'", "'\\''") temp_file = f"/tmp/upload_{int(time.time() * 1000000)}.b64" escaped_temp = temp_file.replace("'", "'\\''") # Use stdin to write base64 data to avoid command-line length limits # Using head -c ensures the command terminates after reading the expected size b64_data = base64_content.encode('utf-8') if isinstance(base64_content, str) else base64_content content_len = len(b64_data) write_base64_cmd = ["sh", "-c", f"head -c {content_len} > '{escaped_temp}'"] write_result = self.exec_command_with_stdin(namespace, pod_name, write_base64_cmd, base64_content, container_name=container_name) # Check if write succeeded by verifying temp file exists check_temp_cmd = ["sh", "-c", f"test -f '{escaped_temp}' && echo 'temp_exists' || echo 'temp_missing'"] temp_check_result = self.exec_command(namespace, pod_name, check_temp_cmd, container_name=container_name) if 'temp_exists' not in temp_check_result: logger.error(f"Failed to write temp base64 file: {write_result}") return False # Decode base64 file to final location with error checking decode_cmd = ["sh", "-c", f"base64 -d < '{escaped_temp}' > '{escaped_path}' 2>&1 && rm -f '{escaped_temp}' && echo 'decode_success' || (rm -f '{escaped_temp}' && echo 'decode_failed')"] decode_result = self.exec_command(namespace, pod_name, decode_cmd, container_name=container_name) if 'decode_failed' in decode_result: logger.error(f"Failed to decode base64 file: {decode_result}") return False # Verify file was created and has content check_cmd = ["sh", "-c", f"test -f '{escaped_path}' && test -s '{escaped_path}' && echo 'exists' || echo 'missing'"] result = self.exec_command(namespace, pod_name, check_cmd, container_name=container_name) if 'exists' not in result: logger.error(f"File verification failed: {result}") return False return True except Exception as e: logger.error(f"Error writing binary file {file_path}: {e}") import traceback logger.error(traceback.format_exc()) return False def write_file_binary_from_disk(self, namespace: str, pod_name: str, file_path: str, local_file_path: str, container_name: str = None) -> bool: """Write binary file to pod from a local file on disk - streams file to avoid memory issues This is the preferred method for large files as it doesn't load the entire file into memory""" if not self.connected: return True try: # Ensure directory exists dir_path = '/'.join(file_path.split('/')[:-1]) if '/' in file_path else '.' if dir_path and dir_path != '.': self.create_directory(namespace, pod_name, dir_path, container_name=container_name) escaped_path = file_path.replace("'", "'\\''") temp_file = f"/tmp/upload_{int(time.time() * 1000000)}.b64" escaped_temp = temp_file.replace("'", "'\\''") # Stream file from disk, base64 encode, and write to pod temp file # Calculate expected base64 size to use head -c for automatic termination import math local_size = os.path.getsize(local_file_path) b64_size = math.ceil(local_size / 3) * 4 write_base64_cmd = ["sh", "-c", f"head -c {b64_size} > '{escaped_temp}'"] write_result = self.exec_command_with_file_stdin(namespace, pod_name, write_base64_cmd, local_file_path, base64_encode=True, container_name=container_name) # Check if write succeeded by verifying temp file exists check_temp_cmd = ["sh", "-c", f"test -f '{escaped_temp}' && echo 'temp_exists' || echo 'temp_missing'"] temp_check_result = self.exec_command(namespace, pod_name, check_temp_cmd, container_name=container_name) if 'temp_exists' not in temp_check_result: logger.error(f"Failed to write temp base64 file: {write_result}") return False # Decode base64 file to final location with error checking decode_cmd = ["sh", "-c", f"base64 -d < '{escaped_temp}' > '{escaped_path}' 2>&1 && rm -f '{escaped_temp}' && echo 'decode_success' || (rm -f '{escaped_temp}' && echo 'decode_failed')"] decode_result = self.exec_command(namespace, pod_name, decode_cmd, container_name=container_name) if 'decode_failed' in decode_result: logger.error(f"Failed to decode base64 file: {decode_result}") return False # Verify file was created and has content check_cmd = ["sh", "-c", f"test -f '{escaped_path}' && test -s '{escaped_path}' && echo 'exists' || echo 'missing'"] result = self.exec_command(namespace, pod_name, check_cmd, container_name=container_name) if 'exists' not in result: logger.error(f"File verification failed: {result}") return False return True except Exception as e: logger.error(f"Error writing binary file from disk {file_path}: {e}") import traceback logger.error(traceback.format_exc()) return False def download_file(self, namespace: str, pod_name: str, file_path: str, container_name: str = None) -> str: """Download file from pod as base64""" if not self.connected: return "" try: # Use read_file_base64 for binary-safe file reading # This handles both text and binary files correctly content = self.read_file_base64(namespace, pod_name, file_path, container_name=container_name) return content except Exception as e: logger.error(f"Error downloading file {file_path}: {e}") return "" k8s_manager = KubernetesManager()