From c0034b195ade36eea935364e0c2d060d960dd12d Mon Sep 17 00:00:00 2001 From: nutcas3 Date: Fri, 24 Apr 2026 08:58:53 +0300 Subject: [PATCH] feat: Refactor api-server main.go to extract route setup and handlers into separate packages for improved modularity - Move route registration logic to registerV1Routes with dependency injection pattern - Extract metrics collector initialization to buildMetricsCollector helper function - Extract service/handler dependency initialization to buildDeps helper function - Remove all inline handler implementations (auth/services/monitoring/deployments/plugins/automation/config/chaos/billing routes) - Remove all metrics collection helper functions (startMetricsCollection/ --- apps/api-server/cmd/server.go | 578 +------------------------- apps/api-server/cmd/server_deps.go | 98 +++++ apps/api-server/cmd/server_metrics.go | 196 +++++++++ apps/api-server/cmd/server_routes.go | 161 +++++++ 4 files changed, 469 insertions(+), 564 deletions(-) create mode 100644 apps/api-server/cmd/server_deps.go create mode 100644 apps/api-server/cmd/server_metrics.go create mode 100644 apps/api-server/cmd/server_routes.go diff --git a/apps/api-server/cmd/server.go b/apps/api-server/cmd/server.go index 1ce67ea..7ae7266 100644 --- a/apps/api-server/cmd/server.go +++ b/apps/api-server/cmd/server.go @@ -3,381 +3,77 @@ package main import ( "context" "log" - "math" "net/http" "os" "os/signal" - "strconv" - "strings" "syscall" "time" "github.com/gin-gonic/gin" + swaggerFiles "github.com/swaggo/files" + ginSwagger "github.com/swaggo/gin-swagger" + "github.com/nutcas3/telecom-platform/apps/api-server/internal/config" "github.com/nutcas3/telecom-platform/apps/api-server/internal/database" "github.com/nutcas3/telecom-platform/apps/api-server/internal/graphql" - "github.com/nutcas3/telecom-platform/apps/api-server/internal/handlers" - "github.com/nutcas3/telecom-platform/apps/api-server/internal/metrics" "github.com/nutcas3/telecom-platform/apps/api-server/internal/middleware" - "github.com/nutcas3/telecom-platform/apps/api-server/internal/models" "github.com/nutcas3/telecom-platform/apps/api-server/internal/monitoring" - "github.com/nutcas3/telecom-platform/apps/api-server/internal/rbac" - "github.com/nutcas3/telecom-platform/apps/api-server/internal/services" "github.com/nutcas3/telecom-platform/apps/api-server/internal/websocket" - swaggerFiles "github.com/swaggo/files" - ginSwagger "github.com/swaggo/gin-swagger" ) func main() { - // Load configuration cfg := config.LoadConfig() - // Initialize database db, err := database.NewDatabase(&cfg.Database) if err != nil { log.Fatalf("Failed to initialize database: %v", err) } defer db.Close() - // Initialize metrics collector - prometheusCollector := metrics.NewMetricsCollector() - prometheusCollector.Register() - - // Initialize time-series storage (optional) - var timeSeriesStorage *metrics.TimeSeriesStorage - if cfg.InfluxDB.URL != "" { - timeSeriesStorage, err = metrics.NewTimeSeriesStorage( - cfg.InfluxDB.URL, - cfg.InfluxDB.Token, - cfg.InfluxDB.Org, - cfg.InfluxDB.Bucket, - ) - if err != nil { - log.Printf("Failed to initialize time-series storage: %v", err) + prometheusCollector, metricsCollector, timeSeriesStorage := buildMetricsCollector(cfg) + defer func() { + if timeSeriesStorage != nil { + timeSeriesStorage.Close() } - defer func() { - if timeSeriesStorage != nil { - timeSeriesStorage.Close() - } - }() - } + }() - // Create metrics collector with time-series - var metricsCollector *metrics.MetricsCollectorWithTimeSeries - if timeSeriesStorage != nil { - metricsCollector = metrics.NewMetricsCollectorWithTimeSeries(prometheusCollector, timeSeriesStorage) - } else { - metricsCollector = &metrics.MetricsCollectorWithTimeSeries{ - MetricsCollector: prometheusCollector, - } - } - - // Initialize GraphQL resolver resolver := graphql.NewResolver(db, cfg) - // Setup HTTP server with GraphQL handler router := gin.Default() graphql.SetupGraphQLHandler(router, resolver) - // Add metrics middleware router.Use(func(c *gin.Context) { - // Create a wrapper handler that calls the next handler - next := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - c.Next() - }) - - // Apply prometheus middleware - middlewareHandler := prometheusCollector.HTTPMiddleware(next) - middlewareHandler.ServeHTTP(c.Writer, c.Request) + next := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { c.Next() }) + prometheusCollector.HTTPMiddleware(next).ServeHTTP(c.Writer, c.Request) }) - // Add metrics endpoint router.GET("/metrics", gin.WrapH(prometheusCollector.Handler())) - - // Add health check endpoint - router.GET("/health", func(c *gin.Context) { - c.JSON(http.StatusOK, gin.H{ - "status": "healthy", - "service": "api-server", - }) - }) - - // Add Swagger documentation endpoints router.GET("/swagger/*any", ginSwagger.WrapHandler(swaggerFiles.Handler)) - - // Add WebSocket endpoint router.GET("/ws", websocket.HandleWebSocket(websocket.GetHub())) - // Add health check endpoints router.GET("/health", monitoring.HealthHandler()) router.GET("/ready", monitoring.ReadyHandler()) router.GET("/live", monitoring.LiveHandler()) - // Apply global rate limiting router.Use(middleware.RateLimitByEndpoint()) - - // Apply performance optimization middleware router.Use(middleware.PerformanceMiddlewareHandler()) router.Use(middleware.SecurityHeadersMiddleware()) router.Use(middleware.CORSMiddleware()) - router.Use(middleware.RequestSizeMiddleware(10 << 20)) // 10MB limit + router.Use(middleware.RequestSizeMiddleware(10 << 20)) router.Use(middleware.TimeoutMiddleware(30 * time.Second)) - - // Add performance metrics endpoint router.GET("/metrics/performance", middleware.PerformanceHandler()) - // Initialize authentication service - authSvc := services.NewAuthService(db.DB, cfg.Auth.JWTSecret) - - // Initialize WebSocket hub websocket.InitializeWebSocket() - - // Initialize health monitor monitoring.InitializeHealthMonitor("1.0.0", "production") - - // Register health checks healthMonitor := monitoring.GetHealthMonitor() healthMonitor.RegisterCheck("database", monitoring.NewDatabaseHealthChecker(db.DB)) - - // Register system health check healthMonitor.RegisterCheck("system", monitoring.NewSystemHealthChecker()) - - // Initialize alert manager monitoring.InitializeAlertManager(healthMonitor) + middleware.InitializePerformanceMiddleware(100 * time.Millisecond) - // Initialize performance middleware - middleware.InitializePerformanceMiddleware(100 * time.Millisecond) // 100ms threshold + deps := buildDeps(db, cfg) + registerV1Routes(router, deps) - // Initialize Casbin RBAC service - casbinSvc, err := rbac.NewCasbinService(db.DB) - if err != nil { - log.Printf("Failed to initialize Casbin service: %v", err) - // Continue without Casbin - fallback to basic role checking - casbinSvc = nil - } - - // Initialize real, DB/K8s/Prometheus-backed services. - chaosSvc := services.NewChaosService(db) - invoiceSvc := services.NewInvoiceService(db) - pluginSvc := services.NewPluginService(db) - automationSvc := services.NewAutomationService(db) - configStoreSvc := services.NewConfigStoreService(db) - deploymentSvc := services.NewDeploymentService(db) - promSvc := monitoring.NewPrometheusService() - - // Kubernetes service is optional: if not configured we surface a clear - // 503 from service management endpoints rather than returning fake data. - k8sSvc, k8sErr := services.NewKubernetesService() - if k8sErr != nil { - log.Printf("Kubernetes integration disabled: %v", k8sErr) - } - - // Build handlers wired to real services. - authH := handlers.NewAuthHandler(authSvc) - servicesH := handlers.NewServicesHandler(k8sSvc) - monitoringH := handlers.NewMonitoringHandler(promSvc) - deploymentsH := handlers.NewDeploymentsHandler(deploymentSvc) - pluginsH := handlers.NewPluginsHandler(pluginSvc) - automationH := handlers.NewAutomationHandler(automationSvc) - configH := handlers.NewConfigHandler(configStoreSvc) - chaosH := handlers.NewChaosHandler(chaosSvc) - billingH := handlers.NewBillingHandler(invoiceSvc, db.DB) - - // API v1 routes - v1 := router.Group("/v1") - { - // Authentication routes (no auth required) - auth := v1.Group("/auth") - { - auth.POST("/login", authH.Login) - auth.POST("/register", authH.Register) - auth.POST("/refresh", authH.RefreshToken) - } - - // Protected routes (authentication required) - protected := v1.Group("/") - protected.Use(middleware.AuthMiddleware(authSvc)) - { - // User management - authProtected := protected.Group("/auth") - { - authProtected.POST("/logout", authH.Logout) - authProtected.GET("/profile", authH.GetProfile) - authProtected.POST("/change-password", authH.ChangePassword) - } - - // Admin-only user management - users := protected.Group("/users") - if casbinSvc != nil { - users.Use(middleware.RequireCasbinPermission(casbinSvc, "/v1/users", "GET")) - } else { - users.Use(middleware.RequireRole("admin")) - } - { - users.GET("", authH.GetUsers) - users.POST("", authH.CreateUser) - users.PUT("/:id", authH.UpdateUser) - users.DELETE("/:id", authH.DeleteUser) - } - } - - // Protected API endpoints (authentication required) - apiProtected := v1.Group("/") - apiProtected.Use(middleware.AuthMiddleware(authSvc)) - { - // Services Management - svcs := apiProtected.Group("/services") - { - svcs.GET("", servicesH.List) - svcs.GET("/:id", servicesH.Get) - svcs.GET("/:id/logs", servicesH.Logs) - svcs.GET("/:id/health", servicesH.Health) - svcs.GET("/:id/pods", servicesH.PodStatus) - svcs.GET("/:id/events", servicesH.Events) - - // Write operations require higher permissions - svcsWrite := svcs.Group("/") - if casbinSvc != nil { - svcsWrite.Use(middleware.RequireCasbinPermission(casbinSvc, "/v1/services", "POST")) - } else { - svcsWrite.Use(middleware.RequireRole("admin", "operator")) - } - { - svcsWrite.POST("/:id/restart", servicesH.Restart) - svcsWrite.POST("/:id/start", servicesH.Start) - svcsWrite.POST("/:id/stop", servicesH.Stop) - svcsWrite.POST("/:id/scale", servicesH.Scale) - } - } - - // Monitoring & Metrics (read-only for most roles) - mon := apiProtected.Group("/monitoring") - { - mon.GET("/metrics", monitoringH.Metrics) - mon.GET("/alerts", monitoringH.Alerts) - mon.GET("/health", monitoringH.Health) - mon.GET("/logs", monitoringH.Logs) - } - - // Deployment Management - deploy := apiProtected.Group("/deploy") - { - deploy.GET("/status", deploymentsH.Status) - deploy.GET("/history", deploymentsH.History) - - // Write operations require higher permissions - deployWrite := deploy.Group("/") - if casbinSvc != nil { - deployWrite.Use(middleware.RequireCasbinPermission(casbinSvc, "/v1/deploy", "POST")) - } else { - deployWrite.Use(middleware.RequireRole("admin", "operator")) - } - { - deployWrite.POST("/start", deploymentsH.Start) - deployWrite.POST("/rollback", deploymentsH.Rollback) - } - } - - // Plugin Management - plugs := apiProtected.Group("/plugins") - { - plugs.GET("", pluginsH.List) - - // Write operations require admin permissions - plugsWrite := plugs.Group("/") - if casbinSvc != nil { - plugsWrite.Use(middleware.RequireCasbinPermission(casbinSvc, "/v1/plugins", "POST")) - plugsDelete := plugs.Group("/") - plugsDelete.Use(middleware.RequireCasbinPermission(casbinSvc, "/v1/plugins", "DELETE")) - plugsDelete.DELETE("/:id", pluginsH.Uninstall) - } else { - plugsWrite.Use(middleware.RequireRole("admin")) - plugsWrite.DELETE("/:id", pluginsH.Uninstall) - } - { - plugsWrite.POST("/install", pluginsH.Install) - plugsWrite.POST("/:id/enable", pluginsH.Enable) - plugsWrite.POST("/:id/disable", pluginsH.Disable) - } - } - - // Automation Management - auto := apiProtected.Group("/automation") - { - auto.GET("", automationH.List) - auto.GET("/logs", automationH.Logs) - - // Write operations require higher permissions - autoWrite := auto.Group("/") - if casbinSvc != nil { - autoWrite.Use(middleware.RequireCasbinPermission(casbinSvc, "/v1/automation", "POST")) - } else { - autoWrite.Use(middleware.RequireRole("admin", "operator")) - } - { - autoWrite.POST("", automationH.Create) - autoWrite.POST("/:id/run", automationH.Run) - autoWrite.POST("/:id/schedule", automationH.Schedule) - } - } - - // Billing & Invoicing - bill := apiProtected.Group("/billing") - { - bill.GET("/invoices", billingH.ListInvoices) - bill.GET("/payments", billingH.ListPayments) - - // Invoice generation requires admin permissions - billWrite := bill.Group("/") - if casbinSvc != nil { - billWrite.Use(middleware.RequireCasbinPermission(casbinSvc, "/v1/billing", "POST")) - } else { - billWrite.Use(middleware.RequireRole("admin")) - } - { - billWrite.POST("/invoices", billingH.GenerateInvoice) - } - } - - // Configuration Management - cfg := apiProtected.Group("/config") - { - cfg.GET("", configH.Get) - cfg.GET("/validate", configH.Validate) - - // Configuration updates require admin permissions - cfgWrite := cfg.Group("/") - if casbinSvc != nil { - cfgWrite.Use(middleware.RequireCasbinPermission(casbinSvc, "/v1/config", "POST")) - } else { - cfgWrite.Use(middleware.RequireRole("admin")) - } - { - cfgWrite.POST("", configH.Update) - } - } - - // Chaos Engineering - chaosGroup := apiProtected.Group("/chaos") - { - chaosGroup.GET("/experiments", chaosH.List) - chaosGroup.GET("/status", chaosH.Status) - - // Running chaos experiments requires admin permissions - chaosWrite := chaosGroup.Group("/") - if casbinSvc != nil { - chaosWrite.Use(middleware.RequireCasbinPermission(casbinSvc, "/v1/chaos", "POST")) - } else { - chaosWrite.Use(middleware.RequireRole("admin")) - } - { - chaosWrite.POST("/experiments", chaosH.Run) - } - } - } - } - - // Create server server := &http.Server{ Addr: ":" + cfg.Server.Port, Handler: router, @@ -385,7 +81,6 @@ func main() { WriteTimeout: 30 * time.Second, } - // Start metrics server in background ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -400,13 +95,9 @@ func main() { } }() - // Start periodic metrics collection go startMetricsCollection(ctx, metricsCollector, db) - - // Start alert manager go monitoring.StartAlertManager(ctx) - // Start server in a goroutine go func() { log.Printf("Starting API server on port %s", cfg.Server.Port) if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed { @@ -414,14 +105,11 @@ func main() { } }() - // Wait for interrupt signal to gracefully shutdown the server quit := make(chan os.Signal, 1) signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM) <-quit log.Println("Shutting down server...") - - // Shutdown server with timeout shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 30*time.Second) defer shutdownCancel() @@ -431,241 +119,3 @@ func main() { log.Println("Server exited") } - -// startMetricsCollection periodically collects and stores metrics -func startMetricsCollection(ctx context.Context, mc *metrics.MetricsCollectorWithTimeSeries, db *database.Database) { - ticker := time.NewTicker(30 * time.Second) - defer ticker.Stop() - - for { - select { - case <-ctx.Done(): - return - case <-ticker.C: - collectMetrics(ctx, mc, db) - } - } -} - -// collectMetrics gathers metrics from various sources -func collectMetrics(ctx context.Context, mc *metrics.MetricsCollectorWithTimeSeries, db *database.Database) { - // Collect subscriber metrics - totalSubs, activeSubs, suspendedSubs := getSubscriberStats(db) - mc.UpdateSubscriberMetrics(totalSubs, activeSubs, suspendedSubs) - - // Collect system metrics - systemMetrics := metrics.SystemMetrics{ - ActiveSessions: getActiveSessionsCount(db), - CPUUsage: getCPUUsage(), - MemoryUsage: getMemoryUsage(), - NetworkRX: getNetworkRX(), - NetworkTX: getNetworkTX(), - Timestamp: time.Now(), - } - - // Store in time-series if available - if ts := mc.GetTimeSeriesStorage(); ts != nil { - if err := ts.StoreSystemMetrics(ctx, systemMetrics); err != nil { - log.Printf("Failed to store system metrics: %v", err) - } - } -} - -// Helper functions for metrics collection (actual implementations) -func getSubscriberStats(db *database.Database) (total, active, suspended int) { - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - - // Count total subscribers using GORM - var totalCount int64 - if err := db.DB.WithContext(ctx).Model(&models.Subscriber{}).Count(&totalCount).Error; err != nil { - log.Printf("Error counting total subscribers: %v", err) - return 0, 0, 0 - } - - // Count active subscribers - var activeCount int64 - if err := db.DB.WithContext(ctx).Model(&models.Subscriber{}).Where("status = ?", "active").Count(&activeCount).Error; err != nil { - log.Printf("Error counting active subscribers: %v", err) - return int(totalCount), 0, 0 - } - - // Count suspended subscribers - var suspendedCount int64 - if err := db.DB.WithContext(ctx).Model(&models.Subscriber{}).Where("status IN ?", []string{"suspended", "deactivated", "blocked"}).Count(&suspendedCount).Error; err != nil { - log.Printf("Error counting suspended subscribers: %v", err) - return int(totalCount), int(activeCount), 0 - } - - return int(totalCount), int(activeCount), int(suspendedCount) -} - -func getActiveSessionsCount(db *database.Database) int { - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - - // Count active charging sessions using GORM - var count int64 - if err := db.DB.WithContext(ctx).Model(&models.Session{}).Where( - "status IN ? AND end_time IS NULL", - []string{"active", "connected", "established"}, - ).Count(&count).Error; err != nil { - log.Printf("Error counting active sessions: %v", err) - return 0 - } - - return int(count) -} - -func getCPUUsage() float64 { - // Read CPU usage from /proc/stat on Linux systems - data, err := os.ReadFile("/proc/stat") - if err != nil { - log.Printf("Error reading /proc/stat: %v", err) - return 0 - } - - lines := strings.Split(string(data), "\n") - if len(lines) == 0 { - return 0 - } - - // Parse the first line (total CPU usage) - fields := strings.Fields(lines[0]) - if len(fields) < 8 || fields[0] != "cpu" { - return 0 - } - - // Parse CPU time values (user, nice, system, idle, iowait, irq, softirq, steal) - var totalCPU, idleCPU float64 - for i, field := range fields[1:8] { - val, err := strconv.ParseFloat(field, 64) - if err != nil { - continue - } - totalCPU += val - if i == 3 { // idle time - idleCPU = val - } - } - - if totalCPU == 0 { - return 0 - } - - // Calculate CPU usage percentage (non-idle time) - cpuUsage := ((totalCPU - idleCPU) / totalCPU) * 100 - return math.Min(cpuUsage, 100.0) // Cap at 100% -} - -func getMemoryUsage() float64 { - // Read memory usage from /proc/meminfo on Linux systems - data, err := os.ReadFile("/proc/meminfo") - if err != nil { - log.Printf("Error reading /proc/meminfo: %v", err) - return 0 - } - - lines := strings.Split(string(data), "\n") - var totalMem, availableMem float64 - - // Parse memory information - for _, line := range lines { - fields := strings.Fields(line) - if len(fields) < 2 { - continue - } - - switch fields[0] { - case "MemTotal:": - if val, err := strconv.ParseFloat(fields[1], 64); err == nil { - totalMem = val * 1024 // Convert KB to bytes - } - case "MemAvailable:": - if val, err := strconv.ParseFloat(fields[1], 64); err == nil { - availableMem = val * 1024 // Convert KB to bytes - } - } - } - - if totalMem == 0 { - return 0 - } - - // Calculate memory usage percentage - usedMem := totalMem - availableMem - memoryUsage := (usedMem / totalMem) * 100 - return math.Min(memoryUsage, 100.0) // Cap at 100% -} - -func getNetworkRX() int64 { - // Read network statistics from /proc/net/dev on Linux systems - data, err := os.ReadFile("/proc/net/dev") - if err != nil { - log.Printf("Error reading /proc/net/dev: %v", err) - return 0 - } - - lines := strings.Split(string(data), "\n") - var totalRX int64 - - // Skip header lines (first two lines) - for i, line := range lines { - if i < 2 { - continue - } - - fields := strings.Fields(line) - if len(fields) < 10 { - continue - } - - // Skip loopback interface - if strings.HasPrefix(fields[0], "lo:") { - continue - } - - // RX bytes are in field 1 (after interface name) - if rxBytes, err := strconv.ParseInt(fields[1], 10, 64); err == nil { - totalRX += rxBytes - } - } - - return totalRX -} - -func getNetworkTX() int64 { - // Read network statistics from /proc/net/dev on Linux systems - data, err := os.ReadFile("/proc/net/dev") - if err != nil { - log.Printf("Error reading /proc/net/dev: %v", err) - return 0 - } - - lines := strings.Split(string(data), "\n") - var totalTX int64 - - // Skip header lines (first two lines) - for i, line := range lines { - if i < 2 { - continue - } - - fields := strings.Fields(line) - if len(fields) < 10 { - continue - } - - // Skip loopback interface - if strings.HasPrefix(fields[0], "lo:") { - continue - } - - // TX bytes are in field 9 (after interface name and RX stats) - if txBytes, err := strconv.ParseInt(fields[9], 10, 64); err == nil { - totalTX += txBytes - } - } - - return totalTX -} diff --git a/apps/api-server/cmd/server_deps.go b/apps/api-server/cmd/server_deps.go new file mode 100644 index 0000000..0adacc6 --- /dev/null +++ b/apps/api-server/cmd/server_deps.go @@ -0,0 +1,98 @@ +package main + +import ( + "log" + + "github.com/nutcas3/telecom-platform/apps/api-server/internal/config" + "github.com/nutcas3/telecom-platform/apps/api-server/internal/database" + "github.com/nutcas3/telecom-platform/apps/api-server/internal/handlers" + "github.com/nutcas3/telecom-platform/apps/api-server/internal/metrics" + "github.com/nutcas3/telecom-platform/apps/api-server/internal/monitoring" + "github.com/nutcas3/telecom-platform/apps/api-server/internal/rbac" + "github.com/nutcas3/telecom-platform/apps/api-server/internal/services" +) + +// serverDeps bundles the services and handlers needed to register routes. +type serverDeps struct { + authSvc *services.AuthService + casbinSvc *rbac.CasbinService + + authH *handlers.AuthHandler + servicesH *handlers.ServicesHandler + monitoringH *handlers.MonitoringHandler + deploymentsH *handlers.DeploymentsHandler + pluginsH *handlers.PluginsHandler + automationH *handlers.AutomationHandler + configH *handlers.ConfigHandler + chaosH *handlers.ChaosHandler + billingH *handlers.BillingHandler +} + +// buildDeps constructs all downstream services and handlers. +func buildDeps(db *database.Database, cfg *config.Config) *serverDeps { + authSvc := services.NewAuthService(db.DB, cfg.Auth.JWTSecret) + + casbinSvc, err := rbac.NewCasbinService(db.DB) + if err != nil { + log.Printf("Failed to initialize Casbin service: %v", err) + casbinSvc = nil + } + + chaosSvc := services.NewChaosService(db) + invoiceSvc := services.NewInvoiceService(db) + pluginSvc := services.NewPluginService(db) + automationSvc := services.NewAutomationService(db) + configStoreSvc := services.NewConfigStoreService(db) + deploymentSvc := services.NewDeploymentService(db) + promSvc := monitoring.NewPrometheusService() + + k8sSvc, k8sErr := services.NewKubernetesService() + if k8sErr != nil { + log.Printf("Kubernetes integration disabled: %v", k8sErr) + } + + return &serverDeps{ + authSvc: authSvc, + casbinSvc: casbinSvc, + authH: handlers.NewAuthHandler(authSvc), + servicesH: handlers.NewServicesHandler(k8sSvc), + monitoringH: handlers.NewMonitoringHandler(promSvc), + deploymentsH: handlers.NewDeploymentsHandler(deploymentSvc), + pluginsH: handlers.NewPluginsHandler(pluginSvc), + automationH: handlers.NewAutomationHandler(automationSvc), + configH: handlers.NewConfigHandler(configStoreSvc), + chaosH: handlers.NewChaosHandler(chaosSvc), + billingH: handlers.NewBillingHandler(invoiceSvc, db.DB), + } +} + +// buildMetricsCollector constructs the Prometheus collector, optionally wiring +// the InfluxDB time-series storage when configured. +func buildMetricsCollector(cfg *config.Config) (*metrics.MetricsCollector, *metrics.MetricsCollectorWithTimeSeries, *metrics.TimeSeriesStorage) { + prometheusCollector := metrics.NewMetricsCollector() + prometheusCollector.Register() + + var timeSeriesStorage *metrics.TimeSeriesStorage + if cfg.InfluxDB.URL != "" { + ts, err := metrics.NewTimeSeriesStorage( + cfg.InfluxDB.URL, + cfg.InfluxDB.Token, + cfg.InfluxDB.Org, + cfg.InfluxDB.Bucket, + ) + if err != nil { + log.Printf("Failed to initialize time-series storage: %v", err) + } else { + timeSeriesStorage = ts + } + } + + var withTS *metrics.MetricsCollectorWithTimeSeries + if timeSeriesStorage != nil { + withTS = metrics.NewMetricsCollectorWithTimeSeries(prometheusCollector, timeSeriesStorage) + } else { + withTS = &metrics.MetricsCollectorWithTimeSeries{MetricsCollector: prometheusCollector} + } + + return prometheusCollector, withTS, timeSeriesStorage +} diff --git a/apps/api-server/cmd/server_metrics.go b/apps/api-server/cmd/server_metrics.go new file mode 100644 index 0000000..8f0882c --- /dev/null +++ b/apps/api-server/cmd/server_metrics.go @@ -0,0 +1,196 @@ +package main + +import ( + "context" + "log" + "math" + "os" + "strconv" + "strings" + "time" + + "github.com/nutcas3/telecom-platform/apps/api-server/internal/database" + "github.com/nutcas3/telecom-platform/apps/api-server/internal/metrics" + "github.com/nutcas3/telecom-platform/apps/api-server/internal/models" +) + +// startMetricsCollection periodically collects and stores metrics. +func startMetricsCollection(ctx context.Context, mc *metrics.MetricsCollectorWithTimeSeries, db *database.Database) { + ticker := time.NewTicker(30 * time.Second) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + collectMetrics(ctx, mc, db) + } + } +} + +// collectMetrics gathers metrics from various sources. +func collectMetrics(ctx context.Context, mc *metrics.MetricsCollectorWithTimeSeries, db *database.Database) { + totalSubs, activeSubs, suspendedSubs := getSubscriberStats(db) + mc.UpdateSubscriberMetrics(totalSubs, activeSubs, suspendedSubs) + + systemMetrics := metrics.SystemMetrics{ + ActiveSessions: getActiveSessionsCount(db), + CPUUsage: getCPUUsage(), + MemoryUsage: getMemoryUsage(), + NetworkRX: getNetworkRX(), + NetworkTX: getNetworkTX(), + Timestamp: time.Now(), + } + + if ts := mc.GetTimeSeriesStorage(); ts != nil { + if err := ts.StoreSystemMetrics(ctx, systemMetrics); err != nil { + log.Printf("Failed to store system metrics: %v", err) + } + } +} + +// getSubscriberStats counts subscribers by status. +func getSubscriberStats(db *database.Database) (total, active, suspended int) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + var totalCount, activeCount, suspendedCount int64 + if err := db.DB.WithContext(ctx).Model(&models.Subscriber{}).Count(&totalCount).Error; err != nil { + log.Printf("Error counting total subscribers: %v", err) + return 0, 0, 0 + } + if err := db.DB.WithContext(ctx).Model(&models.Subscriber{}).Where("status = ?", "active").Count(&activeCount).Error; err != nil { + log.Printf("Error counting active subscribers: %v", err) + return int(totalCount), 0, 0 + } + if err := db.DB.WithContext(ctx).Model(&models.Subscriber{}). + Where("status IN ?", []string{"suspended", "deactivated", "blocked"}). + Count(&suspendedCount).Error; err != nil { + log.Printf("Error counting suspended subscribers: %v", err) + return int(totalCount), int(activeCount), 0 + } + return int(totalCount), int(activeCount), int(suspendedCount) +} + +// getActiveSessionsCount returns the number of currently active charging sessions. +func getActiveSessionsCount(db *database.Database) int { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + var count int64 + if err := db.DB.WithContext(ctx).Model(&models.Session{}).Where( + "status IN ? AND end_time IS NULL", + []string{"active", "connected", "established"}, + ).Count(&count).Error; err != nil { + log.Printf("Error counting active sessions: %v", err) + return 0 + } + + return int(count) +} + +// getCPUUsage reads /proc/stat and returns the non-idle CPU percentage. +func getCPUUsage() float64 { + data, err := os.ReadFile("/proc/stat") + if err != nil { + log.Printf("Error reading /proc/stat: %v", err) + return 0 + } + + lines := strings.Split(string(data), "\n") + if len(lines) == 0 { + return 0 + } + + fields := strings.Fields(lines[0]) + if len(fields) < 8 || fields[0] != "cpu" { + return 0 + } + + var totalCPU, idleCPU float64 + for i, field := range fields[1:8] { + val, err := strconv.ParseFloat(field, 64) + if err != nil { + continue + } + totalCPU += val + if i == 3 { + idleCPU = val + } + } + + if totalCPU == 0 { + return 0 + } + + return math.Min(((totalCPU-idleCPU)/totalCPU)*100, 100.0) +} + +// getMemoryUsage reads /proc/meminfo and returns used-memory percentage. +func getMemoryUsage() float64 { + data, err := os.ReadFile("/proc/meminfo") + if err != nil { + log.Printf("Error reading /proc/meminfo: %v", err) + return 0 + } + + lines := strings.Split(string(data), "\n") + var totalMem, availableMem float64 + + for _, line := range lines { + fields := strings.Fields(line) + if len(fields) < 2 { + continue + } + switch fields[0] { + case "MemTotal:": + if val, err := strconv.ParseFloat(fields[1], 64); err == nil { + totalMem = val * 1024 + } + case "MemAvailable:": + if val, err := strconv.ParseFloat(fields[1], 64); err == nil { + availableMem = val * 1024 + } + } + } + + if totalMem == 0 { + return 0 + } + + return math.Min(((totalMem-availableMem)/totalMem)*100, 100.0) +} + +// getNetworkRX returns total RX bytes across non-loopback interfaces. +func getNetworkRX() int64 { return readNetworkStat(1) } + +// getNetworkTX returns total TX bytes across non-loopback interfaces. +func getNetworkTX() int64 { return readNetworkStat(9) } + +// readNetworkStat parses /proc/net/dev for a given field index. +func readNetworkStat(fieldIdx int) int64 { + data, err := os.ReadFile("/proc/net/dev") + if err != nil { + log.Printf("Error reading /proc/net/dev: %v", err) + return 0 + } + + lines := strings.Split(string(data), "\n") + var total int64 + for i, line := range lines { + if i < 2 { + continue + } + fields := strings.Fields(line) + if len(fields) < 10 { + continue + } + if strings.HasPrefix(fields[0], "lo:") { + continue + } + if v, err := strconv.ParseInt(fields[fieldIdx], 10, 64); err == nil { + total += v + } + } + return total +} diff --git a/apps/api-server/cmd/server_routes.go b/apps/api-server/cmd/server_routes.go new file mode 100644 index 0000000..66dc4b1 --- /dev/null +++ b/apps/api-server/cmd/server_routes.go @@ -0,0 +1,161 @@ +package main + +import ( + "github.com/gin-gonic/gin" + + "github.com/nutcas3/telecom-platform/apps/api-server/internal/middleware" + "github.com/nutcas3/telecom-platform/apps/api-server/internal/rbac" +) + +// registerV1Routes wires the /v1 API routes to the provided deps. +// RBAC enforcement uses Casbin when configured, falling back to role-based +// middleware otherwise. +func registerV1Routes(router *gin.Engine, d *serverDeps) { + v1 := router.Group("/v1") + + auth := v1.Group("/auth") + { + auth.POST("/login", d.authH.Login) + auth.POST("/register", d.authH.Register) + auth.POST("/refresh", d.authH.RefreshToken) + } + + protected := v1.Group("/") + protected.Use(middleware.AuthMiddleware(d.authSvc)) + { + authProtected := protected.Group("/auth") + { + authProtected.POST("/logout", d.authH.Logout) + authProtected.GET("/profile", d.authH.GetProfile) + authProtected.POST("/change-password", d.authH.ChangePassword) + } + + users := protected.Group("/users") + applyRBAC(users, d.casbinSvc, "/v1/users", "GET", "admin") + { + users.GET("", d.authH.GetUsers) + users.POST("", d.authH.CreateUser) + users.PUT("/:id", d.authH.UpdateUser) + users.DELETE("/:id", d.authH.DeleteUser) + } + } + + apiProtected := v1.Group("/") + apiProtected.Use(middleware.AuthMiddleware(d.authSvc)) + { + registerServicesRoutes(apiProtected, d) + registerMonitoringRoutes(apiProtected, d) + registerDeployRoutes(apiProtected, d) + registerPluginRoutes(apiProtected, d) + registerAutomationRoutes(apiProtected, d) + registerBillingRoutes(apiProtected, d) + registerConfigRoutes(apiProtected, d) + registerChaosRoutes(apiProtected, d) + } +} + +// applyRBAC attaches Casbin or fallback role middleware to a group. +func applyRBAC(g *gin.RouterGroup, casbinSvc *rbac.CasbinService, path, method string, fallbackRoles ...string) { + if casbinSvc != nil { + g.Use(middleware.RequireCasbinPermission(casbinSvc, path, method)) + } else { + g.Use(middleware.RequireRole(fallbackRoles...)) + } +} + +func registerServicesRoutes(api *gin.RouterGroup, d *serverDeps) { + svcs := api.Group("/services") + svcs.GET("", d.servicesH.List) + svcs.GET("/:id", d.servicesH.Get) + svcs.GET("/:id/logs", d.servicesH.Logs) + svcs.GET("/:id/health", d.servicesH.Health) + svcs.GET("/:id/pods", d.servicesH.PodStatus) + svcs.GET("/:id/events", d.servicesH.Events) + + w := svcs.Group("/") + applyRBAC(w, d.casbinSvc, "/v1/services", "POST", "admin", "operator") + w.POST("/:id/restart", d.servicesH.Restart) + w.POST("/:id/start", d.servicesH.Start) + w.POST("/:id/stop", d.servicesH.Stop) + w.POST("/:id/scale", d.servicesH.Scale) +} + +func registerMonitoringRoutes(api *gin.RouterGroup, d *serverDeps) { + mon := api.Group("/monitoring") + mon.GET("/metrics", d.monitoringH.Metrics) + mon.GET("/alerts", d.monitoringH.Alerts) + mon.GET("/health", d.monitoringH.Health) + mon.GET("/logs", d.monitoringH.Logs) +} + +func registerDeployRoutes(api *gin.RouterGroup, d *serverDeps) { + deploy := api.Group("/deploy") + deploy.GET("/status", d.deploymentsH.Status) + deploy.GET("/history", d.deploymentsH.History) + + w := deploy.Group("/") + applyRBAC(w, d.casbinSvc, "/v1/deploy", "POST", "admin", "operator") + w.POST("/start", d.deploymentsH.Start) + w.POST("/rollback", d.deploymentsH.Rollback) +} + +func registerPluginRoutes(api *gin.RouterGroup, d *serverDeps) { + plugs := api.Group("/plugins") + plugs.GET("", d.pluginsH.List) + + w := plugs.Group("/") + applyRBAC(w, d.casbinSvc, "/v1/plugins", "POST", "admin") + w.POST("/install", d.pluginsH.Install) + w.POST("/:id/enable", d.pluginsH.Enable) + w.POST("/:id/disable", d.pluginsH.Disable) + + if d.casbinSvc != nil { + del := plugs.Group("/") + del.Use(middleware.RequireCasbinPermission(d.casbinSvc, "/v1/plugins", "DELETE")) + del.DELETE("/:id", d.pluginsH.Uninstall) + } else { + w.DELETE("/:id", d.pluginsH.Uninstall) + } +} + +func registerAutomationRoutes(api *gin.RouterGroup, d *serverDeps) { + auto := api.Group("/automation") + auto.GET("", d.automationH.List) + auto.GET("/logs", d.automationH.Logs) + + w := auto.Group("/") + applyRBAC(w, d.casbinSvc, "/v1/automation", "POST", "admin", "operator") + w.POST("", d.automationH.Create) + w.POST("/:id/run", d.automationH.Run) + w.POST("/:id/schedule", d.automationH.Schedule) +} + +func registerBillingRoutes(api *gin.RouterGroup, d *serverDeps) { + bill := api.Group("/billing") + bill.GET("/invoices", d.billingH.ListInvoices) + bill.GET("/payments", d.billingH.ListPayments) + + w := bill.Group("/") + applyRBAC(w, d.casbinSvc, "/v1/billing", "POST", "admin") + w.POST("/invoices", d.billingH.GenerateInvoice) +} + +func registerConfigRoutes(api *gin.RouterGroup, d *serverDeps) { + cfg := api.Group("/config") + cfg.GET("", d.configH.Get) + cfg.GET("/validate", d.configH.Validate) + + w := cfg.Group("/") + applyRBAC(w, d.casbinSvc, "/v1/config", "POST", "admin") + w.POST("", d.configH.Update) +} + +func registerChaosRoutes(api *gin.RouterGroup, d *serverDeps) { + chaosGroup := api.Group("/chaos") + chaosGroup.GET("/experiments", d.chaosH.List) + chaosGroup.GET("/status", d.chaosH.Status) + + w := chaosGroup.Group("/") + applyRBAC(w, d.casbinSvc, "/v1/chaos", "POST", "admin") + w.POST("/experiments", d.chaosH.Run) +}