Coverage for src / server_list / spec / models.py: 97%

1#!/usr/bin/env python3 

2""" 

3Data models for server-list. 

4 

5Provides typed dataclasses for VM info, host info, CPU benchmarks, etc. 

6Used internally by data_collector.py and cpu_benchmark.py. 

7""" 

8 

9from dataclasses import dataclass 

10 

11 

12@dataclass 

13class VMInfo: 

14 """VM information collected from ESXi.""" 

15 

16 esxi_host: str 

17 vm_name: str 

18 cpu_count: int | None 

19 ram_mb: int | None 

20 storage_gb: float | None 

21 power_state: str | None 

22 cpu_usage_mhz: int | None = None 

23 memory_usage_mb: int | None = None 

24 collected_at: str | None = None 

25 

26 @classmethod 

27 def parse_row(cls, row: tuple, esxi_host: str) -> "VMInfo": 

28 """Create VMInfo from DB row (without esxi_host column).""" 

29 return cls( 

30 esxi_host=esxi_host, 

31 vm_name=row[0], 

32 cpu_count=row[1], 

33 ram_mb=row[2], 

34 storage_gb=round(row[3], 1) if row[3] else None, 

35 power_state=row[4], 

36 cpu_usage_mhz=row[5], 

37 memory_usage_mb=row[6], 

38 collected_at=row[7], 

39 ) 

40 

41 @classmethod 

42 def parse_row_full(cls, row: tuple) -> "VMInfo": 

43 """Create VMInfo from DB row (with esxi_host column).""" 

44 return cls( 

45 vm_name=row[0], 

46 cpu_count=row[1], 

47 ram_mb=row[2], 

48 storage_gb=round(row[3], 1) if row[3] else None, 

49 power_state=row[4], 

50 esxi_host=row[5], 

51 cpu_usage_mhz=row[6], 

52 memory_usage_mb=row[7], 

53 collected_at=row[8], 

54 ) 

55 

56 

57@dataclass 

58class HostInfo: 

59 """Host information collected from ESXi or Prometheus.""" 

60 

61 host: str 

62 boot_time: str | None 

63 uptime_seconds: float | None 

64 status: str 

65 cpu_threads: int | None = None 

66 cpu_cores: int | None = None 

67 os_version: str | None = None 

68 cpu_usage_percent: float | None = None 

69 memory_usage_percent: float | None = None 

70 memory_total_bytes: float | None = None 

71 memory_used_bytes: float | None = None 

72 collected_at: str | None = None 

73 

74 @classmethod 

75 def parse_row(cls, row: tuple) -> "HostInfo": 

76 """Create HostInfo from DB row.""" 

77 return cls( 

78 host=row[0], 

79 boot_time=row[1], 

80 uptime_seconds=row[2], 

81 status=row[3], 

82 cpu_threads=row[4], 

83 cpu_cores=row[5], 

84 os_version=row[6], 

85 cpu_usage_percent=row[7], 

86 memory_usage_percent=row[8], 

87 memory_total_bytes=row[9], 

88 memory_used_bytes=row[10], 

89 collected_at=row[11], 

90 ) 

91 

92 

93@dataclass 

94class PowerInfo: 

95 """Power consumption information from iLO.""" 

96 

97 power_watts: int | None 

98 power_average_watts: int | None 

99 power_max_watts: int | None 

100 power_min_watts: int | None 

101 collected_at: str | None = None 

102 

103 @classmethod 

104 def parse_row(cls, row: tuple) -> "PowerInfo": 

105 """Create PowerInfo from DB row (without host column).""" 

106 return cls( 

107 power_watts=row[0], 

108 power_average_watts=row[1], 

109 power_max_watts=row[2], 

110 power_min_watts=row[3], 

111 collected_at=row[4], 

112 ) 

113 

114 @classmethod 

115 def parse_row_with_host(cls, row: tuple) -> tuple[str, "PowerInfo"]: 

116 """Create (host, PowerInfo) tuple from DB row (with host column).""" 

117 return ( 

118 row[0], 

119 cls( 

120 power_watts=row[1], 

121 power_average_watts=row[2], 

122 power_max_watts=row[3], 

123 power_min_watts=row[4], 

124 collected_at=row[5], 

125 ), 

126 ) 

127 

128 

129@dataclass 

130class CollectionStatus: 

131 """Data collection status for a host.""" 

132 

133 host: str 

134 last_fetch: str | None 

135 status: str 

136 

137 @classmethod 

138 def parse_row(cls, row: tuple) -> "CollectionStatus": 

139 """Create CollectionStatus from DB row.""" 

140 return cls( 

141 host=row[0], 

142 last_fetch=row[1], 

143 status=row[2], 

144 ) 

145 

146 

147@dataclass 

148class ZfsPoolInfo: 

149 """ZFS pool information from Prometheus.""" 

150 

151 pool_name: str 

152 size_bytes: float | None 

153 allocated_bytes: float | None 

154 free_bytes: float | None 

155 health: float | None 

156 collected_at: str | None = None 

157 

158 @classmethod 

159 def parse_row(cls, row: tuple) -> "ZfsPoolInfo": 

160 """Create ZfsPoolInfo from DB row.""" 

161 return cls( 

162 pool_name=row[0], 

163 size_bytes=row[1], 

164 allocated_bytes=row[2], 

165 free_bytes=row[3], 

166 health=row[4], 

167 collected_at=row[5], 

168 ) 

169 

170 

171@dataclass 

172class MountInfo: 

173 """Mount point information from Prometheus.""" 

174 

175 mountpoint: str 

176 size_bytes: float | None 

177 avail_bytes: float | None 

178 used_bytes: float | None 

179 collected_at: str | None = None 

180 

181 @classmethod 

182 def parse_row(cls, row: tuple) -> "MountInfo": 

183 """Create MountInfo from DB row.""" 

184 return cls( 

185 mountpoint=row[0], 

186 size_bytes=row[1], 

187 avail_bytes=row[2], 

188 used_bytes=row[3], 

189 collected_at=row[4], 

190 ) 

191 

192 

193@dataclass 

194class CPUBenchmark: 

195 """CPU benchmark scores from cpubenchmark.net.""" 

196 

197 cpu_name: str 

198 multi_thread_score: int | None 

199 single_thread_score: int | None 

200 

201 

202@dataclass 

203class UsageMetrics: 

204 """CPU/メモリ使用率データ (Prometheus から取得).""" 

205 

206 cpu_usage_percent: float | None = None 

207 memory_usage_percent: float | None = None 

208 memory_total_bytes: float | None = None 

209 memory_used_bytes: float | None = None 

210 

211 

212@dataclass 

213class UptimeData: 

214 """稼働時間データ (Prometheus から取得).""" 

215 

216 boot_time: str 

217 uptime_seconds: float 

218 status: str 

219 

220 

221@dataclass 

222class StorageMetrics: 

223 """ストレージメトリクス (Prometheus から取得).""" 

224 

225 size_bytes: float 

226 avail_bytes: float 

227 used_bytes: float 

228 

229 

230@dataclass 

231class UPSInfo: 

232 """UPS information from NUT (Network UPS Tools).""" 

233 

234 ups_name: str 

235 host: str 

236 model: str | None = None 

237 battery_charge: float | None = None # % 

238 battery_runtime: int | None = None # seconds 

239 ups_load: float | None = None # % 

240 ups_status: str | None = None # OL, OB, etc. 

241 ups_temperature: float | None = None # °C 

242 input_voltage: float | None = None 

243 output_voltage: float | None = None 

244 collected_at: str | None = None 

245 

246 @classmethod 

247 def parse_row(cls, row: tuple) -> "UPSInfo": 

248 """Create UPSInfo from DB row.""" 

249 return cls( 

250 ups_name=row[0], 

251 host=row[1], 

252 model=row[2], 

253 battery_charge=row[3], 

254 battery_runtime=row[4], 

255 ups_load=row[5], 

256 ups_status=row[6], 

257 ups_temperature=row[7], 

258 input_voltage=row[8], 

259 output_voltage=row[9], 

260 collected_at=row[10], 

261 ) 

262 

263 

264@dataclass 

265class UPSClient: 

266 """Client connected to UPS (from NUT).""" 

267 

268 ups_name: str 

269 host: str 

270 client_ip: str 

271 client_hostname: str | None = None 

272 esxi_host: str | None = None # If client is a VM, the ESXi host running it 

273 machine_name: str | None = None # Machine name for linking (ESXi host for VMs, hostname for physical) 

274 collected_at: str | None = None 

275 

276 @classmethod 

277 def parse_row(cls, row: tuple) -> "UPSClient": 

278 """Create UPSClient from DB row.""" 

279 return cls( 

280 ups_name=row[0], 

281 host=row[1], 

282 client_ip=row[2], 

283 client_hostname=row[3], 

284 esxi_host=row[4], 

285 machine_name=row[5], 

286 collected_at=row[6], 

287 )