snapshot: 数据源管理+大屏增强前快照

This commit is contained in:
attacker committed 2026-06-18 14:01:22 +00:00
1 parent 2bf3e7c308
commit 1726ad65ac
19 files changed
+2540 -273

No files matched your search

+37 -14
View File
@@ -87,13 +87,15 @@ func (h *AgentHandler) ReceiveReport(c *gin.Context) {
result := h.db.Model(&models.Server{}).
Where("hostname = ? OR ip = ?", req.Hostname, req.Hostname).
Updates(map[string]interface{}{
"cpu_usage": req.CPU.UsagePercent,
"memory_usage": req.Memory.UsagePercent,
"disk_usage": diskAvg,
"cpu": req.CPU.CoreCount,
"memory": memMB,
"disk": int64(totalDiskGB),
"status": "running",
"cpu_usage": req.CPU.UsagePercent,
"memory_usage": req.Memory.UsagePercent,
"disk_usage": diskAvg,
"cpu": req.CPU.CoreCount,
"memory": memMB,
"disk": int64(totalDiskGB),
"status": "running",
"agent_status": "online",
"agent_version": req.AgentVersion,
})
// WebSocket 广播:向所有前端客户端推送该主机的最新指标 // Logan.Li
@@ -155,20 +157,41 @@ func (h *AgentHandler) writeHotMetric(c *gin.Context, key string, value float64,
// GetLatestReports 获取各主机最新上报
// GET /api/v1/agent/latest
func (h *AgentHandler) GetLatestReports(c *gin.Context) {
// 子查询:每个 hostname 最新的一条记录
var reports []models.AgentReport
var results []models.AgentLatestResponse
h.db.Raw(`
SELECT * FROM agent_reports
WHERE id IN (
SELECT
a.hostname,
COALESCE(s.ip, '') as ip,
a.cpu_usage,
a.cpu_cores,
a.mem_usage,
a.mem_total,
a.mem_used,
COALESCE(s.disk_usage, 0) as disk_usage,
a.disk_info,
a.net_info,
a.report_time,
COALESCE(s.agent_version, '') as agent_version,
COALESCE(s.agent_status, 'offline') as agent_status,
COALESCE(s.os, '') as os,
COALESCE(r.name, '') as region
FROM agent_reports a
LEFT JOIN servers s ON s.hostname = a.hostname
LEFT JOIN regions r ON r.id = s.region_id
WHERE a.id IN (
SELECT MAX(id) FROM agent_reports GROUP BY hostname
)
ORDER BY report_time DESC
`).Scan(&reports)
ORDER BY a.report_time DESC
`).Scan(&results)
if results == nil {
results = []models.AgentLatestResponse{}
}
c.JSON(http.StatusOK, models.Response{
Code: 200,
Message: "success",
Data: reports,
Data: results,
})
}
@@ -0,0 +1,352 @@
package handlers
import (
"crypto/tls"
"encoding/base64"
"encoding/json"
"net/http"
"time"
"ezops/internal/models"
"github.com/gin-gonic/gin"
"gorm.io/gorm"
)
// PromDSHandler Prometheus 数据源管理 handler // Logan.Li
type PromDSHandler struct {
db *gorm.DB
}
// NewPromDSHandler 创建 Prometheus 数据源 handler
func NewPromDSHandler(db *gorm.DB) *PromDSHandler {
return &PromDSHandler{db: db}
}
// ============================================
// CRUD
// ============================================
// List 数据源列表
// GET /api/v1/prometheus-datasources
func (h *PromDSHandler) List(c *gin.Context) {
var sources []models.PrometheusDataSource
if err := h.db.Order("is_default DESC, id ASC").Find(&sources).Error; err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "查询失败: " + err.Error()})
return
}
if sources == nil {
sources = []models.PrometheusDataSource{}
}
c.JSON(http.StatusOK, models.Response{Code: 200, Data: sources})
}
// Get 获取单个数据源
// GET /api/v1/prometheus-datasources/:id
func (h *PromDSHandler) Get(c *gin.Context) {
id := c.Param("id")
var ds models.PrometheusDataSource
if err := h.db.First(&ds, id).Error; err != nil {
c.JSON(http.StatusNotFound, models.Response{Code: 404, Message: "数据源不存在"})
return
}
c.JSON(http.StatusOK, models.Response{Code: 200, Data: ds})
}
// Create 创建数据源
// POST /api/v1/prometheus-datasources
func (h *PromDSHandler) Create(c *gin.Context) {
var req models.CreatePromDSReq
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误: " + err.Error()})
return
}
// 如果设为默认,先清除其他默认
if req.IsDefault {
h.db.Model(&models.PrometheusDataSource{}).Where("is_default = ?", true).Update("is_default", false)
}
// 默认值
if req.Access == "" {
req.Access = "server"
}
if req.HTTPMethod == "" {
req.HTTPMethod = "POST"
}
if req.ScrapeInterval <= 0 {
req.ScrapeInterval = 15
}
if req.QueryTimeout <= 0 {
req.QueryTimeout = 60
}
enableAlerting := true
if req.EnableAlerting != nil {
enableAlerting = *req.EnableAlerting
}
ds := models.PrometheusDataSource{
Name: req.Name,
URL: req.URL,
Access: req.Access,
BasicAuth: req.BasicAuth,
BasicAuthUser: req.BasicAuthUser,
BasicAuthPass: req.BasicAuthPass,
WithCredentials: req.WithCredentials,
TLSClientAuth: req.TLSClientAuth,
WithCACert: req.WithCACert,
CACert: req.CACert,
SkipTLSVerify: req.SkipTLSVerify,
CustomHeaders: req.CustomHeaders,
EnableAlerting: enableAlerting,
AlertmanagerDS: req.AlertmanagerDS,
ScrapeInterval: req.ScrapeInterval,
QueryTimeout: req.QueryTimeout,
HTTPMethod: req.HTTPMethod,
DisableMetricsLookup: req.DisableMetricsLookup,
CustomQueryParams: req.CustomQueryParams,
IsDefault: req.IsDefault,
Status: 1,
}
if err := h.db.Create(&ds).Error; err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "创建失败: " + err.Error()})
return
}
h.recordAudit(c, "create", "prometheus_datasource", ds.Name)
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "创建成功", Data: ds})
}
// Update 更新数据源
// PUT /api/v1/prometheus-datasources/:id
func (h *PromDSHandler) Update(c *gin.Context) {
id := c.Param("id")
var ds models.PrometheusDataSource
if err := h.db.First(&ds, id).Error; err != nil {
c.JSON(http.StatusNotFound, models.Response{Code: 404, Message: "数据源不存在"})
return
}
var req models.UpdatePromDSReq
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误: " + err.Error()})
return
}
updates := map[string]interface{}{}
if req.Name != nil {
updates["name"] = *req.Name
}
if req.URL != nil {
updates["url"] = *req.URL
}
if req.Access != nil {
updates["access"] = *req.Access
}
if req.BasicAuth != nil {
updates["basic_auth"] = *req.BasicAuth
}
if req.BasicAuthUser != nil {
updates["basic_auth_user"] = *req.BasicAuthUser
}
if req.BasicAuthPass != nil {
updates["basic_auth_pass"] = *req.BasicAuthPass
}
if req.WithCredentials != nil {
updates["with_credentials"] = *req.WithCredentials
}
if req.TLSClientAuth != nil {
updates["tls_client_auth"] = *req.TLSClientAuth
}
if req.WithCACert != nil {
updates["with_ca_cert"] = *req.WithCACert
}
if req.CACert != nil {
updates["ca_cert"] = *req.CACert
}
if req.SkipTLSVerify != nil {
updates["skip_tls_verify"] = *req.SkipTLSVerify
}
if req.CustomHeaders != nil {
updates["custom_headers"] = *req.CustomHeaders
}
if req.EnableAlerting != nil {
updates["enable_alerting"] = *req.EnableAlerting
}
if req.AlertmanagerDS != nil {
updates["alertmanager_ds"] = *req.AlertmanagerDS
}
if req.ScrapeInterval != nil {
updates["scrape_interval"] = *req.ScrapeInterval
}
if req.QueryTimeout != nil {
updates["query_timeout"] = *req.QueryTimeout
}
if req.HTTPMethod != nil {
updates["http_method"] = *req.HTTPMethod
}
if req.DisableMetricsLookup != nil {
updates["disable_metrics_lookup"] = *req.DisableMetricsLookup
}
if req.CustomQueryParams != nil {
updates["custom_query_params"] = *req.CustomQueryParams
}
if req.IsDefault != nil {
// 如果设为默认,先清除其他默认
if *req.IsDefault {
h.db.Model(&models.PrometheusDataSource{}).Where("is_default = ?", true).Update("is_default", false)
}
updates["is_default"] = *req.IsDefault
}
if req.Status != nil {
updates["status"] = *req.Status
}
if len(updates) == 0 {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "无更新字段"})
return
}
if err := h.db.Model(&ds).Updates(updates).Error; err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "更新失败: " + err.Error()})
return
}
// 重新查询返回
h.db.First(&ds, id)
h.recordAudit(c, "update", "prometheus_datasource", ds.Name)
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "更新成功", Data: ds})
}
// Delete 删除数据源
// DELETE /api/v1/prometheus-datasources/:id
func (h *PromDSHandler) Delete(c *gin.Context) {
id := c.Param("id")
var ds models.PrometheusDataSource
if err := h.db.First(&ds, id).Error; err != nil {
c.JSON(http.StatusNotFound, models.Response{Code: 404, Message: "数据源不存在"})
return
}
if err := h.db.Delete(&ds).Error; err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "删除失败: " + err.Error()})
return
}
h.recordAudit(c, "delete", "prometheus_datasource", ds.Name)
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "删除成功"})
}
// SetDefault 设为默认数据源
// POST /api/v1/prometheus-datasources/:id/default
func (h *PromDSHandler) SetDefault(c *gin.Context) {
id := c.Param("id")
var ds models.PrometheusDataSource
if err := h.db.First(&ds, id).Error; err != nil {
c.JSON(http.StatusNotFound, models.Response{Code: 404, Message: "数据源不存在"})
return
}
// 清除所有默认
h.db.Model(&models.PrometheusDataSource{}).Where("is_default = ?", true).Update("is_default", false)
// 设置当前为默认
h.db.Model(&ds).Update("is_default", true)
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "已设为默认"})
}
// ============================================
// Test Connection
// ============================================
// TestConnection 测试数据源连接
// POST /api/v1/prometheus-datasources/test
func (h *PromDSHandler) TestConnection(c *gin.Context) {
var req models.TestPromDSReq
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误: " + err.Error()})
return
}
client := &http.Client{Timeout: 15 * time.Second}
// TLS 配置
if req.SkipTLSVerify {
client.Transport = &http.Transport{
TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
}
}
// 构造请求
apiURL := req.URL + "/api/v1/query?query=up&time=" + time.Now().UTC().Format(time.RFC3339)
httpReq, err := http.NewRequest("GET", apiURL, nil)
if err != nil {
c.JSON(http.StatusOK, models.Response{Code: 400, Data: gin.H{
"success": false, "message": "URL 格式错误: " + err.Error(),
}})
return
}
// Basic Auth
if req.BasicAuth {
auth := base64.StdEncoding.EncodeToString([]byte(req.BasicAuthUser + ":" + req.BasicAuthPass))
httpReq.Header.Set("Authorization", "Basic "+auth)
}
resp, err := client.Do(httpReq)
if err != nil {
c.JSON(http.StatusOK, models.Response{Code: 200, Data: gin.H{
"success": false, "message": "连接失败: " + err.Error(),
}})
return
}
defer resp.Body.Close()
if resp.StatusCode >= 200 && resp.StatusCode < 300 {
// 尝试解析响应获取 target 数量
var result struct {
Data struct {
Result []interface{} `json:"result"`
} `json:"data"`
}
json.NewDecoder(resp.Body).Decode(&result)
c.JSON(http.StatusOK, models.Response{Code: 200, Data: gin.H{
"success": true,
"message": "连接成功",
"status_code": resp.StatusCode,
"target_count": len(result.Data.Result),
}})
} else {
c.JSON(http.StatusOK, models.Response{Code: 200, Data: gin.H{
"success": false,
"message": "连接返回异常状态码",
"status_code": resp.StatusCode,
}})
}
}
// ============================================
// 审计
// ============================================
func (h *PromDSHandler) recordAudit(c *gin.Context, action, resource, name string) {
userID, _ := c.Get("userID")
username, _ := c.Get("username")
var uid uint
var uname string
if v, ok := userID.(uint); ok {
uid = v
}
if v, ok := username.(string); ok {
uname = v
}
h.db.Create(&models.AuditLog{
UserID: &uid,
Username: uname,
IP: c.ClientIP(),
UserAgent: c.Request.UserAgent(),
Module: "prometheus_datasource",
Action: action,
Resource: resource,
ResourceID: name,
Status: 200,
})
}
@@ -0,0 +1,331 @@
package handlers
import (
"encoding/json"
"fmt"
"io"
"math"
"net/http"
"net/url"
"strconv"
"strings"
"time"
"ezops/internal/models"
"github.com/gin-gonic/gin"
)
// PrometheusProxyHandler Prometheus API 代理 — 转发 PromQL 查询到 Prometheus 服务器
type PrometheusProxyHandler struct{}
func NewPrometheusProxyHandler() *PrometheusProxyHandler {
return &PrometheusProxyHandler{}
}
// PromConnectReq 连接 Prometheus 请求
type PromConnectReq struct {
URL string `json:"url" binding:"required"`
}
// PromConnect 连接 Prometheus 并返回概览数据
// POST /api/v1/prometheus/connect
func (h *PrometheusProxyHandler) Connect(c *gin.Context) {
var req PromConnectReq
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误: " + err.Error()})
return
}
client := &http.Client{Timeout: 15 * time.Second}
// 测试连接:查询 up 指标
status, targetCount := h.testConnect(client, req.URL)
if !status {
c.JSON(http.StatusBadGateway, models.Response{Code: 502, Message: "无法连接到 Prometheus,请检查 URL"})
return
}
c.JSON(http.StatusOK, models.Response{
Code: 200,
Message: "success",
Data: gin.H{
"status": "connected",
"targets": targetCount,
},
})
}
// PromQuery 执行 PromQL 即时查询
// POST /api/v1/prometheus/query
func (h *PrometheusProxyHandler) Query(c *gin.Context) {
var req struct {
URL string `json:"url" binding:"required"`
Query string `json:"query" binding:"required"`
}
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误"})
return
}
client := &http.Client{Timeout: 15 * time.Second}
result, err := h.execPromQL(client, req.URL, req.Query)
if err != nil {
c.JSON(http.StatusBadGateway, models.Response{Code: 502, Message: "查询失败: " + err.Error()})
return
}
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "success", Data: result})
}
// PromQueryRange 执行 PromQL 范围查询
// POST /api/v1/prometheus/query_range
func (h *PrometheusProxyHandler) QueryRange(c *gin.Context) {
var req struct {
URL string `json:"url" binding:"required"`
Query string `json:"query" binding:"required"`
Start string `json:"start"` // RFC3339
End string `json:"end"`
Step string `json:"step"` // e.g. "15s", "1m"
}
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误"})
return
}
if req.Step == "" {
req.Step = "1m"
}
client := &http.Client{Timeout: 30 * time.Second}
u := fmt.Sprintf("%s/api/v1/query_range?query=%s", strings.TrimRight(req.URL, "/"),
url.QueryEscape(req.Query))
if req.Start != "" {
u += "&start=" + url.QueryEscape(req.Start)
}
if req.End != "" {
u += "&end=" + url.QueryEscape(req.End)
}
u += "&step=" + url.QueryEscape(req.Step)
resp, err := client.Get(u)
if err != nil {
c.JSON(http.StatusBadGateway, models.Response{Code: 502, Message: "查询失败: " + err.Error()})
return
}
defer resp.Body.Close()
body, _ := io.ReadAll(resp.Body)
var promResult map[string]interface{}
json.Unmarshal(body, &promResult)
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "success", Data: promResult})
}
// PromDashboard 获取预定义仪表盘数据(K8s / Node / Pod / JVM)
// POST /api/v1/prometheus/dashboard
func (h *PrometheusProxyHandler) Dashboard(c *gin.Context) {
var req struct {
URL string `json:"url" binding:"required"`
Type string `json:"type" binding:"required"` // k8s, node, pod, jvm
}
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误"})
return
}
client := &http.Client{Timeout: 30 * time.Second}
var result map[string]interface{}
switch req.Type {
case "node":
result = h.dashboardNode(client, req.URL)
case "pod":
result = h.dashboardPod(client, req.URL)
case "jvm":
result = h.dashboardJVM(client, req.URL)
default:
result = h.dashboardK8s(client, req.URL)
}
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "success", Data: result})
}
// ============================================
// 内部方法
// ============================================
func (h *PrometheusProxyHandler) testConnect(client *http.Client, baseURL string) (bool, int) {
result, err := h.execPromQL(client, baseURL, "count(up)")
if err != nil {
return false, 0
}
targets := 0
if data, ok := result["data"].(map[string]interface{}); ok {
if res, ok := data["result"].([]interface{}); ok {
if len(res) > 0 {
if val, ok := res[0].(map[string]interface{})["value"].([]interface{}); ok && len(val) > 1 {
targets, _ = strconv.Atoi(fmt.Sprintf("%v", val[1]))
}
}
}
}
return true, targets
}
func (h *PrometheusProxyHandler) execPromQL(client *http.Client, baseURL, query string) (map[string]interface{}, error) {
u := fmt.Sprintf("%s/api/v1/query?query=%s", strings.TrimRight(baseURL, "/"), url.QueryEscape(query))
resp, err := client.Get(u)
if err != nil {
return nil, err
}
defer resp.Body.Close()
body, _ := io.ReadAll(resp.Body)
var result map[string]interface{}
if err := json.Unmarshal(body, &result); err != nil {
return nil, err
}
if status, ok := result["status"].(string); !ok || status != "success" {
return nil, fmt.Errorf("prometheus error: %v", result)
}
return result, nil
}
func parseVector(result map[string]interface{}) []map[string]interface{} {
var out []map[string]interface{}
data, _ := result["data"].(map[string]interface{})
res, _ := data["result"].([]interface{})
for _, r := range res {
rm, _ := r.(map[string]interface{})
metric, _ := rm["metric"].(map[string]interface{})
value, _ := rm["value"].([]interface{})
entry := map[string]interface{}{}
for k, v := range metric {
entry[k] = v
}
if len(value) > 1 {
if v, err := strconv.ParseFloat(fmt.Sprintf("%v", value[1]), 64); err == nil {
entry["value"] = math.Round(v*100) / 100
}
}
out = append(out, entry)
}
return out
}
// ============================================
// 预定义仪表盘查询
// ============================================
func (h *PrometheusProxyHandler) dashboardNode(client *http.Client, baseURL string) map[string]interface{} {
results := gin.H{}
// 节点概览表
results["nodes"] = parseVector(h.mustQuery(client, baseURL, `node_uname_info - 0`))
// CPU
results["cpuUsage"] = parseVector(h.mustQuery(client, baseURL,
`(1 - avg(irate(node_cpu_seconds_total{mode="idle"}[5m])) by (instance)) * 100`))
// 内存
results["memUsage"] = parseVector(h.mustQuery(client, baseURL,
`(1 - (node_memory_MemAvailable_bytes / node_memory_MemTotal_bytes)) * 100`))
// 磁盘
results["diskUsage"] = parseVector(h.mustQuery(client, baseURL,
`(1 - sum(node_filesystem_avail_bytes{fstype=~"xfs|ext.*"}) by (instance) / sum(node_filesystem_size_bytes{fstype=~"xfs|ext.*"}) by (instance)) * 100`))
// Load
results["load5"] = parseVector(h.mustQuery(client, baseURL, `node_load5`))
// 网络
results["netRx"] = parseVector(h.mustQuery(client, baseURL,
`max(irate(node_network_receive_bytes_total[5m])) by (instance)`))
results["netTx"] = parseVector(h.mustQuery(client, baseURL,
`max(irate(node_network_transmit_bytes_total[5m])) by (instance)`))
results["up"] = parseVector(h.mustQuery(client, baseURL, `up`))
return results
}
func (h *PrometheusProxyHandler) dashboardPod(client *http.Client, baseURL string) map[string]interface{} {
results := gin.H{}
results["cpuTop"] = parseVector(h.mustQuery(client, baseURL,
`topk(10, sum(rate(container_cpu_usage_seconds_total{container!="",pod!=""}[5m])) by (namespace, pod))`))
results["memTop"] = parseVector(h.mustQuery(client, baseURL,
`topk(10, sum(container_memory_working_set_bytes{container!="",pod!=""}) by (namespace, pod))`))
results["netRxTop"] = parseVector(h.mustQuery(client, baseURL,
`topk(10, sum(rate(container_network_receive_bytes_total{pod!=""}[5m])) by (namespace, pod))`))
results["netTxTop"] = parseVector(h.mustQuery(client, baseURL,
`topk(10, sum(rate(container_network_transmit_bytes_total{pod!=""}[5m])) by (namespace, pod))`))
results["restartTop"] = parseVector(h.mustQuery(client, baseURL,
`topk(10, sum(kube_pod_container_status_restarts_total) by (namespace, pod))`))
results["running"] = parseVector(h.mustQuery(client, baseURL,
`sum(kube_pod_status_phase{phase="Running"})`))
results["pending"] = parseVector(h.mustQuery(client, baseURL,
`sum(kube_pod_status_phase{phase="Pending"}) or vector(0)`))
return results
}
func (h *PrometheusProxyHandler) dashboardJVM(client *http.Client, baseURL string) map[string]interface{} {
results := gin.H{}
results["uptime"] = parseVector(h.mustQuery(client, baseURL, `process_uptime_seconds`))
results["heapUsed"] = parseVector(h.mustQuery(client, baseURL,
`sum(jvm_memory_used_bytes{area="heap"}) / sum(jvm_memory_max_bytes{area="heap"}) * 100`))
results["nonHeapUsed"] = parseVector(h.mustQuery(client, baseURL,
`sum(jvm_memory_used_bytes{area="nonheap"}) / sum(jvm_memory_max_bytes{area="nonheap"}) * 100`))
results["threadCount"] = parseVector(h.mustQuery(client, baseURL,
`jvm_threads_live_threads`))
results["gcPause"] = parseVector(h.mustQuery(client, baseURL,
`rate(jvm_gc_pause_seconds_sum[5m])`))
results["cpuLoad"] = parseVector(h.mustQuery(client, baseURL,
`process_cpu_usage`))
return results
}
func (h *PrometheusProxyHandler) dashboardK8s(client *http.Client, baseURL string) map[string]interface{} {
results := gin.H{}
// 节点数
results["nodeCount"] = parseVector(h.mustQuery(client, baseURL, `count(kube_node_info)`))
// Pod 总数
results["podCount"] = parseVector(h.mustQuery(client, baseURL,
`count(kube_pod_info{created_by_kind!~"<none>|Job"})`))
// 节点 CPU 利用率
results["nodeCPU"] = parseVector(h.mustQuery(client, baseURL,
`sum(irate(container_cpu_usage_seconds_total{container!=""}[2m])) by (node) / sum(kube_pod_container_resource_requests{resource="cpu"}) by (node) * 100`))
// 节点内存利用率
results["nodeMem"] = parseVector(h.mustQuery(client, baseURL,
`sum(container_memory_working_set_bytes{container!=""}) by (node) / sum(kube_pod_container_resource_requests{resource="memory"}) by (node) * 100`))
// Namespace 资源统计
results["nsPods"] = parseVector(h.mustQuery(client, baseURL,
`count(kube_pod_info) by (namespace)`))
// 部署统计
results["deployCount"] = parseVector(h.mustQuery(client, baseURL,
`count(kube_deployment_metadata_generation)`))
return results
}
func (h *PrometheusProxyHandler) mustQuery(client *http.Client, baseURL, query string) map[string]interface{} {
r, err := h.execPromQL(client, baseURL, query)
if err != nil {
return map[string]interface{}{"data": map[string]interface{}{"result": []interface{}{}}}
}
return r
}
@@ -0,0 +1,284 @@
package handlers
import (
"context"
"fmt"
"math"
"net/http"
"strconv"
"strings"
"time"
"ezops/internal/models"
"ezops/pkg/zabbix"
"github.com/gin-gonic/gin"
)
// ZabbixProxyHandler Zabbix 代理处理器 — 将 Zabbix API 数据转为前端面板可用格式
type ZabbixProxyHandler struct{}
func NewZabbixProxyHandler() *ZabbixProxyHandler {
return &ZabbixProxyHandler{}
}
// ZabbixHostSummary Zabbix 主机汇总(前端面板数据)
type ZabbixHostSummary struct {
HostID string `json:"hostid"`
Hostname string `json:"hostname"`
Name string `json:"name"`
IP string `json:"ip"`
Available string `json:"available"` // 0=未知 1=可用 2=不可用
CPUUsage float64 `json:"cpu_usage"`
MemoryUsage float64 `json:"memory_usage"`
DiskUsage float64 `json:"disk_usage"`
Load5 float64 `json:"load5"`
NetInBPS float64 `json:"net_in_bps"`
NetOutBPS float64 `json:"net_out_bps"`
ProblemCount int `json:"problem_count"`
Uptime int64 `json:"uptime_days"` // 天
CPUCount int `json:"cpu_count"`
TotalMemory float64 `json:"total_memory_gb"`
}
// ZabbixConnectReq 连接 Zabbix 请求
type ZabbixConnectReq struct {
URL string `json:"url" binding:"required"`
Username string `json:"username" binding:"required"`
Password string `json:"password" binding:"required"`
}
// ZabbixConnect 连接 Zabbix 并返回主机摘要
// POST /api/v1/zabbix/connect
func (h *ZabbixProxyHandler) Connect(c *gin.Context) {
var req ZabbixConnectReq
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误: " + err.Error()})
return
}
client := zabbix.NewClient(req.URL, req.Username, req.Password)
ctx, cancel := context.WithTimeout(c.Request.Context(), 15*time.Second)
defer cancel()
if err := client.Auth(ctx); err != nil {
c.JSON(http.StatusBadGateway, models.Response{Code: 502, Message: "Zabbix 认证失败: " + err.Error()})
return
}
// 获取 Zabbix 版本
version, _ := client.GetVersion(ctx)
// 获取主机列表(含接口和组信息)
hosts, err := client.GetHosts(ctx, &zabbix.GetHostsParams{
Status: 0, // 只获取被监控的主机
SelectInterfaces: []string{"interfaceid", "hostid", "main", "type", "useip", "ip", "dns", "port"},
SelectGroups: []string{"groupid", "name"},
Output: []string{"hostid", "host", "name", "status", "available", "description"},
})
if err != nil {
client.Logout(ctx)
c.JSON(http.StatusBadGateway, models.Response{Code: 502, Message: "获取 Zabbix 主机失败: " + err.Error()})
return
}
// 为每台主机查询监控项
var summaries []ZabbixHostSummary
itemKeys := []string{
"system.cpu.util[,idle]", // CPU 空闲率
"system.cpu.util", // CPU 使用率(备用)
"vm.memory.size[total]", // 总内存 bytes
"vm.memory.size[available]", // 可用内存 bytes
"vfs.fs.size[/,total]", // 总磁盘 bytes
"vfs.fs.size[/,used]", // 已用磁盘 bytes
"system.cpu.load[all,avg5]", // 5分钟负载
"net.if.in[ens18]", // 网络流入
"net.if.out[ens18]", // 网络流出
"system.cpu.num", // CPU 数量
"system.uptime", // 运行时间
}
for _, host := range hosts {
summary := ZabbixHostSummary{
HostID: host.HostID,
Hostname: host.Host,
Name: host.Name,
Available: host.Available,
}
// 获取 IP
for _, iface := range host.Interfaces {
if iface.Main == "1" && summary.IP == "" {
if iface.UseIP == "1" && iface.IP != "" {
summary.IP = iface.IP
} else if iface.DNS != "" {
summary.IP = iface.DNS
}
}
}
// 获取监控项
items, err := client.GetItems(ctx, &zabbix.GetItemsParams{
HostIDs: []string{host.HostID},
Filter: map[string]interface{}{"key_": itemKeys},
Output: []string{"itemid", "hostid", "name", "key_", "lastvalue", "units"},
SortField: []string{"name"},
})
if err != nil {
summaries = append(summaries, summary)
continue
}
var totalMemBytes, availMemBytes, totalDiskBytes, usedDiskBytes float64
for _, item := range items {
val, _ := strconv.ParseFloat(item.LastValue, 64)
switch item.Key {
case "system.cpu.util[,idle]":
summary.CPUUsage = math.Round((100-val)*100) / 100
case "system.cpu.util":
if summary.CPUUsage == 0 {
summary.CPUUsage = math.Round(val*100) / 100
}
case "vm.memory.size[total]":
totalMemBytes = val
case "vm.memory.size[available]":
availMemBytes = val
case "vfs.fs.size[/,total]":
totalDiskBytes = val
case "vfs.fs.size[/,used]":
usedDiskBytes = val
case "system.cpu.load[all,avg5]":
summary.Load5 = math.Round(val*100) / 100
case "net.if.in[ens18]":
summary.NetInBPS = math.Round(val*100) / 100
case "net.if.out[ens18]":
summary.NetOutBPS = math.Round(val*100) / 100
case "system.cpu.num":
summary.CPUCount = int(val)
case "system.uptime":
summary.Uptime = int64(val) / 86400 // seconds → days
}
}
// 计算内存使用率
if totalMemBytes > 0 {
summary.TotalMemory = math.Round(totalMemBytes/1024/1024/1024*100) / 100 // GB
if availMemBytes > 0 {
summary.MemoryUsage = math.Round((totalMemBytes-availMemBytes)/totalMemBytes*10000) / 100
}
}
// 计算磁盘使用率
if totalDiskBytes > 0 && usedDiskBytes > 0 {
summary.DiskUsage = math.Round(usedDiskBytes/totalDiskBytes*10000) / 100
}
// 获取问题触发器
triggers, _ := client.GetTriggersByHost(ctx, host.HostID, true)
summary.ProblemCount = len(triggers)
summaries = append(summaries, summary)
}
client.Logout(ctx)
c.JSON(http.StatusOK, models.Response{
Code: 200,
Message: "success",
Data: gin.H{
"version": version,
"hosts": summaries,
"total": len(summaries),
},
})
}
// ZabbixHostDetail Zabbix 单主机详情
// GET /api/v1/zabbix/host/:hostid
func (h *ZabbixProxyHandler) GetHostDetail(c *gin.Context) {
hostID := c.Param("hostid")
if hostID == "" {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "缺少主机ID"})
return
}
// 从 query 参数获取连接信息
url := c.Query("url")
username := c.Query("username")
password := c.Query("password")
if url == "" || username == "" {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "缺少 Zabbix 连接信息"})
return
}
client := zabbix.NewClient(url, username, password)
ctx, cancel := context.WithTimeout(c.Request.Context(), 15*time.Second)
defer cancel()
if err := client.Auth(ctx); err != nil {
c.JSON(http.StatusBadGateway, models.Response{Code: 502, Message: "Zabbix 认证失败: " + err.Error()})
return
}
defer client.Logout(ctx)
// 获取主机详情
host, err := client.GetHostByID(ctx, hostID)
if err != nil {
c.JSON(http.StatusBadGateway, models.Response{Code: 502, Message: "获取主机失败: " + err.Error()})
return
}
// 获取所有监控项
items, err := client.GetItemsByHost(ctx, hostID)
if err != nil {
c.JSON(http.StatusBadGateway, models.Response{Code: 502, Message: "获取监控项失败: " + err.Error()})
return
}
// 获取问题触发器
triggers, _ := client.GetTriggersByHost(ctx, hostID, true)
c.JSON(http.StatusOK, models.Response{
Code: 200,
Message: "success",
Data: gin.H{
"host": host,
"items": items,
"triggers": triggers,
"problems": len(triggers),
},
})
}
// formatBps 格式化带宽
func formatBps(bps float64) string {
if bps == 0 {
return "—"
}
units := []string{"B/s", "KB/s", "MB/s", "GB/s", "TB/s"}
unitIdx := 0
val := bps
for val >= 1024 && unitIdx < len(units)-1 {
val /= 1024
unitIdx++
}
return fmt.Sprintf("%.1f %s", val, units[unitIdx])
}
// parseZabbixItemSummary 从 items 中提取汇总信息的辅助函数(供内部使用)
func parseZabbixItemSummary(items []zabbix.Item) map[string]string {
result := make(map[string]string)
for _, item := range items {
key := item.Key
// 标准化 key 名称
if strings.Contains(item.Name, "CPU utilization") || key == "system.cpu.util" {
result["cpu"] = item.LastValue
}
if strings.Contains(item.Name, "Memory utilization") || key == "vm.memory.size[pavailable]" {
result["memory"] = item.LastValue
}
if strings.Contains(item.Name, "Space utilization") || key == "/Space utilization/" {
result["disk"] = item.LastValue
}
}
return result
}
+31
View File
@@ -490,6 +490,37 @@ func Register(r *gin.Engine, db *gorm.DB, log *zap.Logger, cfg *config.Config, c
snmpGroup.GET("/stats", snmpHandler.GetSNMPStats)
}
// Zabbix 代理 — 镜像 Zabbix API 为前端面板可用格式 // Logan.Li
zabbixHandler := handlers.NewZabbixProxyHandler()
zabbixGroup := authorized.Group("/zabbix")
{
zabbixGroup.POST("/connect", zabbixHandler.Connect)
zabbixGroup.GET("/host/:hostid", zabbixHandler.GetHostDetail)
}
// Prometheus 代理 — 转发 PromQL 查询到 Prometheus 服务器 // Logan.Li
promHandler := handlers.NewPrometheusProxyHandler()
promGroup := authorized.Group("/prometheus")
{
promGroup.POST("/connect", promHandler.Connect)
promGroup.POST("/query", promHandler.Query)
promGroup.POST("/query_range", promHandler.QueryRange)
promGroup.POST("/dashboard", promHandler.Dashboard)
}
// Prometheus 数据源管理 — CRUD + 连接测试 // Logan.Li
promDSHandler := handlers.NewPromDSHandler(db)
promDSGroup := authorized.Group("/prometheus-datasources")
{
promDSGroup.GET("", promDSHandler.List)
promDSGroup.GET("/:id", promDSHandler.Get)
promDSGroup.POST("", promDSHandler.Create)
promDSGroup.PUT("/:id", promDSHandler.Update)
promDSGroup.DELETE("/:id", promDSHandler.Delete)
promDSGroup.POST("/:id/default", promDSHandler.SetDefault)
promDSGroup.POST("/test", promDSHandler.TestConnection)
}
// 站点拨测 — HTTP/TCP/DNS/PING 主动监控 // Logan.Li
siteHandler := handlers.NewSiteHandler(db)
siteGroup := authorized.Group("/site")
+28 -8
View File
@@ -32,16 +32,36 @@ func (AgentReport) TableName() string {
return "agent_reports"
}
// AgentLatestResponse Agent 最新状态(含服务器元信息)// Logan.Li
type AgentLatestResponse struct {
Hostname string `json:"hostname"`
IP string `json:"ip"`
CPUUsage float64 `json:"cpu_usage"`
CPUCores int `json:"cpu_cores"`
MemUsage float64 `json:"mem_usage"`
MemTotal float64 `json:"mem_total"`
MemUsed float64 `json:"mem_used"`
DiskUsage float64 `json:"disk_usage"`
DiskInfo string `json:"disk_info"`
NetInfo string `json:"net_info"`
ReportTime string `json:"report_time"`
AgentVersion string `json:"agent_version"`
AgentStatus string `json:"agent_status"`
OS string `json:"os"`
Region string `json:"region"`
}
// AgentReportRequest Agent 上报请求体 // Logan.Li
type AgentReportRequest struct {
Hostname string `json:"hostname" binding:"required"`
Timestamp string `json:"timestamp"`
CPU CPUMetric `json:"cpu"`
Memory MemMetric `json:"memory"`
Disks []DiskMetric `json:"disks"`
Ports []PortMetric `json:"ports"`
Network NetworkStat `json:"network"` // 网络指标
DiskIO DiskIOStat `json:"disk_io"` // 磁盘 IO
Hostname string `json:"hostname" binding:"required"`
Timestamp string `json:"timestamp"`
AgentVersion string `json:"agent_version"` // Agent 版本号
CPU CPUMetric `json:"cpu"`
Memory MemMetric `json:"memory"`
Disks []DiskMetric `json:"disks"`
Ports []PortMetric `json:"ports"`
Network NetworkStat `json:"network"` // 网络指标
DiskIO DiskIOStat `json:"disk_io"` // 磁盘 IO
}
type CPUMetric struct {
@@ -0,0 +1,117 @@
package models
import (
"time"
)
// PrometheusDataSource Prometheus 数据源配置(参考 Grafana 界面设计) // Logan.Li
type PrometheusDataSource struct {
ID uint `gorm:"primaryKey;autoIncrement;comment:主键ID" json:"id"`
Name string `gorm:"type:varchar(128);not null;uniqueIndex;comment:数据源名称" json:"name"`
URL string `gorm:"type:varchar(512);not null;comment:Prometheus 服务地址" json:"url"`
Access string `gorm:"type:varchar(16);default:'server';comment:访问模式: server/proxy" json:"access"`
// Basic Auth
BasicAuth bool `gorm:"default:false;comment:启用基础认证" json:"basic_auth"`
BasicAuthUser string `gorm:"type:varchar(128);comment:认证用户名" json:"basic_auth_user,omitempty"`
BasicAuthPass string `gorm:"type:varchar(256);comment:认证密码" json:"basic_auth_pass,omitempty"`
WithCredentials bool `gorm:"default:false;comment:携带凭证" json:"with_credentials"`
// TLS Client Auth
TLSClientAuth bool `gorm:"default:false;comment:启用TLS客户端认证" json:"tls_client_auth"`
WithCACert bool `gorm:"default:false;comment:携带CA证书" json:"with_ca_cert"`
CACert string `gorm:"type:text;comment:CA证书内容(PEM)" json:"ca_cert,omitempty"`
SkipTLSVerify bool `gorm:"default:false;comment:跳过TLS验证" json:"skip_tls_verify"`
// Custom HTTP Headers (JSON array of {header, value})
CustomHeaders string `gorm:"type:text;comment:自定义HTTP头(JSON)" json:"custom_headers,omitempty"`
// Alerting
EnableAlerting bool `gorm:"default:true;comment:启用告警管理" json:"enable_alerting"`
AlertmanagerDS string `gorm:"type:varchar(128);comment:Alertmanager数据源" json:"alertmanager_ds,omitempty"`
// 采集与查询
ScrapeInterval int `gorm:"default:15;comment:采集间隔(秒)" json:"scrape_interval"`
QueryTimeout int `gorm:"default:60;comment:查询超时(秒)" json:"query_timeout"`
HTTPMethod string `gorm:"type:varchar(8);default:'POST';comment:HTTP方法" json:"http_method"`
// 杂项
DisableMetricsLookup bool `gorm:"default:false;comment:禁用指标查找" json:"disable_metrics_lookup"`
CustomQueryParams string `gorm:"type:text;comment:自定义查询参数(JSON)" json:"custom_query_params,omitempty"`
// 状态
IsDefault bool `gorm:"default:false;index;comment:是否默认数据源" json:"is_default"`
Status int `gorm:"default:1;comment:状态: 1启用 0禁用" json:"status"`
// 时间戳
CreatedAt time.Time `gorm:"not null;index;comment:创建时间" json:"created_at"`
UpdatedAt time.Time `gorm:"not null;comment:更新时间" json:"updated_at"`
DeletedAt *time.Time `gorm:"index;comment:删除时间" json:"deleted_at,omitempty"`
}
// TableName 指定表名
func (PrometheusDataSource) TableName() string {
return "prometheus_datasources"
}
// ============================================
// 请求/响应 struct
// ============================================
// CreatePromDSReq 创建数据源请求
type CreatePromDSReq struct {
Name string `json:"name" binding:"required"`
URL string `json:"url" binding:"required"`
Access string `json:"access"`
BasicAuth bool `json:"basic_auth"`
BasicAuthUser string `json:"basic_auth_user"`
BasicAuthPass string `json:"basic_auth_pass"`
WithCredentials bool `json:"with_credentials"`
TLSClientAuth bool `json:"tls_client_auth"`
WithCACert bool `json:"with_ca_cert"`
CACert string `json:"ca_cert"`
SkipTLSVerify bool `json:"skip_tls_verify"`
CustomHeaders string `json:"custom_headers"`
EnableAlerting *bool `json:"enable_alerting"`
AlertmanagerDS string `json:"alertmanager_ds"`
ScrapeInterval int `json:"scrape_interval"`
QueryTimeout int `json:"query_timeout"`
HTTPMethod string `json:"http_method"`
DisableMetricsLookup bool `json:"disable_metrics_lookup"`
CustomQueryParams string `json:"custom_query_params"`
IsDefault bool `json:"is_default"`
}
// UpdatePromDSReq 更新数据源请求
type UpdatePromDSReq struct {
Name *string `json:"name"`
URL *string `json:"url"`
Access *string `json:"access"`
BasicAuth *bool `json:"basic_auth"`
BasicAuthUser *string `json:"basic_auth_user"`
BasicAuthPass *string `json:"basic_auth_pass"`
WithCredentials *bool `json:"with_credentials"`
TLSClientAuth *bool `json:"tls_client_auth"`
WithCACert *bool `json:"with_ca_cert"`
CACert *string `json:"ca_cert"`
SkipTLSVerify *bool `json:"skip_tls_verify"`
CustomHeaders *string `json:"custom_headers"`
EnableAlerting *bool `json:"enable_alerting"`
AlertmanagerDS *string `json:"alertmanager_ds"`
ScrapeInterval *int `json:"scrape_interval"`
QueryTimeout *int `json:"query_timeout"`
HTTPMethod *string `json:"http_method"`
DisableMetricsLookup *bool `json:"disable_metrics_lookup"`
CustomQueryParams *string `json:"custom_query_params"`
IsDefault *bool `json:"is_default"`
Status *int `json:"status"`
}
// TestPromDSReq 测试数据源连接请求
type TestPromDSReq struct {
URL string `json:"url" binding:"required"`
BasicAuth bool `json:"basic_auth"`
BasicAuthUser string `json:"basic_auth_user"`
BasicAuthPass string `json:"basic_auth_pass"`
SkipTLSVerify bool `json:"skip_tls_verify"`
}
+2 -1
View File
@@ -129,7 +129,8 @@ func AutoMigrate(db *gorm.DB) error {
&models.AlertUpgrade{}, // 告警升级 // Logan.Li
&models.AlertUpgradeRecord{},// 告警升级记录 // Logan.Li
&models.AlertSilence{}, // 告警静默 // Logan.Li
&models.AlertEventV2{}, // 告警事件V2 // Logan.Li
&models.AlertEventV2{}, // 告警事件V2 // Logan.Li
&models.PrometheusDataSource{}, // Prometheus 数据源配置 // Logan.Li
}
var lastErr error
+1
View File
@@ -7,6 +7,7 @@ import (
"fmt"
"io"
"net/http"
"strconv"
"sync"
"time"
)