339 lines
12 KiB
Python
339 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Cloud Instance Status Checker
|
|
Проверяет статус ресурсов в облаке по манифесту Terraform
|
|
"""
|
|
|
|
import json
|
|
import requests
|
|
import sys
|
|
from typing import Dict, Optional, List
|
|
from dataclasses import dataclass
|
|
from enum import Enum
|
|
from datetime import datetime
|
|
|
|
|
|
class InstanceState(Enum):
|
|
"""Возможные состояния инстанса"""
|
|
NOT_CREATED = "not_created"
|
|
CREATING = "creating"
|
|
RUNNING = "running"
|
|
MODIFYING = "modifying"
|
|
SUSPENDING = "suspending"
|
|
SUSPENDED = "suspended"
|
|
RESUMING = "resuming"
|
|
DELETING = "deleting"
|
|
DELETED = "deleted"
|
|
OPERATION_PENDING = "operation_pending"
|
|
OPERATION_FAILED = "operation_failed"
|
|
UNKNOWN = "unknown"
|
|
|
|
|
|
@dataclass
|
|
class InstanceInfo:
|
|
"""Информация об инстансе"""
|
|
uid: str
|
|
name: str
|
|
state: InstanceState
|
|
is_created: bool
|
|
is_deleted: bool
|
|
is_suspended: bool
|
|
uptime_seconds: float
|
|
operation_in_progress: bool
|
|
operation_pending: bool
|
|
last_operation: Optional[str]
|
|
last_operation_status: Optional[str]
|
|
error_log: Optional[str]
|
|
timestamp: str
|
|
|
|
|
|
class CloudInstanceChecker:
|
|
"""Проверяет статус инстансов в облаке"""
|
|
|
|
BASE_URL = "https://deck-api.ngcloud.ru/api/v1/index.cfm"
|
|
|
|
def __init__(self, token: str):
|
|
"""
|
|
Инициализация чекера
|
|
|
|
Args:
|
|
token: JWT токен для аутентификации в API
|
|
"""
|
|
self.token = token
|
|
self.session = requests.Session()
|
|
self.session.headers.update({
|
|
"Authorization": f"Bearer {token}",
|
|
"Content-Type": "application/json"
|
|
})
|
|
|
|
def find_instance(self, display_name: str, service_type: Optional[str] = None) -> Optional[Dict]:
|
|
"""
|
|
Найти инстанс по отображаемому имени
|
|
|
|
Args:
|
|
display_name: Имя ресурса в облаке
|
|
service_type: Тип сервиса (опционально, для фильтрации)
|
|
|
|
Returns:
|
|
Данные инстанса или None если не найден
|
|
"""
|
|
try:
|
|
params = {
|
|
"fields": "instanceConfigDtCreated,instanceUid,displayName,svc,uptime,explainedStatus,svcExtendedName,updaterLogin,updaterShortname,operationIsInProgress,operationIsPending,monitoringUrl",
|
|
"search": display_name,
|
|
"isAuxiliary": "false",
|
|
"isDeleted": "false",
|
|
"page": "1",
|
|
"pageSize": "100"
|
|
}
|
|
|
|
response = self.session.get(f"{self.BASE_URL}/instances", params=params)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
results = data.get("results", [])
|
|
|
|
# Поиск точного совпадения по имени
|
|
for instance in results:
|
|
if instance.get("displayName") == display_name:
|
|
if service_type is None or instance.get("svc") == service_type:
|
|
return instance
|
|
|
|
return None
|
|
except requests.RequestException as e:
|
|
print(f"❌ Ошибка при поиске инстанса '{display_name}': {e}", file=sys.stderr)
|
|
return None
|
|
|
|
def get_instance_details(self, instance_uid: str) -> Optional[Dict]:
|
|
"""
|
|
Получить полную информацию об инстансе
|
|
|
|
Args:
|
|
instance_uid: UUID инстанса
|
|
|
|
Returns:
|
|
Полные данные инстанса или None при ошибке
|
|
"""
|
|
try:
|
|
params = {
|
|
"fields": "instanceConfigDtCreated,instanceUid,displayName,descr,svc,state,operations,availableOperations,uptime,isDeleted,updaterLogin,updaterShortname,explainedStatus,man,dependencies,dependentInstances,svcExtendedName"
|
|
}
|
|
|
|
response = self.session.get(f"{self.BASE_URL}/instances/{instance_uid}", params=params)
|
|
response.raise_for_status()
|
|
|
|
return response.json().get("instance")
|
|
except requests.RequestException as e:
|
|
print(f"❌ Ошибка при получении информации об инстансе {instance_uid}: {e}", file=sys.stderr)
|
|
return None
|
|
|
|
def determine_state(self, instance_data: Dict) -> InstanceState:
|
|
"""
|
|
Определить состояние инстанса на основе флагов
|
|
|
|
Args:
|
|
instance_data: Данные инстанса от API
|
|
|
|
Returns:
|
|
InstanceState enum
|
|
"""
|
|
# Проверка удаления
|
|
if instance_data.get("isDeleted"):
|
|
return InstanceState.DELETED
|
|
|
|
# Проверка создания
|
|
if not instance_data.get("isCreated"):
|
|
return InstanceState.NOT_CREATED
|
|
|
|
# Проверка операции в процессе
|
|
if instance_data.get("operationIsInProgress"):
|
|
operations = instance_data.get("operations", [])
|
|
if operations:
|
|
last_op = operations[-1]
|
|
operation = last_op.get("operation", "unknown")
|
|
|
|
if operation == "create":
|
|
return InstanceState.CREATING
|
|
elif operation == "modify":
|
|
return InstanceState.MODIFYING
|
|
elif operation == "delete":
|
|
return InstanceState.DELETING
|
|
elif operation == "suspend":
|
|
return InstanceState.SUSPENDING
|
|
elif operation == "resume":
|
|
return InstanceState.RESUMING
|
|
|
|
# Проверка ожидающей операции
|
|
if instance_data.get("operationIsPending"):
|
|
operations = instance_data.get("operations", [])
|
|
if operations:
|
|
last_op = operations[-1]
|
|
if not last_op.get("dtStart"):
|
|
return InstanceState.OPERATION_PENDING
|
|
|
|
# Проверка приостановки
|
|
if instance_data.get("isSuspended"):
|
|
return InstanceState.SUSPENDED
|
|
|
|
# Проверка основного статуса
|
|
if instance_data.get("explainedStatus") == "running":
|
|
return InstanceState.RUNNING
|
|
|
|
# Проверка ошибок в операциях
|
|
operations = instance_data.get("operations", [])
|
|
if operations:
|
|
last_op = operations[-1]
|
|
if last_op.get("isSuccessful") == False:
|
|
return InstanceState.OPERATION_FAILED
|
|
|
|
return InstanceState.UNKNOWN
|
|
|
|
def check_instance(self, resource_name: str, service_type: Optional[str] = None) -> Optional[InstanceInfo]:
|
|
"""
|
|
Проверить статус инстанса
|
|
|
|
Args:
|
|
resource_name: Имя ресурса в манифесте
|
|
service_type: Тип сервиса (опционально)
|
|
|
|
Returns:
|
|
InstanceInfo с информацией о статусе
|
|
"""
|
|
# Поиск инстанса
|
|
instance_short = self.find_instance(resource_name, service_type)
|
|
|
|
if not instance_short:
|
|
return InstanceInfo(
|
|
uid="unknown",
|
|
name=resource_name,
|
|
state=InstanceState.NOT_CREATED,
|
|
is_created=False,
|
|
is_deleted=False,
|
|
is_suspended=False,
|
|
uptime_seconds=0,
|
|
operation_in_progress=False,
|
|
operation_pending=False,
|
|
last_operation=None,
|
|
last_operation_status=None,
|
|
error_log=None,
|
|
timestamp=datetime.now().isoformat()
|
|
)
|
|
|
|
# Получить полную информацию
|
|
instance_uid = instance_short.get("instanceUid")
|
|
instance_full = self.get_instance_details(instance_uid)
|
|
|
|
if not instance_full:
|
|
instance_full = instance_short
|
|
|
|
# Определить состояние
|
|
state = self.determine_state(instance_full)
|
|
|
|
# Извлечь информацию об операции
|
|
operations = instance_full.get("operations", [])
|
|
last_operation = None
|
|
last_operation_status = None
|
|
error_log = None
|
|
|
|
if operations:
|
|
last_op = operations[-1]
|
|
last_operation = last_op.get("operation")
|
|
last_operation_status = "success" if last_op.get("isSuccessful") == True else "pending" if last_op.get("isSuccessful") is None else "failed"
|
|
error_log = last_op.get("errorLog")
|
|
|
|
return InstanceInfo(
|
|
uid=instance_uid,
|
|
name=resource_name,
|
|
state=state,
|
|
is_created=instance_full.get("isCreated", False),
|
|
is_deleted=instance_full.get("isDeleted", False),
|
|
is_suspended=instance_full.get("isSuspended", False),
|
|
uptime_seconds=instance_full.get("uptime", 0),
|
|
operation_in_progress=instance_full.get("operationIsInProgress", False),
|
|
operation_pending=instance_full.get("operationIsPending", False),
|
|
last_operation=last_operation,
|
|
last_operation_status=last_operation_status,
|
|
error_log=error_log,
|
|
timestamp=datetime.now().isoformat()
|
|
)
|
|
|
|
def format_output(self, info: InstanceInfo) -> str:
|
|
"""Форматировать результат для вывода"""
|
|
|
|
# Выбрать эмодзи в зависимости от состояния
|
|
emoji_map = {
|
|
InstanceState.NOT_CREATED: "⚪",
|
|
InstanceState.CREATING: "🔄",
|
|
InstanceState.RUNNING: "✅",
|
|
InstanceState.MODIFYING: "🔧",
|
|
InstanceState.SUSPENDING: "⏸️",
|
|
InstanceState.SUSPENDED: "⏹️",
|
|
InstanceState.RESUMING: "▶️",
|
|
InstanceState.DELETING: "🗑️",
|
|
InstanceState.DELETED: "❌",
|
|
InstanceState.OPERATION_PENDING: "⏳",
|
|
InstanceState.OPERATION_FAILED: "💥",
|
|
InstanceState.UNKNOWN: "❓",
|
|
}
|
|
|
|
emoji = emoji_map.get(info.state, "?")
|
|
|
|
result = f"{emoji} {info.name}: {info.state.value.upper()}"
|
|
|
|
if info.uid != "unknown":
|
|
result += f" (UID: {info.uid[:8]}...)"
|
|
|
|
if info.last_operation:
|
|
result += f" | Last op: {info.last_operation} ({info.last_operation_status})"
|
|
|
|
if info.uptime_seconds > 0:
|
|
uptime_hours = info.uptime_seconds / 3600
|
|
result += f" | Uptime: {uptime_hours:.1f}h"
|
|
|
|
if info.error_log:
|
|
result += f"\n ⚠️ Error: {info.error_log[:100]}"
|
|
|
|
return result
|
|
|
|
|
|
def main():
|
|
"""Главная функция"""
|
|
|
|
# Пример использования
|
|
token_file = "/home/naeel/remote_dev/terraform/secrets/prod.token"
|
|
|
|
try:
|
|
with open(token_file, 'r') as f:
|
|
token = f.read().strip()
|
|
except FileNotFoundError:
|
|
print(f"❌ Токен не найден: {token_file}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
checker = CloudInstanceChecker(token)
|
|
|
|
# Примеры ресурсов для проверки из PROD_STAND/PG1/resources.tf
|
|
resources_to_check = [
|
|
("pg-testing", "nubes_postgres"),
|
|
("lucy11", "nubes_lucee"),
|
|
("node_0111", "nubes_nodejs"),
|
|
("baba-buck-test2", "nubes_s3bucket"),
|
|
]
|
|
|
|
print("=" * 80)
|
|
print("🔍 Cloud Instance Status Report")
|
|
print("=" * 80)
|
|
print()
|
|
|
|
for resource_name, service_type in resources_to_check:
|
|
print(f"Checking: {resource_name}")
|
|
info = checker.check_instance(resource_name, service_type)
|
|
if info:
|
|
print(checker.format_output(info))
|
|
print()
|
|
|
|
print("=" * 80)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|