feat: 添加请求追踪中间件并支持查询过期订阅
All checks were successful
Build docker and publish / build (20.15.1) (push) Successful in 5m10s
All checks were successful
Build docker and publish / build (20.15.1) (push) Successful in 5m10s
添加请求追踪中间件以记录请求和响应内容 在用户订阅查询中新增includeExpired参数支持查询历史记录 完善配置系统以支持float64类型默认值解析
This commit is contained in:
parent
55c778b65b
commit
ef64a876cd
@ -23,6 +23,7 @@ import (
|
||||
"github.com/perfect-panel/server/pkg/orm"
|
||||
"github.com/perfect-panel/server/pkg/service"
|
||||
"github.com/perfect-panel/server/pkg/tool"
|
||||
"github.com/perfect-panel/server/pkg/trace"
|
||||
"github.com/perfect-panel/server/queue"
|
||||
"github.com/perfect-panel/server/scheduler"
|
||||
"github.com/spf13/cobra"
|
||||
@ -49,6 +50,7 @@ var startCmd = &cobra.Command{
|
||||
func run() {
|
||||
services := getServers()
|
||||
defer services.Stop()
|
||||
defer trace.StopAgent()
|
||||
go services.Start()
|
||||
quit := make(chan os.Signal, 1)
|
||||
signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)
|
||||
@ -89,6 +91,9 @@ func getServers() *service.Group {
|
||||
logger.Errorf("Logger setup failed: %v", err.Error())
|
||||
}
|
||||
|
||||
// init trace
|
||||
trace.StartAgent(c.Trace)
|
||||
|
||||
// init service context
|
||||
ctx := svc.NewServiceContext(c)
|
||||
services := service.NewServiceGroup()
|
||||
|
||||
@ -5,6 +5,7 @@ import (
|
||||
|
||||
"github.com/perfect-panel/server/pkg/logger"
|
||||
"github.com/perfect-panel/server/pkg/orm"
|
||||
"github.com/perfect-panel/server/pkg/trace"
|
||||
)
|
||||
|
||||
type Config struct {
|
||||
@ -29,6 +30,7 @@ type Config struct {
|
||||
Invite InviteConfig `yaml:"Invite"`
|
||||
Telegram Telegram `yaml:"Telegram"`
|
||||
Log Log `yaml:"Log"`
|
||||
Trace trace.Config `yaml:"Trace"`
|
||||
Administrator struct {
|
||||
Email string `yaml:"Email" default:"admin@ppanel.dev"`
|
||||
Password string `yaml:"Password" default:"password"`
|
||||
|
||||
@ -1,17 +1,26 @@
|
||||
package user
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/perfect-panel/server/internal/logic/public/user"
|
||||
"github.com/perfect-panel/server/internal/svc"
|
||||
"github.com/perfect-panel/server/pkg/constant"
|
||||
"github.com/perfect-panel/server/pkg/result"
|
||||
)
|
||||
|
||||
// Query User Subscribe
|
||||
func QueryUserSubscribeHandler(svcCtx *svc.ServiceContext) func(c *gin.Context) {
|
||||
return func(c *gin.Context) {
|
||||
// 1. Get param from URL Query (?includeExpired=all)
|
||||
value := c.Query("includeExpired")
|
||||
|
||||
l := user.NewQueryUserSubscribeLogic(c.Request.Context(), svcCtx)
|
||||
// 2. Inject param into Request Context
|
||||
// Note: Must use context.WithValue to create new ctx
|
||||
ctx := context.WithValue(c.Request.Context(), constant.CtxKeyIncludeExpired, value)
|
||||
|
||||
l := user.NewQueryUserSubscribeLogic(ctx, svcCtx)
|
||||
resp, err := l.QueryUserSubscribe()
|
||||
result.HttpResult(c, resp, err)
|
||||
}
|
||||
|
||||
@ -40,6 +40,8 @@ import (
|
||||
)
|
||||
|
||||
func RegisterHandlers(router *gin.Engine, serverCtx *svc.ServiceContext) {
|
||||
router.Use(middleware.TraceMiddleware(serverCtx))
|
||||
|
||||
adminAdsGroupRouter := router.Group("/v1/admin/ads")
|
||||
adminAdsGroupRouter.Use(middleware.AuthMiddleware(serverCtx))
|
||||
|
||||
|
||||
@ -1,8 +1,10 @@
|
||||
package middleware
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
|
||||
@ -18,6 +20,17 @@ import (
|
||||
"github.com/perfect-panel/server/pkg/trace"
|
||||
)
|
||||
|
||||
// bodyLogWriter is a wrapper for gin.ResponseWriter to capture response body
|
||||
type bodyLogWriter struct {
|
||||
gin.ResponseWriter
|
||||
body *bytes.Buffer
|
||||
}
|
||||
|
||||
func (w bodyLogWriter) Write(b []byte) (int, error) {
|
||||
w.body.Write(b)
|
||||
return w.ResponseWriter.Write(b)
|
||||
}
|
||||
|
||||
// statusByWriter returns a span status code and message for an HTTP status code
|
||||
// value returned by a server. Status codes in the 400-499 range are not
|
||||
// returned as errors.
|
||||
@ -59,6 +72,13 @@ func TraceMiddleware(_ *svc.ServiceContext) func(ctx *gin.Context) {
|
||||
ctx := c.Request.Context()
|
||||
tracer := trace.TracerFromContext(ctx)
|
||||
|
||||
// Capture Request Body
|
||||
var reqBody []byte
|
||||
if c.Request.Body != nil {
|
||||
reqBody, _ = io.ReadAll(c.Request.Body)
|
||||
c.Request.Body = io.NopCloser(bytes.NewBuffer(reqBody)) // Restore body
|
||||
}
|
||||
|
||||
spanName := c.FullPath()
|
||||
method := c.Request.Method
|
||||
|
||||
@ -78,13 +98,39 @@ func TraceMiddleware(_ *svc.ServiceContext) func(ctx *gin.Context) {
|
||||
attribute.String("http.request_id", requestId),
|
||||
semconv.HTTPRouteKey.String(c.FullPath()),
|
||||
)
|
||||
|
||||
// Record Request Body (limit to 1MB)
|
||||
if len(reqBody) > 0 {
|
||||
limit := 1048576
|
||||
if len(reqBody) > limit {
|
||||
span.SetAttributes(attribute.String("http.request.body", string(reqBody[:limit])+"...(truncated)"))
|
||||
} else {
|
||||
span.SetAttributes(attribute.String("http.request.body", string(reqBody)))
|
||||
}
|
||||
}
|
||||
|
||||
// context with request host
|
||||
ctx = context.WithValue(ctx, constant.CtxKeyRequestHost, c.Request.Host)
|
||||
// restructure context
|
||||
c.Request = c.Request.WithContext(ctx)
|
||||
|
||||
// Wrap ResponseWriter to capture Response Body
|
||||
blw := &bodyLogWriter{body: bytes.NewBufferString(""), ResponseWriter: c.Writer}
|
||||
c.Writer = blw
|
||||
|
||||
c.Next()
|
||||
|
||||
// Record Response Body (limit to 1MB)
|
||||
respBody := blw.body.String()
|
||||
if len(respBody) > 0 {
|
||||
limit := 1048576
|
||||
if len(respBody) > limit {
|
||||
span.SetAttributes(attribute.String("http.response.body", respBody[:limit]+"...(truncated)"))
|
||||
} else {
|
||||
span.SetAttributes(attribute.String("http.response.body", respBody))
|
||||
}
|
||||
}
|
||||
|
||||
// handle response related attributes
|
||||
status := c.Writer.Status()
|
||||
span.SetStatus(statusByWriter(status))
|
||||
@ -97,7 +143,5 @@ func TraceMiddleware(_ *svc.ServiceContext) func(ctx *gin.Context) {
|
||||
span.RecordError(err.Err)
|
||||
}
|
||||
}
|
||||
|
||||
span.SetAttributes(semconv.HTTPResponseBodySizeKey.Int(c.Writer.Size()))
|
||||
}
|
||||
}
|
||||
|
||||
@ -5,6 +5,8 @@ import (
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/perfect-panel/server/pkg/constant"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
@ -75,18 +77,36 @@ func (m *defaultUserModel) FindUsersSubscribeBySubscribeId(ctx context.Context,
|
||||
func (m *defaultUserModel) QueryUserSubscribe(ctx context.Context, userId int64, status ...int64) ([]*SubscribeDetails, error) {
|
||||
var list []*SubscribeDetails
|
||||
key := fmt.Sprintf("%s%d", cacheUserSubscribeUserPrefix, userId)
|
||||
err := m.QueryCtx(ctx, &list, key, func(conn *gorm.DB, v interface{}) error {
|
||||
// 获取当前时间
|
||||
now := time.Now()
|
||||
// 获取当前时间向前推 7 天
|
||||
sevenDaysAgo := time.Now().Add(-7 * 24 * time.Hour)
|
||||
// 基础条件查询
|
||||
conn = conn.Model(&Subscribe{}).Where("`user_id` = ?", userId)
|
||||
if len(status) > 0 {
|
||||
conn = conn.Where("`status` IN ?", status)
|
||||
|
||||
// 1. Get includeExpired from Context
|
||||
includeExpired := ""
|
||||
if v := ctx.Value(constant.CtxKeyIncludeExpired); v != nil {
|
||||
includeExpired, _ = v.(string)
|
||||
}
|
||||
// 订阅过期时间大于当前时间或者订阅结束时间大于当前时间
|
||||
return conn.Where("`expire_time` > ? OR `finished_at` >= ? OR `expire_time` = ?", now, sevenDaysAgo, time.UnixMilli(0)).
|
||||
|
||||
// 2. If query mode is different, must modify Cache Key
|
||||
if includeExpired == "all" {
|
||||
key += ":all"
|
||||
}
|
||||
|
||||
err := m.QueryCtx(ctx, &list, key, func(conn *gorm.DB, v interface{}) error {
|
||||
// Base condition
|
||||
db := conn.Model(&Subscribe{}).Where("`user_id` = ?", userId)
|
||||
if len(status) > 0 {
|
||||
db = db.Where("`status` IN ?", status)
|
||||
}
|
||||
|
||||
// 3. Adjust SQL based on param
|
||||
if includeExpired == "all" {
|
||||
// Mode A: Query all history
|
||||
return db.Order("created_at DESC").Preload("Subscribe").Find(&list).Error
|
||||
}
|
||||
|
||||
// Mode B: Default only query valid subscriptions
|
||||
// Logic: ExpireTime > Now OR FinishedAt >= 7 days ago OR ExpireTime = 0 (Never expire)
|
||||
now := time.Now()
|
||||
sevenDaysAgo := now.Add(-7 * 24 * time.Hour)
|
||||
return db.Where("`expire_time` > ? OR `finished_at` >= ? OR `expire_time` = ?", now, sevenDaysAgo, time.UnixMilli(0)).
|
||||
Preload("Subscribe").
|
||||
Find(&list).Error
|
||||
})
|
||||
|
||||
@ -56,6 +56,10 @@ func parseDefaultValue(kind reflect.Kind, defaultValue string) any {
|
||||
var i uint32
|
||||
_, _ = fmt.Sscanf(defaultValue, "%d", &i)
|
||||
return i
|
||||
case reflect.Float64:
|
||||
var f float64
|
||||
_, _ = fmt.Sscanf(defaultValue, "%f", &f)
|
||||
return f
|
||||
default:
|
||||
fmt.Printf("类型 %v 没有处理, 值为: %v \n", kind, defaultValue)
|
||||
panic("unhandled default case")
|
||||
|
||||
@ -9,4 +9,5 @@ const (
|
||||
CtxKeyPlatform CtxKey = "platform"
|
||||
CtxKeyPayment CtxKey = "payment"
|
||||
LoginType CtxKey = "loginType"
|
||||
CtxKeyIncludeExpired CtxKey = "includeExpired"
|
||||
)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user