diff --git a/cmd/futriis/main.go b/cmd/futriis/main.go new file mode 100644 index 0000000..eab7d7e --- /dev/null +++ b/cmd/futriis/main.go @@ -0,0 +1,454 @@ +/* + * Copyright 2026 Safronov Grigorii + * + * Licensed under the CDDL, Version 1.0 (the "License"); + * you may not use this file except in compliance with the License. + * + * You may obtain a copy of the License at + * https://opensource.org/licenses/CDDL-1.0 + */ + +package main + +import ( + "context" + "encoding/json" + "fmt" + "os" + "os/signal" + "syscall" + "time" + + "futriis/internal/acl" + "futriis/internal/api" + "futriis/internal/cluster" + "futriis/internal/config" + "futriis/internal/log" + "futriis/internal/plugin" + "futriis/internal/repl" + "futriis/internal/storage" + "futriis/pkg/utils" +) + +func main() { + utils.SetColorEnabled(true) + + logLevel := os.Getenv("LOG_LEVEL") + if logLevel == "" { + logLevel = "info" + } + + if err := log.InitDefaultLogger("futriis/futriis.log", logLevel); err != nil { + fmt.Printf("Failed to initialize logger: %v\n", err) + os.Exit(1) + } + + log.Info("Futriis DB starting...") + log.Infof("Log level: %s", logLevel) + + cfg, err := config.LoadConfig("config.toml") + if err != nil { + log.Error("Failed to load config: " + err.Error()) + utils.PrintError("Failed to load config: " + err.Error()) + os.Exit(1) + } + + logger, err := log.NewLogger(cfg.Log.LogFile, cfg.Log.LogLevel) + if err != nil { + log.Error("Failed to initialize logger: " + err.Error()) + utils.PrintError("Failed to initialize logger: " + err.Error()) + os.Exit(1) + } + defer logger.Close() + logger.Info("futriix database starting...") + + store := storage.NewStorage(cfg.Storage.PageSizeMB, logger) + + storage.SetGlobalStorage(store) + + if err := storage.InitTransactionManager("futriix.wal"); err != nil { + logger.Warn("Failed to initialize transaction manager: " + err.Error()) + } else { + storage.SetTransactionLogger(logger) + logger.Info("Transaction manager initialized") + } + + // Инициализация менеджера триггеров (функция не возвращает ошибку) + storage.InitTriggerManager(logger) + logger.Info("Trigger manager initialized") + + aclManager := acl.NewACLManager() + logger.Info("ACL manager initialized") + + // Исправлено: передаём store как второй аргумент + raftCoordinator, err := cluster.NewRaftCoordinator(cfg, store, logger) + if err != nil { + logger.Error("Failed to start Raft coordinator: " + err.Error()) + utils.PrintError("Failed to start Raft coordinator: " + err.Error()) + os.Exit(1) + } + + // ========== НОВАЯ ФУНКЦИОНАЛЬНОСТЬ: Проверка статуса миграций схемы данных ========== + if raftCoordinator != nil { + schemaMigrator := raftCoordinator.GetSchemaMigrator() + if schemaMigrator != nil { + logger.Info("Checking for schema migrations...") + status := schemaMigrator.GetStatus() + logger.Infof("Current schema version: %s, total migrations: %d, applied: %d, pending: %d", + status.CurrentVersion, status.TotalMigrations, status.AppliedMigrations, status.PendingMigrations) + + // Выводим список миграций + for _, m := range status.Migrations { + if m.Applied { + logger.Debugf(" [APPLIED] %s v%s: %s (at %s)", + m.ID, m.Version, m.Description, time.UnixMilli(m.AppliedAt).Format("2006-01-02 15:04:05")) + } else { + logger.Debugf(" [PENDING] %s v%s: %s", m.ID, m.Version, m.Description) + } + } + + if status.PendingMigrations > 0 { + logger.Info("Applying pending schema migrations...") + if err := schemaMigrator.Migrate("2.1.0"); err != nil { + logger.Warn(fmt.Sprintf("Schema migration warning: %v", err)) + } else { + logger.Info("Schema migrations completed successfully") + } + } + } else { + logger.Warn("Schema migrator not available") + } + + // ========== НОВАЯ ФУНКЦИОНАЛЬНОСТЬ: Проверка статуса Fallback менеджера ========== + fallbackMgr := raftCoordinator.GetFallbackManager() + if fallbackMgr != nil { + logger.Info("Leader fallback manager is active (Single Point of Failure protection enabled)") + fallbackStats := raftCoordinator.GetFallbackStats() + logger.Debugf("Fallback manager stats: %+v", fallbackStats) + } else { + logger.Warn("Leader fallback manager not available") + } + + // ========== НОВАЯ ФУНКЦИОНАЛЬНОСТЬ: Проверка статуса Panic Recovery менеджера ========== + panicRecoveryMgr := raftCoordinator.GetPanicRecoveryManager() + if panicRecoveryMgr != nil { + logger.Info("Panic recovery manager is active (automatic goroutine recovery enabled)") + recoveryStats := raftCoordinator.GetPanicRecoveryStats() + logger.Debugf("Panic recovery stats: %+v", recoveryStats) + } else { + logger.Warn("Panic recovery manager not available") + } + } + + if cfg.Cluster.Bootstrap || len(cfg.Cluster.Nodes) <= 1 { + maxRetriesLocal := 10 + for i := 0; i < maxRetriesLocal; i++ { + if raftCoordinator.IsLeader() { + break + } + time.Sleep(1 * time.Second) + } + } + + node := cluster.NewNode(cfg.Cluster.NodeIP, cfg.Cluster.NodePort, store, logger) + + // Исправлено: объявляем переменную maxRetries здесь, чтобы она была доступна + maxRetries := 5 + var registerErr error + for i := 0; i < maxRetries; i++ { + registerErr = raftCoordinator.RegisterNode(node) + if registerErr == nil { + break + } + if i < maxRetries-1 { + logger.Warn(fmt.Sprintf("Failed to register node (attempt %d/%d): %v, retrying...", i+1, maxRetries, registerErr)) + time.Sleep(2 * time.Second) + } + } + + if registerErr != nil { + logger.Error("Failed to register node: " + registerErr.Error()) + utils.PrintError("Failed to register node: " + registerErr.Error()) + os.Exit(1) + } + + // Исправлено: передаём отдельные параметры из конфигурации плагинов + pluginManager := plugin.NewPluginManager( + cfg.Plugins.ScriptDir, + logger, + store, + cfg.Plugins.Enabled, + ) + + if cfg.Plugins.Enabled { + logger.Info(fmt.Sprintf("Plugin manager initialized, directory: %s", cfg.Plugins.ScriptDir)) + + plugins := pluginManager.ListPlugins() + for _, p := range plugins { + if err := pluginManager.StartPlugin(p.Name); err != nil { + logger.Warn(fmt.Sprintf("Failed to start plugin %s: %v", p.Name, err)) + } + } + } + + httpPort := cfg.API.Port + httpServer := api.NewHTTPServer(httpPort, store, raftCoordinator, aclManager, logger) + go func() { + if err := httpServer.Start(); err != nil { + logger.Error("HTTP server error: " + err.Error()) + utils.PrintError("HTTP server error: " + err.Error()) + } + }() + logger.Info(fmt.Sprintf("HTTP API server started on port %d", httpPort)) + + webUIPort := cfg.WebUI.Port + if webUIPort == 0 { + webUIPort = 8081 + } + webUI := api.NewWebUIServer(webUIPort, cfg.WebUI.Enabled, store, raftCoordinator, aclManager, logger) + go func() { + if err := webUI.Start(); err != nil && cfg.WebUI.Enabled { + logger.Error("Web UI error: " + err.Error()) + utils.PrintError("Web UI error: " + err.Error()) + } + }() + + if cfg.WebUI.Enabled { + logger.Info(fmt.Sprintf("Web UI started on port %d", webUIPort)) + } + + if raftCoordinator != nil { + logger.Info("Cluster features enabled: Pipeline Replication, Batch Commit, Dynamic Resharding, Joint Consensus") + logger.Info("Observability metrics and health checks available at /api/webui/metrics and /api/webui/health") + + // ========== НОВАЯ ФУНКЦИОНАЛЬНОСТЬ: Вывод информации о новых компонентах ========== + logger.Info("Additional features:") + logger.Info(" - Single Point of Failure protection (Leader Fallback) - ENABLED") + logger.Info(" - Automatic Panic Recovery for goroutines - ENABLED") + logger.Info(" - Schema Migration support - ENABLED") + logger.Info(" - Full configuration validation - ENABLED") + } + + displayBanner(cfg.Cluster.Name, cfg.WebUI.Enabled, webUIPort, httpPort, raftCoordinator) + + replInstance := repl.NewRepl(store, raftCoordinator, logger, cfg, aclManager, pluginManager) + + // ========== GRACEFUL SHUTDOWN ========== + // Канал для сигналов завершения + sigChan := make(chan os.Signal, 1) + signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM, syscall.SIGHUP) + + // Запускаем graceful shutdown в отдельной горутине + go func() { + <-sigChan + utils.Println("\nReceived shutdown signal, starting graceful shutdown...") + logger.Info("Received shutdown signal, starting graceful shutdown...") + + // 1. Останавливаем REPL + logger.Info("Stopping REPL...") + replInstance.Close() + logger.Info("REPL stopped") + + // 2. Останавливаем HTTP сервер с таймаутом + logger.Info("Stopping HTTP server...") + shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 30*time.Second) + defer shutdownCancel() + + httpShutdownDone := make(chan struct{}) + go func() { + if err := httpServer.Stop(); err != nil { + logger.Error(fmt.Sprintf("HTTP server stop error: %v", err)) + } + close(httpShutdownDone) + }() + + select { + case <-httpShutdownDone: + logger.Info("HTTP server stopped") + case <-shutdownCtx.Done(): + logger.Warn("HTTP server shutdown timeout") + } + + // 3. Останавливаем WebUI + logger.Info("Stopping WebUI...") + webUIStopDone := make(chan struct{}) + go func() { + if err := webUI.Stop(); err != nil { + logger.Error(fmt.Sprintf("WebUI stop error: %v", err)) + } + close(webUIStopDone) + }() + + select { + case <-webUIStopDone: + logger.Info("WebUI stopped") + case <-time.After(10 * time.Second): + logger.Warn("WebUI shutdown timeout") + } + + // 4. Останавливаем плагины + if cfg.Plugins.Enabled && pluginManager != nil { + logger.Info("Stopping plugins...") + plugins := pluginManager.ListPlugins() + pluginStopDone := make(chan struct{}) + go func() { + for _, p := range plugins { + if err := pluginManager.StopPlugin(p.Name); err != nil { + logger.Warn(fmt.Sprintf("Failed to stop plugin %s: %v", p.Name, err)) + } + } + close(pluginStopDone) + }() + + select { + case <-pluginStopDone: + logger.Info("Plugins stopped") + case <-time.After(5 * time.Second): + logger.Warn("Plugins shutdown timeout") + } + } + + // 5. Сохраняем данные на диск + logger.Info("Persisting data to disk...") + persistDone := make(chan struct{}) + go func() { + // Используем публичный метод GetPersistenceManager() + persistenceMgr := raftCoordinator.GetPersistenceManager() + if raftCoordinator != nil && persistenceMgr != nil { + if err := persistenceMgr.SaveAll(); err != nil { + logger.Error(fmt.Sprintf("Failed to save data: %v", err)) + } else { + logger.Info("Data persisted to disk") + } + } + close(persistDone) + }() + + select { + case <-persistDone: + logger.Info("Persistence completed") + case <-time.After(15 * time.Second): + logger.Warn("Persistence timeout") + } + + // 6. Останавливаем координатор + logger.Info("Stopping Raft coordinator...") + raftStopDone := make(chan struct{}) + go func() { + raftCoordinator.Stop() + close(raftStopDone) + }() + + select { + case <-raftStopDone: + logger.Info("Raft coordinator stopped") + case <-time.After(15 * time.Second): + logger.Warn("Raft coordinator shutdown timeout") + } + + // 7. Останавливаем узел + logger.Info("Stopping cluster node...") + nodeStopDone := make(chan struct{}) + go func() { + node.Stop() + close(nodeStopDone) + }() + + select { + case <-nodeStopDone: + logger.Info("Cluster node stopped") + case <-time.After(10 * time.Second): + logger.Warn("Node shutdown timeout") + } + + // 8. Синхронизируем и закрываем логгер + logger.Info("Finalizing logger...") + if err := logger.Sync(); err != nil { + fmt.Printf("Failed to sync logger: %v\n", err) + } + logger.Close() + + utils.DisableColorMode() + fmt.Println("Futriis DB shutdown complete") + os.Exit(0) + }() + + // Запускаем REPL (блокирующий вызов) + if err := replInstance.Run(); err != nil { + logger.Error("REPL error: " + err.Error()) + utils.PrintError("REPL error: " + err.Error()) + os.Exit(1) + } +} + +func displayBanner(clusterName string, webUIEnabled bool, webUIPort int, httpPort int, coordinator *cluster.RaftCoordinator) { + utils.Println("") + bannerLines := []string{ + " futriix 3i²(by 02.04.2026) ", + " Distributed Document-Store in-memory database with support lua plugins ", + " Cluster status: enable (Raft consensus)", + " Cluster features: Pipeline Replication, Batch Commit, Dynamic Resharding", + " Cluster name: " + clusterName, + " HTTP API (for curl or wget utils only): http://localhost:" + fmt.Sprintf("%d", httpPort) + "/api/", + } + + if webUIEnabled { + bannerLines = append(bannerLines, fmt.Sprintf(" Web UI: http://localhost:%d/", webUIPort)) + bannerLines = append(bannerLines, " Observability endpoints: /api/webui/metrics, /api/webui/health") + } + + // ========== НОВАЯ ФУНКЦИОНАЛЬНОСТЬ: Добавляем информацию о новых компонентах в баннер ========== + bannerLines = append(bannerLines, " Additional features:") + bannerLines = append(bannerLines, " - Single Point of Failure protection (Leader Fallback)") + bannerLines = append(bannerLines, " - Automatic Panic Recovery for goroutines") + bannerLines = append(bannerLines, " - Schema Migration support") + bannerLines = append(bannerLines, " - Full configuration validation") + + // Выводим статус Fallback менеджера, если он активен + if coordinator != nil { + fallbackMgr := coordinator.GetFallbackManager() + if fallbackMgr != nil { + bannerLines = append(bannerLines, " - SPoF Protection: ACTIVE") + } + + panicRecoveryMgr := coordinator.GetPanicRecoveryManager() + if panicRecoveryMgr != nil { + stats := coordinator.GetPanicRecoveryStats() + if totalPanics, ok := stats["total_panics"].(uint64); ok && totalPanics > 0 { + bannerLines = append(bannerLines, fmt.Sprintf(" - Panics recovered: %d", totalPanics)) + } + } + + migrationStatus := coordinator.GetMigrationStatus() + if migrationStatus != nil && migrationStatus.PendingMigrations > 0 { + bannerLines = append(bannerLines, fmt.Sprintf(" - Pending migrations: %d", migrationStatus.PendingMigrations)) + bannerLines = append(bannerLines, " - Run 'migrate status' for details") + } + } + + bannerLines = append(bannerLines, []string{ + "", + " Type 'quit' or 'exit' to quit", + " Type 'status' to see cluster status", + " Type 'acl login ' to authenticate", + " Type 'plugin list' to see loaded plugins", + " Type 'cluster pipeline' to see pipeline stats", + " Type 'cluster reshard' to trigger manual resharding", + " Type 'migrate status' to see schema migration status", + " Type 'fallback status' to see SPoF protection status", + " Type 'panic stats' to see panic recovery statistics", + }...) + + for _, line := range bannerLines { + utils.PrintInfo(line) + } +} + +func printJSON(data interface{}) { + jsonData, err := json.MarshalIndent(data, "", " ") + if err != nil { + utils.PrintError("Failed to marshal JSON: " + err.Error()) + return + } + fmt.Println(string(jsonData)) +}