Files
opsx d4422f1541
CI / Backend (go vet + test) (push) Canceled after 0s
CI / Frontend (lint + build + test) (push) Canceled after 0s
refactor(analysis): 去掉排行 10/30/50 选择器,改为一次取 50 + 前端分页;零点结算昨天
=== 1. 为什么去掉「显示条数」选择器 ===
topk 的 N 不影响查询成本(扫描的 chunk 集合完全相同),实测:
  topk(10)  17833ms / 11 条
  topk(30)  18019ms / 34 条
  topk(50)   7394ms / 54 条
  topk(100) 17553ms / 74 条
波动来自集群负载,与 N 无关 —— 它只决定最后返回多少序列。
所以"选 30"既不更快也不更慢,本质是个翻页开关;
而选它还要重新等一次十几秒的查询,纯亏。
改为:一次 topk(50),前端 10 条/页 → 5 页只需 1 次查询。
附带收益:占比的分母从"前 10 合计"变成"前 50 合计",更接近真实占比。
副标题明确写清"一次取前 50 个应用(占比按这 50 个合计计算)"。

=== 2. 零点结算昨天(daily volume)===
原来只有 30 分钟 ticker 回补,且启动时间不对齐整点 →
昨天最早的一段可能已经超出 Loki 24h 保留期被删,
结算结果系统性偏低,图上却标"完整一天",属静默错数。
新增 nextMidnightFinalize():每天 00:01 触发 FinalizeYesterdayVolume()
  - 昨天定稿 partial=false
  - 补一次条数(index/stats,一天一次可接受)
  - 顺带刷新今天 partial=true 的柱子
ticker 仍保留,负责当天数据的 30 分钟刷新。

=== 3. 顺带 ===
- 去掉每日量"其中 N 天平台没有留存记录"的提示文案(按用户要求)
- 修 QuickQueryTemplates.tsx 的未使用变量导致 tsc -b 失败
- 工作区里上一轮中断的"排行体积列"改动一并纳入(app-ranking 返回 bytes,
  TopAppsTable 增加体积列,快照记录 per-app 字节数)

Pod = Hub latest = a16978d2f88b
2026-10-30 17:13:18 +00:00

512 lines
16 KiB
Go
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package controllers
import (
"context"
"errors"
"net/http"
"strconv"
"strings"
"time"
"lokistack/dto"
"lokistack/models"
"lokistack/services"
"lokistack/utils"
"github.com/gin-gonic/gin"
)
// 上下文前后条数的默认值与上限(原来是硬编码 5)
const (
DefaultContextLines = 2
MaxContextLines = 50
)
// normalizeContextLines 把前端传入的条数收敛到 [1, MaxContextLines],
// 不传(0)或负数一律用默认值。
func normalizeContextLines(n int) int {
if n <= 0 {
return DefaultContextLines
}
if n > MaxContextLines {
return MaxContextLines
}
return n
}
func (ctrl *LogsController) Context(c *gin.Context) {
var req dto.LogContextRequest
if !utils.BindAndValidate(c, &req) {
return
}
src := services.GetLogSource(req.SourceID)
if src == nil {
utils.BadRequest(c, "数据源不存在")
return
}
ts, err := strconv.ParseInt(req.TimestampNs, 10, 64)
if err != nil || ts < 1 {
utils.BadRequest(c, "时间戳无效")
return
}
selector, err := services.ContextLogSelector(req.Labels)
if err != nil {
utils.BadRequest(c, err.Error())
return
}
query, err := services.AuthorizedLogQuery(src, c.GetUint("userId"), c.GetString("role") == "admin", selector)
if err != nil {
utils.Forbidden(c, err.Error())
return
}
// 前后条数由前端传入(默认 2 条,上限 50 —— 再多一次弹窗就是几百条日志)
beforeN := normalizeContextLines(req.Before)
afterN := normalizeContextLines(req.After)
// 时间窗随条数自适应:原来固定 5 分钟,条数调大时 5 分钟可能装不下,
// 条数调小时又白扫一大片。按「每条 30s」估算,最少 1 分钟,最多 2 小时。
span := int64(time.Duration(max(beforeN, afterN)*30) * time.Second)
if span < int64(time.Minute) {
span = int64(time.Minute)
}
if span > int64(2*time.Hour) {
span = int64(2 * time.Hour)
}
before, err := services.QueryLokiLogs(c.Request.Context(), src, query, ts-span, ts-1, beforeN, "backward", "")
if err != nil {
utils.Error(c, 502, err.Error())
return
}
after, err := services.QueryLokiLogs(c.Request.Context(), src, query, ts+1, ts+span, afterN, "forward", "")
if err != nil {
utils.Error(c, 502, err.Error())
return
}
utils.Success(c, gin.H{"before": before.Entries, "after": after.Entries})
}
// LogsController 日志查询代理 + 数据源管理
type LogsController struct{}
func NewLogsController() *LogsController {
return &LogsController{}
}
// ==================== 数据源 ====================
func (ctrl *LogsController) ListSources(c *gin.Context) {
if c.GetString("role") == "admin" {
services.EnsureLogSourcesForRunningDeployments()
}
list := services.ListVisibleLogSources(c.GetUint("userId"), c.GetString("role") == "admin")
utils.Success(c, utils.PageData{List: list, Total: int64(len(list)), Page: 1, PageSize: 100})
}
func (ctrl *LogsController) CreateSource(c *gin.Context) {
var req dto.LogSourceRequest
if !utils.BindAndValidate(c, &req) {
return
}
if req.Name == "" || req.TargetType == "" {
utils.BadRequest(c, "名称与类型必填")
return
}
if err := services.ValidateSourceAccess(req.Access); err != nil {
utils.BadRequest(c, err.Error())
return
}
src := &models.LogSource{
Name: req.Name, TargetType: req.TargetType,
Namespace: req.Namespace, Service: req.Service, Port: req.Port,
URL: req.URL, AuthType: req.AuthType, BasicUser: req.BasicUser,
BasicPass: req.BasicPass, BearerToken: req.BearerToken, Headers: req.Headers, SkipTLSVerify: req.SkipTLSVerify, IsDefault: req.IsDefault, Access: req.Access,
}
if req.TargetType == "cluster-svc" {
if req.ClusterID > 0 {
id := req.ClusterID
src.ClusterID = &id
} else {
utils.BadRequest(c, "cluster-svc 类型必须选择集群")
return
}
}
created, err := services.CreateLogSource(src)
if err != nil {
utils.Error(c, 400, err.Error())
return
}
utils.Success(c, created.Sanitized())
}
func (ctrl *LogsController) UpdateSource(c *gin.Context) {
id, err := strconv.ParseUint(c.Param("id"), 10, 64)
if err != nil {
utils.BadRequest(c, "ID 无效")
return
}
var updates map[string]interface{}
if !utils.BindAndValidate(c, &updates) {
return
}
if src := services.GetLogSource(uint(id)); src != nil && src.Managed && updates["isDefault"] == nil {
utils.BadRequest(c, "平台部署的数据源由系统管理,不可编辑")
return
}
if err := services.UpdateLogSource(uint(id), updates); err != nil {
utils.Error(c, 400, err.Error())
return
}
utils.Success(c, gin.H{"message": "updated"})
}
func (ctrl *LogsController) DeleteSource(c *gin.Context) {
id, err := strconv.ParseUint(c.Param("id"), 10, 64)
if err != nil {
utils.BadRequest(c, "ID 无效")
return
}
// 托管数据源也允许删除:用户可能只是不想让它出现在查询页下拉里。
// 删除后把对应部署的 AutoSourceDisabled 置位,否则列表接口/启动同步
// 会立刻把它补回来(用户看到的现象就是"删了还在")。
if src := services.GetLogSource(uint(id)); src != nil && src.Managed {
services.DisableAutoSourceFor(src.Name)
}
if err := services.DeleteLogSource(uint(id)); err != nil {
utils.Error(c, 400, err.Error())
return
}
utils.Success(c, gin.H{"message": "deleted"})
}
// TestSource 数据源连通性测试(按 id 或表单参数)
func (ctrl *LogsController) TestSource(c *gin.Context) {
var req dto.LogSourceRequest
if !utils.BindAndValidate(c, &req) {
return
}
var src *models.LogSource
if req.ID > 0 {
src = services.GetLogSource(req.ID)
if src == nil {
utils.NotFound(c, "数据源不存在")
return
}
} else {
src = &models.LogSource{
TargetType: req.TargetType, Namespace: req.Namespace, Service: req.Service,
Port: req.Port, URL: req.URL, AuthType: req.AuthType,
BasicUser: req.BasicUser, BasicPass: req.BasicPass, BearerToken: req.BearerToken, Headers: req.Headers, SkipTLSVerify: req.SkipTLSVerify,
}
if req.TargetType == "cluster-svc" && req.ClusterID > 0 {
id := req.ClusterID
src.ClusterID = &id
}
}
utils.Success(c, services.TestLogSource(c.Request.Context(), src))
}
// ==================== 查询代理 ====================
func (ctrl *LogsController) Query(c *gin.Context) {
var req dto.LogQueryRequest
if !utils.BindAndValidate(c, &req) {
return
}
src := services.GetLogSource(req.SourceID)
if src == nil {
utils.BadRequest(c, "数据源不存在")
return
}
query, err := services.AuthorizedLogQuery(src, c.GetUint("userId"), c.GetString("role") == "admin", req.Query)
if err != nil {
utils.Forbidden(c, err.Error())
return
}
result, err := services.QueryLokiLogs(c.Request.Context(), src, query,
req.Start, req.End, req.Limit, req.Direction, req.Cursor)
if err != nil {
utils.Error(c, 502, err.Error())
return
}
utils.Success(c, result)
}
func (ctrl *LogsController) Labels(c *gin.Context) {
id, _ := strconv.ParseUint(c.Query("sourceId"), 10, 64)
src := services.GetLogSource(uint(id))
if src == nil {
utils.BadRequest(c, "数据源不存在")
return
}
if _, err := services.AuthorizedLogQuery(src, c.GetUint("userId"), c.GetString("role") == "admin", "{}"); err != nil {
utils.Forbidden(c, err.Error())
return
}
start, _ := strconv.ParseInt(c.Query("start"), 10, 64)
end, _ := strconv.ParseInt(c.Query("end"), 10, 64)
labels, err := services.LokiLabels(c.Request.Context(), src, start, end)
if err != nil {
utils.Error(c, 502, err.Error())
return
}
utils.Success(c, labels)
}
func (ctrl *LogsController) LabelValues(c *gin.Context) {
id, _ := strconv.ParseUint(c.Query("sourceId"), 10, 64)
src := services.GetLogSource(uint(id))
if src == nil {
utils.BadRequest(c, "数据源不存在")
return
}
start, _ := strconv.ParseInt(c.Query("start"), 10, 64)
end, _ := strconv.ParseInt(c.Query("end"), 10, 64)
requestedQuery := c.Query("query")
if strings.TrimSpace(requestedQuery) == "" {
requestedQuery = "{}"
}
query, err := services.AuthorizedLogQuery(src, c.GetUint("userId"), c.GetString("role") == "admin", requestedQuery)
if err != nil {
utils.Forbidden(c, err.Error())
return
}
values, err := services.LokiLabelValues(c.Request.Context(), src,
c.Param("name"), query, start, end)
if err != nil {
utils.Error(c, 502, err.Error())
return
}
utils.Success(c, values)
}
// MyScope 返回当前登录用户在某数据源上的授权范围。
//
// 快捷 LogQL 用它做两件事:
// 1. 把授权策略里配置的 namespace 作为默认选中值(管理员配了 default/astralx 后开箱即用)
// 2. 把该值合并进 namespace 候选列表 —— 否则像 default 这种「策略里有、Loki 里没日志」
// 的命名空间不会出现在下拉里,用户会以为策略没生效
func (ctrl *LogsController) MyScope(c *gin.Context) {
id, _ := strconv.ParseUint(c.Query("sourceId"), 10, 64)
src := services.GetLogSource(uint(id))
if src == nil {
utils.BadRequest(c, "数据源不存在")
return
}
if c.GetString("role") == "admin" {
// 管理员的查询范围不受策略限制(unrestricted=true),
// 但仍返回策略里的 labels 作为快捷 LogQL 的默认选中值——
// 否则管理员在小黑框里看到的下拉永远没有策略配的那个 namespace,
// 会误以为「默认策略没生效」。
labels := services.UserAccessLabels(src, c.GetUint("userId"))
if labels == nil {
labels = map[string]string{}
}
utils.Success(c, gin.H{"labels": labels, "unrestricted": true})
return
}
labels := services.UserAccessLabels(src, c.GetUint("userId"))
if labels == nil {
utils.Forbidden(c, "无权查询此数据源")
return
}
utils.Success(c, gin.H{"labels": labels, "unrestricted": len(labels) == 0})
}
// Health Loki /ready 探活(查询页状态点)
func (ctrl *LogsController) Health(c *gin.Context) {
id, _ := strconv.ParseUint(c.Query("sourceId"), 10, 64)
src := services.GetLogSource(uint(id))
if src == nil {
utils.BadRequest(c, "数据源不存在")
return
}
if _, err := services.AuthorizedLogQuery(src, c.GetUint("userId"), c.GetString("role") == "admin", "{}"); err != nil {
utils.Forbidden(c, err.Error())
return
}
ctx, cancel := context.WithTimeout(c.Request.Context(), 15*time.Second)
defer cancel()
utils.Success(c, gin.H{"ready": services.LokiReady(ctx, src)})
}
// AppRanking 按应用排行(长范围走平台预聚合,毫秒级;短范围由前端走实时查询)。
func (ctrl *LogsController) AppRanking(c *gin.Context) {
sourceID, err := strconv.ParseUint(c.Query("sourceId"), 10, 64)
if err != nil || sourceID == 0 {
utils.BadRequest(c, "sourceId 无效")
return
}
from, err1 := strconv.ParseInt(c.DefaultQuery("from", "0"), 10, 64)
to, err2 := strconv.ParseInt(c.DefaultQuery("to", "0"), 10, 64)
if err1 != nil || err2 != nil || from <= 0 || to <= 0 || to <= from {
utils.BadRequest(c, "时间范围无效(from/to 需为纳秒时间戳且 to > from)")
return
}
limit := 10
if raw := c.Query("limit"); raw != "" {
if v, convErr := strconv.Atoi(raw); convErr == nil && v > 0 && v <= 100 {
limit = v
}
}
rows := services.GetAppRanking(uint(sourceID), time.Unix(0, from), time.Unix(0, to), limit, c.DefaultQuery("sort", "bytes"))
hours, bytesHours := services.AppVolumeCoverageInfo(uint(sourceID))
utils.Success(c, gin.H{"list": rows, "hours": hours, "bytesHours": bytesHours})
}
// LabelValuesBatch 一次取多个 label 的候选值:GET /logs/label-values?names=app,pod&query={...}
//
// 快捷查询的下拉是级联的(命名空间 → 应用 → Pod),而"应用"本身要走
// app → service_name → container 三级回退。逐个请求时前端一次要发 3 个往返,
// 每个都要经 K8s API 代理到 Loki 索引;合并成一个接口后只发一次。
// 返回 {"app": [...], "service_name": [...], ...},单个 label 失败返回空数组不报错。
func (ctrl *LogsController) LabelValuesBatch(c *gin.Context) {
id, _ := strconv.ParseUint(c.Query("sourceId"), 10, 64)
src := services.GetLogSource(uint(id))
if src == nil {
utils.BadRequest(c, "数据源不存在")
return
}
raw := strings.TrimSpace(c.Query("names"))
if raw == "" {
utils.BadRequest(c, "缺少 names(逗号分隔的 label 名,如 app,pod)")
return
}
names := make([]string, 0, 4)
for _, name := range strings.Split(raw, ",") {
if name = strings.TrimSpace(name); name != "" {
names = append(names, name)
}
}
if len(names) == 0 {
utils.BadRequest(c, "names 里没有有效的 label 名")
return
}
if len(names) > 10 {
utils.BadRequest(c, "一次最多取 10 个 label 的值")
return
}
requestedQuery := strings.TrimSpace(c.Query("query"))
if requestedQuery == "" {
requestedQuery = "{}"
}
query, err := services.AuthorizedLogQuery(src, c.GetUint("userId"), c.GetString("role") == "admin", requestedQuery)
if err != nil {
utils.Forbidden(c, err.Error())
return
}
start, _ := strconv.ParseInt(c.Query("start"), 10, 64)
end, _ := strconv.ParseInt(c.Query("end"), 10, 64)
values, err := services.LokiLabelValuesBatch(c.Request.Context(), src, names, query, start, end)
if err != nil {
utils.Error(c, 502, err.Error())
return
}
utils.Success(c, values)
}
// DailyVolume 每日日志量历史(平台留存,不受 Loki retention 限制)。
//
// query: sourceId, days(默认 7,最长 400)
func (ctrl *LogsController) DailyVolume(c *gin.Context) {
sourceID, _ := strconv.Atoi(c.Query("sourceId"))
if sourceID <= 0 {
utils.BadRequest(c, "缺少 sourceId")
return
}
days, _ := strconv.Atoi(c.Query("days"))
if days <= 0 {
days = 7
}
utils.Success(c, services.GetDailyVolumeHistory(uint(sourceID), days))
}
// Stats 返回该查询在时间范围内的「总条数/字节/块数」,用于前端展示总页数。
// 走 index/stats(只读索引),成本远低于真数日志。
func (ctrl *LogsController) Stats(c *gin.Context) {
var req dto.LogQueryRequest
if !utils.BindAndValidate(c, &req) {
return
}
src := services.GetLogSource(req.SourceID)
if src == nil {
utils.BadRequest(c, "数据源不存在")
return
}
query, err := services.AuthorizedLogQuery(src, c.GetUint("userId"), c.GetString("role") == "admin", req.Query)
if err != nil {
utils.Forbidden(c, err.Error())
return
}
st, err := services.LokiStatsCached(c.Request.Context(), src, query, req.Start, req.End)
if err != nil {
// Loki 的 index/stats 只支持标签选择器。查询里带 `|= "error"` 这类行过滤时
// Loki 回 400 "only label matchers are supported",原来被包装成 502
// ("Kubernetes API 代理返回 HTTP 400…"),前端只能当失败静默处理,
// 用户看到的就是"没有总页数"。这里改成 400 + 人话,前端据此显示
// "含过滤条件时无法统计总页数",而不是假装统计不出来。
if strings.Contains(err.Error(), "only label matchers") {
utils.Error(c, 400, "含过滤条件(|= / |~)时无法统计总页数:Loki 只支持按标签统计")
return
}
utils.Error(c, 502, err.Error())
return
}
utils.Success(c, st)
}
func (ctrl *LogsController) Histogram(c *gin.Context) {
var req dto.LogQueryRequest
if !utils.BindAndValidate(c, &req) {
return
}
src := services.GetLogSource(req.SourceID)
if src == nil {
utils.BadRequest(c, "数据源不存在")
return
}
query, err := services.AuthorizedLogQuery(src, c.GetUint("userId"), c.GetString("role") == "admin", req.Query)
if err != nil {
utils.Forbidden(c, err.Error())
return
}
buckets, err := services.LokiHistogram(c.Request.Context(), src, query, req.Start, req.End)
if err != nil {
switch {
case errors.Is(err, services.ErrTooManyQueries):
// 429:并发重查询超限,让前端提示"稍后重试",而不是当成服务故障
utils.Error(c, http.StatusTooManyRequests, err.Error())
case strings.Contains(err.Error(), "统计范围过大"):
utils.Error(c, http.StatusBadRequest, err.Error())
default:
utils.Error(c, 502, err.Error())
}
return
}
utils.Success(c, buckets)
}
// Metrics 日志分析指标查询(仅 admin,路由层 AdminAuth 拦截)
func (ctrl *LogsController) Metrics(c *gin.Context) {
var req dto.LogMetricQueryRequest
if !utils.BindAndValidate(c, &req) {
return
}
src := services.GetLogSource(req.SourceID)
if src == nil {
utils.BadRequest(c, "数据源不存在")
return
}
query, err := services.AuthorizedLogQuery(src, c.GetUint("userId"), c.GetString("role") == "admin", req.Query)
if err != nil {
utils.Forbidden(c, err.Error())
return
}
series, err := services.LokiMetrics(c.Request.Context(), src, query, req.Start, req.End)
if err != nil {
utils.Error(c, 502, err.Error())
return
}
utils.Success(c, series)
}