Skip to content
Snippets Groups Projects
Commit cb6e0ec7 authored by Daniel Fässler's avatar Daniel Fässler
Browse files

Format using blue

parent 616182cc
Branches
No related tags found
No related merge requests found
Pipeline #337290 failed
Stage: tests
Stage: build
Hide whitespace changes
Inline Side-by-side
backend/app/server/providers/terraform_hetzner/__init__.py
+ 1
− 3
View file @ cb6e0ec7
from .templates import SupersetHetznerTemplate
from .templates import LinuxInstanceHetznerTemplate
__all__ = [
'LinuxInstanceHetznerTemplate','SupersetHetznerTemplate'
]
__all__ = ['LinuxInstanceHetznerTemplate', 'SupersetHetznerTemplate']
backend/app/server/providers/terraform_hetzner/base.py
+ 205
− 65
View file @ cb6e0ec7
......@@ -30,7 +30,7 @@ from server.server_registration import (
ServerInfo,
ServerTypeBase,
StopServerMixin,
)
)
logger = logging.getLogger(__name__)
......@@ -49,37 +49,45 @@ terraform_status_to_server_state = {
def _get_server_info():
try:
result = subprocess.run(["terraform", "output", "-json"], capture_output=True, text=True, check=True)
result = subprocess.run(
['terraform', 'output', '-json'],
capture_output=True,
text=True,
check=True,
)
output_json = json.loads(result.stdout.strip())
server_id = output_json.get("server_id", {}).get("value", "")
server_name = output_json.get("server_name", {}).get("value", "")
server_state = terraform_status_to_server_state.get(output_json.get("server_state", {}).get("value", "unknown"))
server_address = output_json.get("server_address", {}).get("value", "")
labels = output_json.get("server_labels", {}).get("value", {})
server_id = output_json.get('server_id', {}).get('value', '')
server_name = output_json.get('server_name', {}).get('value', '')
server_state = terraform_status_to_server_state.get(
output_json.get('server_state', {}).get('value', 'unknown')
)
server_address = output_json.get('server_address', {}).get('value', '')
labels = output_json.get('server_labels', {}).get('value', {})
return ServerInfo(
server_id=server_id,
server_name=server_name,
server_state=server_state,
created=datetime.now(),
server_address=server_address,
labels=labels
labels=labels,
)
except subprocess.CalledProcessError as e:
print(f"Error: {e}")
print(f'Error: {e}')
return None
_image_name: str = "ubuntu-22.04"
_server_type: str = "cx11"
_server_location: str = "nbg1"
HCLOUD_TOKEN: str = os.environ.get("HCLOUD_TOKEN")
_image_name: str = 'ubuntu-22.04'
_server_type: str = 'cx11'
_server_location: str = 'nbg1'
HCLOUD_TOKEN: str = os.environ.get('HCLOUD_TOKEN')
terraform_directory: str = "/terraform_workspace"
terraform_directory: str = '/terraform_workspace'
if not HCLOUD_TOKEN:
raise ValueError("HCLOUD_TOKEN missing from environment.")
raise ValueError('HCLOUD_TOKEN missing from environment.')
def _create_random_string(
......@@ -95,29 +103,49 @@ def _create_random_name():
_server_password: str = _create_random_string()
def apply_configuration(server_name, server_password, hcloud_token, server_type: Optional[str] = None, server_image: Optional[str] = None, server_location: Optional[str] = None, server_labels: Optional[str] = None, server_action: Optional[str] = None, server_password_reset: Optional[str] = None, destroy: Optional[str] = None):
def apply_configuration(
server_name,
server_password,
hcloud_token,
server_type: Optional[str] = None,
server_image: Optional[str] = None,
server_location: Optional[str] = None,
server_labels: Optional[str] = None,
server_action: Optional[str] = None,
server_password_reset: Optional[str] = None,
destroy: Optional[str] = None,
):
# Select Terraform workspace
subprocess.run(["terraform", "workspace", "select", "default"])
subprocess.run(['terraform', 'workspace', 'select', 'default'])
# Apply Terraform configuration with variable inputs
terraform_apply_args = ["terraform", "apply", "-auto-approve",
"-var", f"hcloud_token={hcloud_token}",
"-var", f"server_name={server_name}",
"-var", f"server_password={server_password}"]
terraform_apply_args = [
'terraform',
'apply',
'-auto-approve',
'-var',
f'hcloud_token={hcloud_token}',
'-var',
f'server_name={server_name}',
'-var',
f'server_password={server_password}',
]
if server_type:
terraform_apply_args.extend(["-var", f"server_type={server_type}"])
terraform_apply_args.extend(['-var', f'server_type={server_type}'])
if server_image:
terraform_apply_args.extend(["-var", f"server_image={server_image}"])
terraform_apply_args.extend(['-var', f'server_image={server_image}'])
if server_location:
terraform_apply_args.extend(["-var", f"server_location={server_location}"])
terraform_apply_args.extend(
['-var', f'server_location={server_location}']
)
if server_labels:
terraform_apply_args.extend(["-var", f"server_labels={server_labels}"])
terraform_apply_args.extend(['-var', f'server_labels={server_labels}'])
if server_action:
terraform_apply_args.extend(["-var", f"server_action={server_action}"])
terraform_apply_args.extend(['-var', f'server_action={server_action}'])
if server_password_reset:
terraform_apply_args.extend(["-var", "server_password_reset=true"])
terraform_apply_args.extend(['-var', 'server_password_reset=true'])
if destroy:
terraform_apply_args.extend(["-destroy"])
terraform_apply_args.extend(['-destroy'])
subprocess.run(terraform_apply_args)
......@@ -130,7 +158,9 @@ def create_terraform_server(
location,
description: str,
) -> ServerCreatedInfo:
server_name: str = f"{server_variant}-{_create_random_name()}-{_create_random_name()}"
server_name: str = (
f'{server_variant}-{_create_random_name()}-{_create_random_name()}'
)
server_password = _server_password
server_type: str = instance_type
server_image: str = image_name
......@@ -141,9 +171,9 @@ def create_terraform_server(
.replace('+', '-')
)
labels = {
"usage": f"{server_variant}",
"created-on": f"{created_date}",
"username": f"{username}",
'usage': f'{server_variant}',
'created-on': f'{created_date}',
'username': f'{username}',
}
labels_json = json.dumps(labels)
......@@ -152,9 +182,17 @@ def create_terraform_server(
exit(1)
os.chdir(terraform_directory)
subprocess.run(["terraform","init"])
apply_configuration(server_name, server_password, HCLOUD_TOKEN, server_type, server_image, server_location, labels_json)
subprocess.run(['terraform', 'init'])
apply_configuration(
server_name,
server_password,
HCLOUD_TOKEN,
server_type,
server_image,
server_location,
labels_json,
)
info = _get_server_info()
if info:
return ServerCreatedInfo(
......@@ -165,74 +203,154 @@ def create_terraform_server(
server_address=info.server_address,
labels=info.labels,
description=description,
server_user="root",
server_user='root',
server_password=server_password,
)
def status() -> ServerInfo:
return _get_server_info()
def reboot(server_name, server_password, server_type, server_image, server_location, labels) -> ServerInfo:
def reboot(
server_name,
server_password,
server_type,
server_image,
server_location,
labels,
) -> ServerInfo:
if not os.path.isdir(terraform_directory):
print(f"Directory '{terraform_directory}' does not exist.")
exit(1)
os.chdir(terraform_directory)
apply_configuration(server_name, server_password, HCLOUD_TOKEN, server_type, server_image, server_location, labels, server_action="reboot")
apply_configuration(
server_name,
server_password,
HCLOUD_TOKEN,
server_type,
server_image,
server_location,
labels,
server_action='reboot',
)
# wait for server to be up again
sleep(30)
return _get_server_info()
def stop(server_name, server_password, server_type, server_image, server_location, labels) -> ServerInfo:
def stop(
server_name,
server_password,
server_type,
server_image,
server_location,
labels,
) -> ServerInfo:
info = _get_server_info()
if not os.path.isdir(terraform_directory):
print(f"Directory '{terraform_directory}' does not exist.")
exit(1)
os.chdir(terraform_directory)
apply_configuration(info.server_name, server_password, HCLOUD_TOKEN, "", "", "", info.labels, server_action="poweroff")
apply_configuration(
info.server_name,
server_password,
HCLOUD_TOKEN,
'',
'',
'',
info.labels,
server_action='poweroff',
)
# wait for server to be up again
sleep(30)
return _get_server_info()
def start(server_name, server_password, server_type, server_image, server_location, labels) -> ServerInfo:
def start(
server_name,
server_password,
server_type,
server_image,
server_location,
labels,
) -> ServerInfo:
if not os.path.isdir(terraform_directory):
print(f"Directory '{terraform_directory}' does not exist.")
exit(1)
os.chdir(terraform_directory)
apply_configuration(server_name, server_password, HCLOUD_TOKEN, server_type, server_image, server_location, labels, server_action="poweron")
apply_configuration(
server_name,
server_password,
HCLOUD_TOKEN,
server_type,
server_image,
server_location,
labels,
server_action='poweron',
)
# wait for server to be up again
sleep(30)
return _get_server_info()
def reset_pw(server_name, server_password, server_type, server_image, server_location, labels) -> ServerPasswordResetInfo:
def reset_pw(
server_name,
server_password,
server_type,
server_image,
server_location,
labels,
) -> ServerPasswordResetInfo:
if not os.path.isdir(terraform_directory):
print(f"Directory '{terraform_directory}' does not exist.")
exit(1)
os.chdir(terraform_directory)
apply_configuration(server_name, server_password, HCLOUD_TOKEN, server_type, server_image, server_location, labels, server_action=None, server_password_reset=True)
apply_configuration(
server_name,
server_password,
HCLOUD_TOKEN,
server_type,
server_image,
server_location,
labels,
server_action=None,
server_password_reset=True,
)
# wait for server to be up again
sleep(30)
return _get_server_info()
def destroy(server_name, server_password, server_type, server_image, server_location, labels) -> ServerDeletedInfo:
def destroy(
server_name,
server_password,
server_type,
server_image,
server_location,
labels,
) -> ServerDeletedInfo:
if not os.path.isdir(terraform_directory):
print(f"Directory '{terraform_directory}' does not exist.")
exit(1)
os.chdir(terraform_directory)
apply_configuration(server_name, server_password, HCLOUD_TOKEN, server_type, server_image, server_location, labels, server_action=None)
sleep(30)
return ServerDeletedInfo(
deleted=True
apply_configuration(
server_name,
server_password,
HCLOUD_TOKEN,
server_type,
server_image,
server_location,
labels,
server_action=None,
)
sleep(30)
return ServerDeletedInfo(deleted=True)
class ServerTypeTerraform(
......@@ -242,13 +360,13 @@ class ServerTypeTerraform(
StartServerMixin,
ServerTypeBase,
):
server_variant: str = ""
server_variant: str = ''
location: str = _server_location
instance_type = _server_type
image_name: str = _image_name
def create_instance(
self, model_instance_id, *args, **kwargs
self, model_instance_id, *args, **kwargs
) -> ServerCreatedInfo:
from server.models import ProvisionedServerInstance
......@@ -262,9 +380,8 @@ class ServerTypeTerraform(
username=username,
image_name=self.image_name,
location=self.location,
description=server_instance.server_type.description or "",
description=server_instance.server_type.description or '',
)
def get_server_info(
self, model_instance_id: str, *args, **kwargs
......@@ -272,28 +389,51 @@ class ServerTypeTerraform(
instance = self.get_server_instance(model_instance_id)
return status()
def reset_password(
self, model_instance_id, *args, **kwargs
) -> ServerPasswordResetInfo:
instance = self.get_server_instance(model_instance_id)
return reset_pw(instance.server_name, instance.server_password, _server_type, _image_name, _server_location, None)
return reset_pw(
instance.server_name,
instance.server_password,
_server_type,
_image_name,
_server_location,
None,
)
def start_server(self, model_instance_id, *args, **kwargs) -> ServerInfo:
instance = self.get_server_instance(model_instance_id)
return start(instance.server_name, instance.server_password, _server_type, _image_name, _server_location, None)
return start(
instance.server_name,
instance.server_password,
_server_type,
_image_name,
_server_location,
None,
)
def restart_server(self, model_instance_id, *args, **kwargs) -> ServerInfo:
instance = self.get_server_instance(model_instance_id)
return reboot(instance.server_name, instance.server_password, _server_type, _image_name, _server_location, None)
return reboot(
instance.server_name,
instance.server_password,
_server_type,
_image_name,
_server_location,
None,
)
def stop_server(self, model_instance_id, *args, **kwargs) -> ServerInfo:
instance = self.get_server_instance(model_instance_id)
return stop(instance.server_name, instance.server_password, _server_type, _image_name, _server_location, None)
return stop(
instance.server_name,
instance.server_password,
_server_type,
_image_name,
_server_location,
None,
)
def delete_server(
self, model_instance_id, *args, **kwargs
......@@ -310,7 +450,7 @@ class ServerTypeTerraform(
deleted = True
except APIException:
logger.exception(
f"Could not delete hetzner server with id {instance.server_id} but continuing anyway."
f'Could not delete hetzner server with id {instance.server_id} but continuing anyway.'
)
return ServerDeletedInfo(server_id=model_instance_id, deleted=deleted)
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment