mirror of
https://github.com/taylanbakircioglu/flowfish.git
synced 2026-09-20 17:43:24 +00:00
taylanbakircioglu
d7ca50b387
Multi-cluster dependency mapping, real-time network monitoring, impact analysis, and CI/CD integration capabilities. Made-with: Cursor
215 lines
8.3 KiB
Python
215 lines
8.3 KiB
Python
"""Database operations for Cluster Manager"""
|
|
|
|
import logging
|
|
from typing import List, Optional, Dict, Any
|
|
from datetime import datetime
|
|
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
|
|
from sqlalchemy import select, update, delete
|
|
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
|
|
from sqlalchemy import Integer, String, Boolean, DateTime, Text
|
|
|
|
from app.config import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# SQLAlchemy Base
|
|
class Base(DeclarativeBase):
|
|
pass
|
|
|
|
|
|
# Cluster Model
|
|
class Cluster(Base):
|
|
__tablename__ = "clusters"
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True)
|
|
name: Mapped[str] = mapped_column(String(100), unique=True, nullable=False)
|
|
description: Mapped[Optional[str]] = mapped_column(Text)
|
|
cluster_type: Mapped[str] = mapped_column(String(20), nullable=False) # kubernetes, openshift
|
|
api_url: Mapped[str] = mapped_column(String(500), nullable=False)
|
|
kubeconfig: Mapped[Optional[str]] = mapped_column(Text) # Base64 encoded
|
|
service_account_token: Mapped[Optional[str]] = mapped_column(Text) # Base64 encoded
|
|
inspektor_gadget_grpc_endpoint: Mapped[Optional[str]] = mapped_column(String(500))
|
|
inspektor_gadget_token: Mapped[Optional[str]] = mapped_column(String(500))
|
|
ssl_verify: Mapped[bool] = mapped_column(Boolean, default=True)
|
|
is_default: Mapped[bool] = mapped_column(Boolean, default=False)
|
|
health_status: Mapped[str] = mapped_column(String(20), default="unknown")
|
|
node_count: Mapped[int] = mapped_column(Integer, default=0)
|
|
pod_count: Mapped[int] = mapped_column(Integer, default=0)
|
|
namespace_count: Mapped[int] = mapped_column(Integer, default=0)
|
|
last_sync_at: Mapped[Optional[datetime]] = mapped_column(DateTime)
|
|
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
|
|
updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
|
|
|
|
|
|
# Database Manager
|
|
class DatabaseManager:
|
|
"""Async database manager"""
|
|
|
|
def __init__(self):
|
|
# Convert postgresql:// to postgresql+asyncpg:// for async driver
|
|
db_url = settings.DATABASE_URL
|
|
if db_url.startswith("postgresql://"):
|
|
db_url = db_url.replace("postgresql://", "postgresql+asyncpg://", 1)
|
|
|
|
self.engine = create_async_engine(
|
|
db_url,
|
|
echo=False,
|
|
pool_pre_ping=True,
|
|
)
|
|
self.async_session = async_sessionmaker(
|
|
self.engine,
|
|
class_=AsyncSession,
|
|
expire_on_commit=False,
|
|
)
|
|
logger.info(f"Database manager initialized")
|
|
|
|
async def create_tables(self):
|
|
"""Create tables if they don't exist"""
|
|
async with self.engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
logger.info("Database tables created")
|
|
|
|
async def get_cluster(self, cluster_id: int) -> Optional[Cluster]:
|
|
"""Get cluster by ID"""
|
|
async with self.async_session() as session:
|
|
result = await session.execute(
|
|
select(Cluster).where(Cluster.id == cluster_id)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def list_clusters(self) -> List[Cluster]:
|
|
"""List all clusters"""
|
|
async with self.async_session() as session:
|
|
result = await session.execute(select(Cluster))
|
|
return list(result.scalars().all())
|
|
|
|
async def create_cluster(self, cluster_data: Dict[str, Any]) -> Cluster:
|
|
"""Create new cluster"""
|
|
async with self.async_session() as session:
|
|
cluster = Cluster(**cluster_data)
|
|
session.add(cluster)
|
|
await session.commit()
|
|
await session.refresh(cluster)
|
|
logger.info(f"Created cluster: {cluster.name}")
|
|
return cluster
|
|
|
|
async def update_cluster(self, cluster_id: int, cluster_data: Dict[str, Any]) -> Optional[Cluster]:
|
|
"""Update cluster"""
|
|
async with self.async_session() as session:
|
|
cluster = await session.get(Cluster, cluster_id)
|
|
if not cluster:
|
|
return None
|
|
|
|
for key, value in cluster_data.items():
|
|
setattr(cluster, key, value)
|
|
|
|
cluster.updated_at = datetime.utcnow()
|
|
await session.commit()
|
|
await session.refresh(cluster)
|
|
logger.info(f"Updated cluster: {cluster.name}")
|
|
return cluster
|
|
|
|
async def delete_cluster(self, cluster_id: int) -> bool:
|
|
"""Delete cluster"""
|
|
async with self.async_session() as session:
|
|
cluster = await session.get(Cluster, cluster_id)
|
|
if not cluster:
|
|
return False
|
|
|
|
await session.delete(cluster)
|
|
await session.commit()
|
|
logger.info(f"Deleted cluster: {cluster.name}")
|
|
return True
|
|
|
|
async def update_cluster_health(
|
|
self,
|
|
cluster_id: int,
|
|
health_status: str,
|
|
node_count: int,
|
|
pod_count: int,
|
|
namespace_count: int
|
|
) -> bool:
|
|
"""Update cluster health status"""
|
|
async with self.async_session() as session:
|
|
cluster = await session.get(Cluster, cluster_id)
|
|
if not cluster:
|
|
return False
|
|
|
|
cluster.health_status = health_status
|
|
cluster.node_count = node_count
|
|
cluster.pod_count = pod_count
|
|
cluster.namespace_count = namespace_count
|
|
cluster.last_sync_at = datetime.utcnow()
|
|
|
|
await session.commit()
|
|
return True
|
|
|
|
async def close(self):
|
|
"""Close database connection"""
|
|
await self.engine.dispose()
|
|
logger.info("Database connection closed")
|
|
|
|
async def get_cluster_credentials(self, cluster_id: int) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Get cluster credentials from backend's clusters table.
|
|
|
|
This method accesses the backend's clusters table (not cluster-manager's)
|
|
to retrieve encrypted credentials for remote cluster connections.
|
|
|
|
The gateway architecture requires cluster-manager to:
|
|
1. Receive cluster_id from gRPC request
|
|
2. Fetch encrypted credentials from database
|
|
3. Decrypt and create appropriate K8s client
|
|
|
|
Returns:
|
|
Dict with cluster config including encrypted credentials, or None if not found
|
|
"""
|
|
# Use raw SQL to access backend's clusters table with correct column names
|
|
# Note: Live database uses api_server_url (not api_url) and status='active' (not is_active)
|
|
# This is due to migration job schema differing from postgresql-schema.sql
|
|
query = """
|
|
SELECT
|
|
id,
|
|
name,
|
|
connection_type,
|
|
api_server_url,
|
|
token_encrypted,
|
|
ca_cert_encrypted,
|
|
kubeconfig_encrypted,
|
|
skip_tls_verify,
|
|
gadget_namespace,
|
|
status
|
|
FROM clusters
|
|
WHERE id = :cluster_id AND status = 'active'
|
|
"""
|
|
|
|
try:
|
|
async with self.engine.connect() as conn:
|
|
from sqlalchemy import text
|
|
result = await conn.execute(text(query), {"cluster_id": cluster_id})
|
|
row = result.fetchone()
|
|
|
|
if row:
|
|
return {
|
|
"id": row[0],
|
|
"name": row[1],
|
|
"connection_type": row[2],
|
|
"api_server_url": row[3], # Key name kept for compatibility with grpc_server.py
|
|
"token_encrypted": row[4],
|
|
"ca_cert_encrypted": row[5],
|
|
"kubeconfig_encrypted": row[6],
|
|
"skip_tls_verify": row[7],
|
|
"gadget_namespace": row[8],
|
|
"status": row[9] # Live DB uses status='active' not is_active boolean
|
|
}
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"Failed to get cluster credentials for cluster {cluster_id}: {e}")
|
|
return None
|
|
|
|
|
|
# Global database instance
|
|
db_manager = DatabaseManager()
|
|
|