From 3e3b665b8912f389a3991e0e2dadafef1de9a086 Mon Sep 17 00:00:00 2001 From: Jin Hai Date: Thu, 5 Mar 2026 15:18:40 +0800 Subject: [PATCH] RAGFlow admin server go version (#13394) ### What problem does this PR solve? 1. init go admin server 2. refactor api server router 3. add benchmark CI to 450s time limit 4. remove docker builder container after building ### Type of change - [x] Bug Fix (non-breaking change which fixes an issue) - [x] New Feature (non-breaking change which adds functionality) --------- Signed-off-by: Jin Hai --- .github/workflows/tests.yml | 3 + cmd/admin_server.go | 129 +++++++++++++++++ internal/admin/handler.go | 270 ++++++++++++++++++++++++++++++++++++ internal/admin/router.go | 65 +++++++++ internal/admin/service.go | 230 ++++++++++++++++++++++++++++++ internal/router/router.go | 102 +++++++------- 6 files changed, 748 insertions(+), 51 deletions(-) create mode 100644 cmd/admin_server.go create mode 100644 internal/admin/handler.go create mode 100644 internal/admin/router.go create mode 100644 internal/admin/service.go diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 8d660e3c3f..47d94c212f 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -137,6 +137,9 @@ jobs: sudo docker run --privileged -d --name ${BUILDER_CONTAINER} -e TZ=${TZ} -e UV_INDEX=https://pypi.tuna.tsinghua.edu.cn/simple -v ${PWD}:/ragflow -v ${PWD}/internal/cpp/resource:/usr/share/infinity/resource infiniflow/infinity_builder:ubuntu22_clang20 sudo docker exec ${BUILDER_CONTAINER} bash -c "git config --global safe.directory \"*\" && cd /ragflow && ./build.sh --cpp" ./build.sh --go + if [[ -n "${BUILDER_CONTAINER}" ]]; then + sudo docker rm -f -v "${BUILDER_CONTAINER}" + fi - name: Build ragflow:nightly run: | diff --git a/cmd/admin_server.go b/cmd/admin_server.go new file mode 100644 index 0000000000..d553a44d9e --- /dev/null +++ b/cmd/admin_server.go @@ -0,0 +1,129 @@ +// +// Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + +package main + +import ( + "flag" + "os" + + "github.com/gin-gonic/gin" + "go.uber.org/zap" + + "ragflow/internal/admin" + "ragflow/internal/dao" + "ragflow/internal/logger" + "ragflow/internal/server" + "ragflow/internal/utility" +) + +// AdminServer admin server +type AdminServer struct { + router *admin.Router + handler *admin.Handler + service *admin.Service + engine *gin.Engine + port string +} + +// NewAdminServer create admin server +func NewAdminServer(port string) *AdminServer { + return &AdminServer{ + port: port, + } +} + +// Init initialize admin server +func (s *AdminServer) Init() error { + gin.SetMode(gin.ReleaseMode) + s.engine = gin.New() + s.engine.Use(gin.Recovery()) + + // Initialize layers + s.service = admin.NewService() + s.handler = admin.NewHandler(s.service) + s.router = admin.NewRouter(s.handler) + + // Setup routes + s.router.Setup(s.engine) + + return nil +} + +// Run start admin server +func (s *AdminServer) Run() error { + logger.Info("Starting admin server", zap.String("port", s.port)) + return s.engine.Run(":" + s.port) +} + +func main() { + var configPath string + flag.StringVar(&configPath, "config", "", "Path to configuration file") + flag.Parse() + + // Initialize logger + if err := logger.Init("debug"); err != nil { + panic("failed to initialize logger: " + err.Error()) + } + + // Initialize configuration + if err := server.Init(configPath); err != nil { + logger.Error("Failed to initialize configuration", err) + os.Exit(1) + } + + // Set logger for server package + server.SetLogger(logger.Logger) + + cfg := server.GetConfig() + logger.Info("Configuration loaded", + zap.String("database_host", cfg.Database.Host), + zap.Int("database_port", cfg.Database.Port), + ) + + // Initialize database + if err := dao.InitDB(); err != nil { + logger.Error("Failed to initialize database", err) + os.Exit(1) + } + + // Create and start admin server (port 9381) + adminServer := NewAdminServer("9381") + if err := adminServer.Init(); err != nil { + logger.Error("Failed to initialize admin server", err) + os.Exit(1) + } + + // Print RAGFlow Admin logo + logger.Info("" + + "\n ____ ___ ______________ ___ __ _ \n" + + " / __ \\/ | / ____/ ____/ /___ _ __ / | ____/ /___ ___ (_)___ \n" + + " / /_/ / /| |/ / __/ /_ / / __ \\ | /| / / / /| |/ __ / __ `__ \\/ / __ \\ \n" + + " / _, _/ ___ / /_/ / __/ / / /_/ / |/ |/ / / ___ / /_/ / / / / / / / / / / /\n" + + " /_/ |_/_/ |_\\____/_/ /_/\\____/|__/|__/ /_/ |_\\__,_/_/ /_/ /_/_/_/ /_/ \n") + + // Print RAGFlow version + logger.Info("RAGFlow version", zap.String("version", utility.GetRAGFlowVersion())) + + // Print all configuration settings + server.PrintAll() + + logger.Info("Starting RAGFlow Admin Server", zap.String("port", "9381")) + if err := adminServer.Run(); err != nil { + logger.Error("Admin server error", err) + os.Exit(1) + } +} diff --git a/internal/admin/handler.go b/internal/admin/handler.go new file mode 100644 index 0000000000..18511ea6f3 --- /dev/null +++ b/internal/admin/handler.go @@ -0,0 +1,270 @@ +// +// Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + +package admin + +import ( + "errors" + "net/http" + + "github.com/gin-gonic/gin" +) + +// Common errors +var ( + ErrInvalidCredentials = errors.New("invalid credentials") + ErrUserNotFound = errors.New("user not found") + ErrInvalidToken = errors.New("invalid token") +) + +// Handler admin handler +type Handler struct { + service *Service +} + +// NewHandler create admin handler +func NewHandler(service *Service) *Handler { + return &Handler{service: service} +} + +// Health health check +func (h *Handler) Health(c *gin.Context) { + c.JSON(200, gin.H{"status": "ok"}) +} + +// LoginHTTPRequest login request body +type LoginHTTPRequest struct { + Email string `json:"email" binding:"required"` + Password string `json:"password" binding:"required"` +} + +// Login handle admin login +func (h *Handler) Login(c *gin.Context) { + var req LoginHTTPRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(400, gin.H{"error": err.Error()}) + return + } + + svcReq := &LoginRequest{ + Email: req.Email, + Password: req.Password, + } + + resp, err := h.service.Login(svcReq) + if err != nil { + if errors.Is(err, ErrInvalidCredentials) { + c.JSON(401, gin.H{"error": "invalid credentials"}) + return + } + c.JSON(500, gin.H{"error": err.Error()}) + return + } + + c.JSON(200, gin.H{ + "token": resp.Token, + "user": gin.H{ + "id": resp.UserID, + "email": resp.Email, + "nickname": resp.Nickname, + }, + }) +} + +// ListUsers handle list users +func (h *Handler) ListUsers(c *gin.Context) { + // Parse pagination params + offset := 0 + limit := 20 + + svcReq := &ListUsersRequest{ + Offset: offset, + Limit: limit, + } + + resp, err := h.service.ListUsers(svcReq) + if err != nil { + c.JSON(500, gin.H{"error": err.Error()}) + return + } + + // Convert to response format + var result []gin.H + for _, user := range resp.Users { + result = append(result, gin.H{ + "id": user.ID, + "email": user.Email, + "nickname": user.Nickname, + "is_active": user.IsActive, + "create_time": user.CreateTime, + "update_time": user.UpdateTime, + }) + } + + c.JSON(200, gin.H{ + "data": result, + "total": resp.Total, + }) +} + +// GetUser handle get user +func (h *Handler) GetUser(c *gin.Context) { + id := c.Param("id") + if id == "" { + c.JSON(400, gin.H{"error": "user id is required"}) + return + } + + svcReq := &GetUserRequest{ID: id} + user, err := h.service.GetUser(svcReq) + if err != nil { + if errors.Is(err, ErrUserNotFound) { + c.JSON(404, gin.H{"error": "user not found"}) + return + } + c.JSON(500, gin.H{"error": err.Error()}) + return + } + + c.JSON(200, gin.H{ + "id": user.ID, + "email": user.Email, + "nickname": user.Nickname, + "is_active": user.IsActive, + "create_time": user.CreateTime, + "update_time": user.UpdateTime, + }) +} + +// UpdateUserHTTPRequest update user request body +type UpdateUserHTTPRequest struct { + Nickname string `json:"nickname"` + IsActive *string `json:"is_active,omitempty"` +} + +// UpdateUser handle update user +func (h *Handler) UpdateUser(c *gin.Context) { + id := c.Param("id") + if id == "" { + c.JSON(400, gin.H{"error": "user id is required"}) + return + } + + var req UpdateUserHTTPRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(400, gin.H{"error": err.Error()}) + return + } + + svcReq := &UpdateUserRequest{ + ID: id, + Nickname: req.Nickname, + IsActive: req.IsActive, + } + + if err := h.service.UpdateUser(svcReq); err != nil { + if errors.Is(err, ErrUserNotFound) { + c.JSON(404, gin.H{"error": "user not found"}) + return + } + c.JSON(500, gin.H{"error": err.Error()}) + return + } + + c.JSON(200, gin.H{"message": "user updated"}) +} + +// DeleteUser handle delete user +func (h *Handler) DeleteUser(c *gin.Context) { + id := c.Param("id") + if id == "" { + c.JSON(400, gin.H{"error": "user id is required"}) + return + } + + svcReq := &DeleteUserRequest{ID: id} + if err := h.service.DeleteUser(svcReq); err != nil { + c.JSON(500, gin.H{"error": err.Error()}) + return + } + + c.JSON(200, gin.H{"message": "user deleted"}) +} + +// GetConfig handle get system config +func (h *Handler) GetConfig(c *gin.Context) { + config := h.service.GetSystemConfig() + c.JSON(200, config) +} + +// UpdateConfig handle update system config +func (h *Handler) UpdateConfig(c *gin.Context) { + var req map[string]interface{} + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(400, gin.H{"error": err.Error()}) + return + } + + if err := h.service.UpdateSystemConfig(req); err != nil { + c.JSON(500, gin.H{"error": err.Error()}) + return + } + + c.JSON(200, gin.H{"message": "config updated"}) +} + +// GetStatus handle get system status +func (h *Handler) GetStatus(c *gin.Context) { + status := h.service.GetSystemStatus() + c.JSON(200, status) +} + +// AuthMiddleware JWT auth middleware +func (h *Handler) AuthMiddleware() gin.HandlerFunc { + return func(c *gin.Context) { + token := c.GetHeader("Authorization") + if token == "" { + c.JSON(401, gin.H{"error": "missing authorization header"}) + c.Abort() + return + } + + // Remove "Bearer " prefix + if len(token) > 7 && token[:7] == "Bearer " { + token = token[7:] + } + + // Validate token + user, err := h.service.ValidateToken(token) + if err != nil { + c.JSON(401, gin.H{"error": "invalid token"}) + c.Abort() + return + } + + c.Set("user", user) + c.Next() + } +} + +// HandleNoRoute handle undefined routes +func (h *Handler) HandleNoRoute(c *gin.Context) { + c.JSON(http.StatusNotFound, gin.H{ + "error": "Not Found", + "message": "The requested resource was not found", + "path": c.Request.URL.Path, + }) +} diff --git a/internal/admin/router.go b/internal/admin/router.go new file mode 100644 index 0000000000..fa6572888b --- /dev/null +++ b/internal/admin/router.go @@ -0,0 +1,65 @@ +// +// Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + +package admin + +import ( + "github.com/gin-gonic/gin" +) + +// Router admin router +type Router struct { + handler *Handler +} + +// NewRouter create admin router +func NewRouter(handler *Handler) *Router { + return &Router{handler: handler} +} + +// Setup setup routes +func (r *Router) Setup(engine *gin.Engine) { + // Health check + engine.GET("/health", r.handler.Health) + + // Admin API routes + admin := engine.Group("/admin") + { + // Auth + admin.POST("/login", r.handler.Login) + + // Protected routes + protected := admin.Group("") + protected.Use(r.handler.AuthMiddleware()) + { + // User management + protected.GET("/users", r.handler.ListUsers) + protected.GET("/users/:id", r.handler.GetUser) + protected.PUT("/users/:id", r.handler.UpdateUser) + protected.DELETE("/users/:id", r.handler.DeleteUser) + + // System config + protected.GET("/config", r.handler.GetConfig) + protected.PUT("/config", r.handler.UpdateConfig) + + // System status + protected.GET("/status", r.handler.GetStatus) + } + } + + // Handle undefined routes + engine.NoRoute(r.handler.HandleNoRoute) +} diff --git a/internal/admin/service.go b/internal/admin/service.go new file mode 100644 index 0000000000..1c4b430f58 --- /dev/null +++ b/internal/admin/service.go @@ -0,0 +1,230 @@ +// +// Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + +package admin + +import ( + "time" + + "ragflow/internal/dao" + "ragflow/internal/model" +) + +// Service admin service layer +type Service struct { + userDAO *dao.UserDAO +} + +// NewService create admin service +func NewService() *Service { + return &Service{ + userDAO: dao.NewUserDAO(), + } +} + +// LoginRequest login request +type LoginRequest struct { + Email string + Password string +} + +// LoginResponse login response +type LoginResponse struct { + Token string + UserID string + Email string + Nickname string +} + +// Login admin login +func (s *Service) Login(req *LoginRequest) (*LoginResponse, error) { + // Get user by email + user, err := s.userDAO.GetByEmail(req.Email) + if err != nil { + return nil, ErrInvalidCredentials + } + + // Verify password + if user.Password == nil || *user.Password != req.Password { + return nil, ErrInvalidCredentials + } + + // Generate access token + token := generateToken() + if err := s.userDAO.UpdateAccessToken(user, token); err != nil { + return nil, err + } + + return &LoginResponse{ + Token: token, + UserID: user.ID, + Email: user.Email, + Nickname: user.Nickname, + }, nil +} + +// ListUsersRequest list users request +type ListUsersRequest struct { + Offset int + Limit int +} + +// ListUsersResponse list users response +type ListUsersResponse struct { + Users []*UserInfo + Total int64 +} + +// UserInfo user info +type UserInfo struct { + ID string + Email string + Nickname string + IsActive string + CreateTime int64 + UpdateTime *int64 +} + +// ListUsers list all users +func (s *Service) ListUsers(req *ListUsersRequest) (*ListUsersResponse, error) { + users, total, err := s.userDAO.List(req.Offset, req.Limit) + if err != nil { + return nil, err + } + + var result []*UserInfo + for _, user := range users { + result = append(result, &UserInfo{ + ID: user.ID, + Email: user.Email, + Nickname: user.Nickname, + IsActive: user.IsActive, + CreateTime: user.CreateTime, + UpdateTime: user.UpdateTime, + }) + } + + return &ListUsersResponse{ + Users: result, + Total: total, + }, nil +} + +// GetUserRequest get user request +type GetUserRequest struct { + ID string +} + +// GetUser get user by ID +func (s *Service) GetUser(req *GetUserRequest) (*UserInfo, error) { + var user model.User + err := dao.DB.Where("id = ?", req.ID).First(&user).Error + if err != nil { + return nil, ErrUserNotFound + } + + return &UserInfo{ + ID: user.ID, + Email: user.Email, + Nickname: user.Nickname, + IsActive: user.IsActive, + CreateTime: user.CreateTime, + UpdateTime: user.UpdateTime, + }, nil +} + +// UpdateUserRequest update user request +type UpdateUserRequest struct { + ID string + Nickname string + IsActive *string +} + +// UpdateUser update user +func (s *Service) UpdateUser(req *UpdateUserRequest) error { + var user model.User + if err := dao.DB.Where("id = ?", req.ID).First(&user).Error; err != nil { + return ErrUserNotFound + } + + if req.Nickname != "" { + user.Nickname = req.Nickname + } + if req.IsActive != nil { + user.IsActive = *req.IsActive + } + + return dao.DB.Save(&user).Error +} + +// DeleteUserRequest delete user request +type DeleteUserRequest struct { + ID string +} + +// DeleteUser delete user +func (s *Service) DeleteUser(req *DeleteUserRequest) error { + return dao.DB.Where("id = ?", req.ID).Delete(&model.User{}).Error +} + +// GetSystemConfig get system config +func (s *Service) GetSystemConfig() map[string]interface{} { + // TODO: Load from database or config file + return map[string]interface{}{ + "system_name": "RAGFlow Admin", + "version": "1.0.0", + } +} + +// UpdateSystemConfig update system config +func (s *Service) UpdateSystemConfig(config map[string]interface{}) error { + // TODO: Save to database or config file + return nil +} + +// GetSystemStatus get system status +func (s *Service) GetSystemStatus() map[string]interface{} { + // TODO: Get real status from services + return map[string]interface{}{ + "status": "running", + "uptime": time.Since(time.Now()).String(), + "db_status": "connected", + } +} + +// ValidateToken validate access token +func (s *Service) ValidateToken(token string) (*model.User, error) { + user, err := s.userDAO.GetByAccessToken(token) + if err != nil { + return nil, ErrInvalidToken + } + return user, nil +} + +// generateToken generate a simple token +func generateToken() string { + return time.Now().Format("20060102150405") + randomString(16) +} + +// randomString generate random string +func randomString(n int) string { + const letters = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789" + b := make([]byte, n) + for i := range b { + b[i] = letters[time.Now().UnixNano()%int64(len(letters))] + } + return string(b) +} diff --git a/internal/router/router.go b/internal/router/router.go index a7a7786772..bcc1d68384 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -128,65 +128,65 @@ func (r *Router) Setup(engine *gin.Engine) { { authors.GET("/:author_id/documents", r.documentHandler.GetDocumentsByAuthorID) } + } - // Knowledge base routes - kb := engine.Group("/v1/kb") - { - kb.POST("/list", r.knowledgebaseHandler.ListKbs) - } + // Knowledge base routes + kb := engine.Group("/v1/kb") + { + kb.POST("/list", r.knowledgebaseHandler.ListKbs) + } - // Chunk routes - chunk := engine.Group("/v1/chunk") - { - chunk.POST("/retrieval_test", r.chunkHandler.RetrievalTest) - } + // Chunk routes + chunk := engine.Group("/v1/chunk") + { + chunk.POST("/retrieval_test", r.chunkHandler.RetrievalTest) + } - // LLM routes - llm := engine.Group("/v1/llm") - { - llm.GET("/my_llms", r.llmHandler.GetMyLLMs) - llm.GET("/factories", r.llmHandler.Factories) - llm.GET("/list", r.llmHandler.ListApp) - } + // LLM routes + llm := engine.Group("/v1/llm") + { + llm.GET("/my_llms", r.llmHandler.GetMyLLMs) + llm.GET("/factories", r.llmHandler.Factories) + llm.GET("/list", r.llmHandler.ListApp) + } - // Chat routes - chat := engine.Group("/v1/dialog") - { - chat.GET("/list", r.chatHandler.ListChats) - chat.POST("/next", r.chatHandler.ListChatsNext) - chat.POST("/set", r.chatHandler.SetDialog) - chat.POST("/rm", r.chatHandler.RemoveChats) - } + // Chat routes + chat := engine.Group("/v1/dialog") + { + chat.GET("/list", r.chatHandler.ListChats) + chat.POST("/next", r.chatHandler.ListChatsNext) + chat.POST("/set", r.chatHandler.SetDialog) + chat.POST("/rm", r.chatHandler.RemoveChats) + } - // Chat session (conversation) routes - session := engine.Group("/v1/conversation") - { - session.POST("/set", r.chatSessionHandler.SetChatSession) - session.POST("/rm", r.chatSessionHandler.RemoveChatSessions) - session.GET("/list", r.chatSessionHandler.ListChatSessions) - session.POST("/completion", r.chatSessionHandler.Completion) - } + // Chat session (conversation) routes + session := engine.Group("/v1/conversation") + { + session.POST("/set", r.chatSessionHandler.SetChatSession) + session.POST("/rm", r.chatSessionHandler.RemoveChatSessions) + session.GET("/list", r.chatSessionHandler.ListChatSessions) + session.POST("/completion", r.chatSessionHandler.Completion) + } - // Connector routes - connector := engine.Group("/v1/connector") - { - connector.GET("/list", r.connectorHandler.ListConnectors) - } + // Connector routes + connector := engine.Group("/v1/connector") + { + connector.GET("/list", r.connectorHandler.ListConnectors) + } - // Search routes - search := engine.Group("/v1/search") - { - search.POST("/list", r.searchHandler.ListSearchApps) - } + // Search routes + search := engine.Group("/v1/search") + { + search.POST("/list", r.searchHandler.ListSearchApps) + } - // File routes - file := engine.Group("/v1/file") - { - file.GET("/list", r.fileHandler.ListFiles) - file.GET("/root_folder", r.fileHandler.GetRootFolder) - file.GET("/parent_folder", r.fileHandler.GetParentFolder) - file.GET("/all_parent_folder", r.fileHandler.GetAllParentFolders) - } + // File routes + file := engine.Group("/v1/file") + { + file.GET("/list", r.fileHandler.ListFiles) + file.GET("/root_folder", r.fileHandler.GetRootFolder) + file.GET("/parent_folder", r.fileHandler.GetParentFolder) + file.GET("/all_parent_folder", r.fileHandler.GetAllParentFolders) } // Handle undefined routes