Skip to content

lablink_allocator_service.utils.tofu_utils

Get the instance IDs of the instances created by OpenTofu. Args: tofu_dir (str): The directory where the OpenTofu configuration is located. Raises: RuntimeError: Error running tofu output command. RuntimeError: Error decoding JSON output. ValueError: Expected output to be a list of instance IDs. Returns: list: A list of instance IDs of the instances.

Source code in packages/allocator/src/lablink_allocator_service/utils/tofu_utils.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def get_instance_ids(tofu_dir: str) -> list:
    """Get the instance IDs of the instances created by OpenTofu.
    Args:
        tofu_dir (str): The directory where the OpenTofu configuration is located.
    Raises:
        RuntimeError: Error running tofu output command.
        RuntimeError: Error decoding JSON output.
        ValueError: Expected output to be a list of instance IDs.
    Returns:
        list: A list of instance IDs of the instances.
    """
    tofu_dir = Path(tofu_dir)
    try:
        result = subprocess.run(
            ["tofu", "output", "-json", "vm_instance_ids"],
            cwd=tofu_dir,
            capture_output=True,
            text=True,
            check=True,
        )
    except subprocess.CalledProcessError as e:
        raise RuntimeError(f"Error running tofu output: {e.stderr}")
    try:
        output = json.loads(result.stdout)
    except json.JSONDecodeError as e:
        raise RuntimeError(f"Error decoding JSON output: {e}")
    if not isinstance(output, list):
        raise ValueError("Expected output to be a list of instance IDs")
    return output

Get the names of the instances created by OpenTofu. Args: tofu_dir (str): The directory where the OpenTofu configuration is located. Raises: RuntimeError: Error running tofu output command. RuntimeError: Error decoding JSON output. ValueError: Expected output to be a list of instance names. Returns: list: A list of names assigned to the EC2 instances.

Source code in packages/allocator/src/lablink_allocator_service/utils/tofu_utils.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def get_instance_names(tofu_dir: str) -> list:
    """Get the names of the instances created by OpenTofu.
    Args:
        tofu_dir (str): The directory where the OpenTofu configuration is located.
    Raises:
        RuntimeError: Error running tofu output command.
        RuntimeError: Error decoding JSON output.
        ValueError: Expected output to be a list of instance names.
    Returns:
        list: A list of names assigned to the EC2 instances.
    """
    tofu_dir = Path(tofu_dir)
    try:
        result = subprocess.run(
            ["tofu", "output", "-json", "vm_instance_names"],
            cwd=tofu_dir,
            capture_output=True,
            text=True,
            check=True,
        )
    except subprocess.CalledProcessError as e:
        raise RuntimeError(f"Error running tofu output: {e.stderr}")
    try:
        output = json.loads(result.stdout)
    except json.JSONDecodeError as e:
        raise RuntimeError(f"Error decoding JSON output: {e}")
    if not isinstance(output, list):
        raise ValueError("Expected output to be a list of instance names")
    return output

Get the launch times of the instances created by OpenTofu. Args: tofu_dir (str): The directory where the OpenTofu configuration is located. Raises: RuntimeError: Error running tofu output command. RuntimeError: Error decoding JSON output. ValueError: Expected output to be a dictionary of launch times. Returns: dict: A dictionary mapping instance names to their launch times.

Source code in packages/allocator/src/lablink_allocator_service/utils/tofu_utils.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def get_instance_timings(tofu_dir: str) -> dict:
    """Get the launch times of the instances created by OpenTofu.
    Args:
        tofu_dir (str): The directory where the OpenTofu configuration is located.
    Raises:
        RuntimeError: Error running tofu output command.
        RuntimeError: Error decoding JSON output.
        ValueError: Expected output to be a dictionary of launch times.
    Returns:
        dict: A dictionary mapping instance names to their launch times.
    """
    tofu_dir = Path(tofu_dir)
    try:
        result = subprocess.run(
            ["tofu", "output", "-json", "instance_terraform_apply_times"],
            cwd=tofu_dir,
            capture_output=True,
            text=True,
            check=True,
        )
    except subprocess.CalledProcessError as e:
        raise RuntimeError(f"Error running tofu output: {e.stderr}")
    try:
        timing_data = json.loads(result.stdout)
    except json.JSONDecodeError as e:
        raise RuntimeError(f"Error decoding JSON output: {e}")
    if not isinstance(timing_data, dict):
        raise ValueError("Expected output to be a dictionary of launch times.")

    return timing_data

Get the SSH private key used for connecting to the instances. Args: tofu_dir (str): The directory where the OpenTofu configuration is located. Raises: RuntimeError: Error running tofu output command. Returns: str: The path to the SSH private key file.

Source code in packages/allocator/src/lablink_allocator_service/utils/tofu_utils.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def get_ssh_private_key(tofu_dir: str) -> str:
    """Get the SSH private key used for connecting to the instances.
    Args:
        tofu_dir (str): The directory where the OpenTofu configuration is located.
    Raises:
        RuntimeError: Error running tofu output command.
    Returns:
        str: The path to the SSH private key file.
    """
    tofu_dir = Path(tofu_dir)
    try:
        result = subprocess.run(
            ["tofu", "output", "-raw", "lablink_private_key_pem"],
            cwd=tofu_dir,
            capture_output=True,
            text=True,
            check=True,
        )
    except subprocess.CalledProcessError as e:
        raise RuntimeError(f"Error running tofu output: {e.stderr}")
    key_path = "/tmp/lablink_key.pem"
    with open(key_path, "w") as f:
        f.write(result.stdout)
    os.chmod(key_path, 0o400)
    return key_path