Admin UI: Add message queue to admin UI (#6958)

* add a menu item "Message Queue"

* add a menu item "Message Queue"
  * move the "brokers" link under it.
  * add "topics", "subscribers". Add pages for them.

* refactor

* show topic details

* admin display publisher and subscriber info

* remove publisher and subscribers from the topic row pull down

* collecting more stats from publishers and subscribers

* fix layout

* fix publisher name

* add local listeners for mq broker and agent

* render consumer group offsets

* remove subscribers from left menu

* topic with retention

* support editing topic retention

* show retention when listing topics

* create bucket

* Update s3_buckets_templ.go

* embed the static assets into the binary

fix https://github.com/seaweedfs/seaweedfs/issues/6964

1 files changed, 31 insertions, 0 deletions

diff --git a/weed/admin/handlers/admin_handlers.go b/weed/admin/handlers/admin_handlers.go
index 03d156d08..dc7905bc1 100644
--- a/weed/admin/handlers/admin_handlers.go
+++ b/weed/admin/handlers/admin_handlers.go
@@ -18,6 +18,7 @@ type AdminHandlers struct {
 	fileBrowserHandlers *FileBrowserHandlers
 	userHandlers        *UserHandlers
 	maintenanceHandlers *MaintenanceHandlers
+	mqHandlers          *MessageQueueHandlers
 }
 
 // NewAdminHandlers creates a new instance of AdminHandlers
@@ -27,6 +28,7 @@ func NewAdminHandlers(adminServer *dash.AdminServer) *AdminHandlers {
 	fileBrowserHandlers := NewFileBrowserHandlers(adminServer)
 	userHandlers := NewUserHandlers(adminServer)
 	maintenanceHandlers := NewMaintenanceHandlers(adminServer)
+	mqHandlers := NewMessageQueueHandlers(adminServer)
 	return &AdminHandlers{
 		adminServer:         adminServer,
 		authHandlers:        authHandlers,
@@ -34,6 +36,7 @@ func NewAdminHandlers(adminServer *dash.AdminServer) *AdminHandlers {
 		fileBrowserHandlers: fileBrowserHandlers,
 		userHandlers:        userHandlers,
 		maintenanceHandlers: maintenanceHandlers,
+		mqHandlers:          mqHandlers,
 	}
 }
 
@@ -72,6 +75,11 @@ func (h *AdminHandlers) SetupRoutes(r *gin.Engine, authRequired bool, username,
 		protected.GET("/cluster/volumes/:id/:server", h.clusterHandlers.ShowVolumeDetails)
 		protected.GET("/cluster/collections", h.clusterHandlers.ShowClusterCollections)
 
+		// Message Queue management routes
+		protected.GET("/mq/brokers", h.mqHandlers.ShowBrokers)
+		protected.GET("/mq/topics", h.mqHandlers.ShowTopics)
+		protected.GET("/mq/topics/:namespace/:topic", h.mqHandlers.ShowTopicDetails)
+
 		// Maintenance system routes
 		protected.GET("/maintenance", h.maintenanceHandlers.ShowMaintenanceQueue)
 		protected.GET("/maintenance/workers", h.maintenanceHandlers.ShowMaintenanceWorkers)
@@ -144,6 +152,15 @@ func (h *AdminHandlers) SetupRoutes(r *gin.Engine, authRequired bool, username,
 				maintenanceApi.GET("/config", h.adminServer.GetMaintenanceConfigAPI)
 				maintenanceApi.PUT("/config", h.adminServer.UpdateMaintenanceConfigAPI)
 			}
+
+			// Message Queue API routes
+			mqApi := api.Group("/mq")
+			{
+				mqApi.GET("/topics/:namespace/:topic", h.mqHandlers.GetTopicDetailsAPI)
+				mqApi.POST("/topics/create", h.mqHandlers.CreateTopicAPI)
+				mqApi.POST("/topics/retention/update", h.mqHandlers.UpdateTopicRetentionAPI)
+				mqApi.POST("/retention/purge", h.adminServer.TriggerTopicRetentionPurgeAPI)
+			}
 		}
 	} else {
 		// No authentication required - all routes are public
@@ -166,6 +183,11 @@ func (h *AdminHandlers) SetupRoutes(r *gin.Engine, authRequired bool, username,
 		r.GET("/cluster/volumes/:id/:server", h.clusterHandlers.ShowVolumeDetails)
 		r.GET("/cluster/collections", h.clusterHandlers.ShowClusterCollections)
 
+		// Message Queue management routes
+		r.GET("/mq/brokers", h.mqHandlers.ShowBrokers)
+		r.GET("/mq/topics", h.mqHandlers.ShowTopics)
+		r.GET("/mq/topics/:namespace/:topic", h.mqHandlers.ShowTopicDetails)
+
 		// Maintenance system routes
 		r.GET("/maintenance", h.maintenanceHandlers.ShowMaintenanceQueue)
 		r.GET("/maintenance/workers", h.maintenanceHandlers.ShowMaintenanceWorkers)
@@ -238,6 +260,15 @@ func (h *AdminHandlers) SetupRoutes(r *gin.Engine, authRequired bool, username,
 				maintenanceApi.GET("/config", h.adminServer.GetMaintenanceConfigAPI)
 				maintenanceApi.PUT("/config", h.adminServer.UpdateMaintenanceConfigAPI)
 			}
+
+			// Message Queue API routes
+			mqApi := api.Group("/mq")
+			{
+				mqApi.GET("/topics/:namespace/:topic", h.mqHandlers.GetTopicDetailsAPI)
+				mqApi.POST("/topics/create", h.mqHandlers.CreateTopicAPI)
+				mqApi.POST("/topics/retention/update", h.mqHandlers.UpdateTopicRetentionAPI)
+				mqApi.POST("/retention/purge", h.adminServer.TriggerTopicRetentionPurgeAPI)
+			}
 		}
 	}
 }