150 lines
3.5 KiB
Go
150 lines
3.5 KiB
Go
package controller
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
"net/http"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"ezops/middleware"
|
|
ezredis "ezops/pkg/redis"
|
|
"ezops/pkg/wshub"
|
|
|
|
"github.com/gin-gonic/gin"
|
|
"github.com/gorilla/websocket"
|
|
"gorm.io/gorm"
|
|
)
|
|
|
|
// WSPipelineLogHandler WebSocket 流水线日志推流
|
|
type WSPipelineLogHandler struct {
|
|
db *gorm.DB
|
|
cache ezredis.Cache
|
|
wsMgr *wshub.WSManager
|
|
}
|
|
|
|
// NewWSPipelineLogHandler 创建 WebSocket 日志处理器
|
|
func NewWSPipelineLogHandler(db *gorm.DB, cache ezredis.Cache, wsMgr *wshub.WSManager) *WSPipelineLogHandler {
|
|
return &WSPipelineLogHandler{db: db, cache: cache, wsMgr: wsMgr}
|
|
}
|
|
|
|
// StreamLog WebSocket 端点：/ws/pipeline/:runId/log
|
|
func (h *WSPipelineLogHandler) StreamLog(c *gin.Context) {
|
|
runIDStr := c.Param("runId")
|
|
runID, err := strconv.ParseUint(runIDStr, 10, 32)
|
|
if err != nil {
|
|
c.JSON(http.StatusBadRequest, gin.H{"error": "无效的 runId"})
|
|
return
|
|
}
|
|
|
|
userID := middleware.GetUserID(c)
|
|
username := middleware.GetUsername(c)
|
|
|
|
ws, info, err := h.wsMgr.Upgrade(c, wshub.TypePipeline, userID, username)
|
|
if err != nil {
|
|
return
|
|
}
|
|
defer h.wsMgr.Unregister(info)
|
|
h.wsMgr.StartHeartbeat(ws, info)
|
|
|
|
logKey := fmt.Sprintf("pipeline:run:%d:logs", uint(runID))
|
|
ctx := c.Request.Context()
|
|
|
|
// 首次发送：已有日志
|
|
raw, _ := h.cache.Get(ctx, logKey)
|
|
lastLen := 0
|
|
if raw != "" {
|
|
ws.WriteJSON(map[string]interface{}{"type": "init", "logs": raw})
|
|
var entries []LogEntry
|
|
if err := json.Unmarshal([]byte(raw), &entries); err == nil {
|
|
lastLen = len(entries)
|
|
}
|
|
}
|
|
|
|
// 轮询新日志（500ms 间隔）
|
|
ticker := time.NewTicker(500 * time.Millisecond)
|
|
defer ticker.Stop()
|
|
|
|
done := make(chan struct{})
|
|
go func() {
|
|
defer close(done)
|
|
for {
|
|
_, _, err := ws.ReadMessage()
|
|
if err != nil {
|
|
return
|
|
}
|
|
}
|
|
}()
|
|
|
|
for {
|
|
select {
|
|
case <-done:
|
|
return
|
|
case <-ticker.C:
|
|
raw, err := h.cache.Get(ctx, logKey)
|
|
if err != nil || raw == "" {
|
|
continue
|
|
}
|
|
var entries []LogEntry
|
|
if err := json.Unmarshal([]byte(raw), &entries); err != nil {
|
|
continue
|
|
}
|
|
if len(entries) > lastLen {
|
|
// 增量推送
|
|
newEntries := entries[lastLen:]
|
|
lastLen = len(entries)
|
|
for _, e := range newEntries {
|
|
buf, _ := json.Marshal(e)
|
|
if err := ws.WriteMessage(websocket.TextMessage, buf); err != nil {
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
// 检查流水线是否已结束（日志 key 被删除 = 已写入 DB）
|
|
if lastLen > 0 {
|
|
exists, _ := h.cache.Exists(ctx, logKey)
|
|
if exists == 0 {
|
|
// 发送完成信号
|
|
ws.WriteJSON(map[string]interface{}{"type": "done"})
|
|
return
|
|
}
|
|
}
|
|
case <-ctx.Done():
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
// GetLogHistory HTTP 日志历史（已完成的任务从 DB 读取）
|
|
func (h *WSPipelineLogHandler) GetLogHistory(c *gin.Context) {
|
|
runIDStr := c.Param("runId")
|
|
runID, err := strconv.ParseUint(runIDStr, 10, 32)
|
|
if err != nil {
|
|
c.JSON(http.StatusBadRequest, gin.H{"error": "无效的 runId"})
|
|
return
|
|
}
|
|
|
|
// 先查 Redis（执行中）
|
|
logKey := fmt.Sprintf("pipeline:run:%d:logs", uint(runID))
|
|
ctx := c.Request.Context()
|
|
|
|
raw, _ := h.cache.Get(ctx, logKey)
|
|
if raw != "" && !strings.HasPrefix(raw, "[") {
|
|
raw = "[]"
|
|
}
|
|
if raw != "" {
|
|
c.JSON(http.StatusOK, gin.H{"code": 200, "data": gin.H{"source": "redis", "logs": raw}})
|
|
return
|
|
}
|
|
|
|
// 查 DB（已完成）
|
|
var logs string
|
|
h.db.Table("deploy_pipeline_runs").Where("id = ?", uint(runID)).Select("logs").Scan(&logs)
|
|
if logs == "" {
|
|
logs = "[]"
|
|
}
|
|
c.JSON(http.StatusOK, gin.H{"code": 200, "data": gin.H{"source": "db", "logs": logs}})
|
|
}
|