Create a new VM.
(self, args)
| 829 | print(f"Compose hash: {compose_hash}") |
| 830 | |
| 831 | def create_vm(self, args) -> None: |
| 832 | """Create a new VM.""" |
| 833 | # Read and validate compose file |
| 834 | if not os.path.exists(args.compose): |
| 835 | raise Exception(f"Compose file not found: {args.compose}") |
| 836 | |
| 837 | compose_content = read_utf8(args.compose) |
| 838 | |
| 839 | envs = parse_env_file(args.env_file) |
| 840 | |
| 841 | # Validate: --env-file requires --kms-url and kms_enabled in compose |
| 842 | if envs: |
| 843 | if not args.kms_url: |
| 844 | raise Exception( |
| 845 | "--env-file requires --kms-url to encrypt environment variables" |
| 846 | ) |
| 847 | try: |
| 848 | compose_json = json.loads(compose_content) |
| 849 | if not compose_json.get("kms_enabled", False): |
| 850 | raise Exception( |
| 851 | "--env-file requires kms_enabled=true in the compose file (use --kms when creating compose)" |
| 852 | ) |
| 853 | except json.JSONDecodeError: |
| 854 | pass # Let the server handle invalid JSON |
| 855 | |
| 856 | # Read user config file if provided |
| 857 | user_config = "" |
| 858 | if args.user_config: |
| 859 | user_config = read_utf8(args.user_config) |
| 860 | |
| 861 | # Create VM request |
| 862 | params = { |
| 863 | "name": args.name, |
| 864 | "image": args.image, |
| 865 | "compose_file": compose_content, |
| 866 | "vcpu": args.vcpu, |
| 867 | "memory": args.memory, |
| 868 | "disk_size": args.disk, |
| 869 | "app_id": args.app_id, |
| 870 | "user_config": user_config, |
| 871 | "ports": [parse_port_mapping(port) for port in args.port or []], |
| 872 | "hugepages": args.hugepages, |
| 873 | "pin_numa": args.pin_numa, |
| 874 | "stopped": args.stopped, |
| 875 | "no_tee": args.no_tee, |
| 876 | } |
| 877 | if args.swap is not None: |
| 878 | swap_bytes = max(0, int(round(args.swap)) * 1024 * 1024) |
| 879 | if swap_bytes > 0: |
| 880 | params["swap_size"] = swap_bytes |
| 881 | |
| 882 | if args.ppcie or (args.gpu and "all" in args.gpu): |
| 883 | params["gpus"] = {"attach_mode": "all"} |
| 884 | elif args.gpu: |
| 885 | params["gpus"] = { |
| 886 | "attach_mode": "listed", |
| 887 | "gpus": [{"slot": gpu} for gpu in args.gpu or []], |
| 888 | } |
no test coverage detected