IaC for a Tangled Knot
juanlu.space
1#!/usr/bin/env python3
2"""
3Dynamic inventory script for Clouding.io servers.
4Fetches server list from Clouding API and generates Ansible inventory.
5
6Requirements:
7- CLOUDING_TOKEN environment variable with API token
8
9Usage:
10 ansible-inventory -i inventory/prod/clouding.py --list
11 ansible-playbook -i inventory/prod/clouding.py playbook.yaml
12"""
13
14from __future__ import annotations
15
16import json
17import os
18import sys
19from urllib.request import Request, urlopen
20from urllib.error import URLError, HTTPError
21from dataclasses import dataclass
22
23
24@dataclass
25class CloudingInventory:
26 """Dynamic inventory for Clouding.io servers."""
27
28 inventory: dict
29 api_token: str
30 endpoint: str = "https://api.clouding.io/v1/servers/"
31
32 @classmethod
33 def create(cls, endpoint: str | None = None) -> CloudingInventory:
34 api_token = os.environ.get("CLOUDING_TOKEN")
35 if not api_token:
36 print("Error: CLOUDING_TOKEN environment variable not set", file=sys.stderr)
37 sys.exit(1)
38
39 inventory = {"_meta": {"hostvars": {}}, "all": {"children": ["ungrouped"]}}
40
41 if endpoint:
42 return cls(inventory, api_token, endpoint)
43 else:
44 return cls(inventory, api_token)
45
46 def fetch_servers(self):
47 """Fetch server list from Clouding API."""
48 headers = {"X-API-KEY": self.api_token, "Accept": "application/json"}
49
50 try:
51 request = Request(self.endpoint, headers=headers)
52 with urlopen(request) as response:
53 data = json.loads(response.read().decode("utf-8"))
54 return data.get("servers", [])
55 except HTTPError as e:
56 print(f"HTTP Error {e.code}: {e.reason}", file=sys.stderr)
57 sys.exit(1)
58 except URLError as e:
59 print(f"URL Error: {e.reason}", file=sys.stderr)
60 sys.exit(1)
61 except Exception as e:
62 print(f"Error fetching servers: {e}", file=sys.stderr)
63 sys.exit(1)
64
65 def add_server_to_inventory(self, server):
66 """Add a server to the inventory."""
67 server_id = server.get("id")
68 server_name = server.get("name")
69 status = server.get("status")
70 power_state = server.get("powerState")
71 public_ip = server.get("publicIp")
72
73 if status != "Active" or power_state != "Running":
74 return
75
76 if not public_ip:
77 return
78
79 self.inventory["_meta"]["hostvars"][server_name] = {
80 "ansible_host": public_ip,
81 "ansible_user": "root", # Clouding uses root by default
82 "ansible_python_interpreter": "auto_silent",
83 "clouding_id": server_id,
84 "clouding_hostname": server.get("hostname"),
85 "clouding_status": status,
86 "clouding_power_state": power_state,
87 "clouding_vcores": server.get("vCores"),
88 "clouding_ram_gb": server.get("ramGb"),
89 "clouding_flavor": server.get("flavor"),
90 "clouding_volume_size_gb": server.get("volumeSizeGb"),
91 "clouding_dns_address": server.get("dnsAddress"),
92 }
93
94 image = server.get("image", {})
95 if image:
96 self.inventory["_meta"]["hostvars"][server_name].update(
97 {
98 "clouding_image_id": image.get("id"),
99 "clouding_image_name": image.get("name"),
100 }
101 )
102
103 # Add to 'clouding' group
104 if "clouding" not in self.inventory:
105 self.inventory["clouding"] = {"hosts": []}
106 self.inventory["all"]["children"].append("clouding")
107
108 if server_name not in self.inventory["clouding"]["hosts"]:
109 self.inventory["clouding"]["hosts"].append(server_name)
110
111 def generate_inventory(self):
112 """Generate the complete inventory."""
113 servers = self.fetch_servers()
114 for server in servers:
115 self.add_server_to_inventory(server)
116 return self.inventory
117
118
119def main():
120 """Main entry point."""
121 inventory = CloudingInventory.create()
122 result = inventory.generate_inventory()
123 print(json.dumps(result, indent=2))
124
125
126if __name__ == "__main__":
127 main()