feat: 添加节点可观测性查询功能,支持获取节点的系统配置、指标快照、流量报告和健康事件

This commit is contained in:
ryan
2026-03-14 11:07:32 +08:00
parent d9d02e749a
commit 1f294d72b9
7 changed files with 307 additions and 0 deletions
+40
View File
@@ -12,6 +12,11 @@ type nodeAgentUpdateRequest struct {
TagName string `json:"tag_name"`
}
type nodeObservabilityQuery struct {
Hours int `form:"hours"`
Limit int `form:"limit"`
}
// CreateNode godoc
// @Summary Create node
// @Tags Nodes
@@ -207,3 +212,38 @@ func GetNodeAgentRelease(c *gin.Context) {
}
respondSuccess(c, release)
}
// GetNodeObservability godoc
// @Summary Get node observability details
// @Tags Nodes
// @Produce json
// @Security BearerAuth
// @Param id path int true "Node ID"
// @Param hours query int false "Lookback window in hours"
// @Param limit query int false "Max records per section"
// @Success 200 {object} map[string]interface{}
// @Failure 400 {object} map[string]interface{}
// @Router /api/nodes/{id}/observability [get]
func GetNodeObservability(c *gin.Context) {
id, err := strconv.ParseUint(c.Param("id"), 10, 64)
if err != nil || id == 0 {
respondBadRequest(c, "")
return
}
var query nodeObservabilityQuery
if err = c.ShouldBindQuery(&query); err != nil {
respondBadRequest(c, "")
return
}
view, err := service.GetNodeObservability(uint(id), service.NodeObservabilityQuery{
Hours: query.Hours,
Limit: query.Limit,
})
if err != nil {
respondFailure(c, err.Error())
return
}
respondSuccess(c, view)
}
+1
View File
@@ -117,6 +117,7 @@ func SetApiRouter(router *gin.Engine) {
nodeRoute.GET("/", controller.GetNodes)
nodeRoute.POST("/", controller.CreateNode)
nodeRoute.GET("/:id/agent-release", controller.GetNodeAgentRelease)
nodeRoute.GET("/:id/observability", controller.GetNodeObservability)
nodeRoute.POST("/:id/agent-update", controller.RequestNodeAgentUpdate)
nodeRoute.POST("/:id/openresty-restart", controller.RequestNodeOpenrestyRestart)
nodeRoute.PUT("/:id", controller.UpdateNode)
+7
View File
@@ -265,6 +265,13 @@ func TestPhase2AgentLifecycle(t *testing.T) {
t.Fatal("expected node list to expose openresty message")
}
observabilityResp := performJSONRequest(t, engine, adminToken, http.MethodGet, "/api/nodes/"+toString(createdNode.ID)+"/observability?hours=24&limit=20", nil)
var observability service.NodeObservabilityView
decodeResponseData(t, observabilityResp, &observability)
if observability.NodeID != createdNode.NodeID {
t.Fatalf("expected observability response for node %s, got %s", createdNode.NodeID, observability.NodeID)
}
restartResp := performJSONRequest(t, engine, adminToken, http.MethodPost, "/api/nodes/"+toString(createdNode.ID)+"/openresty-restart", nil)
decodeResponseData(t, restartResp, &createdNode)
if !createdNode.RestartOpenrestyRequested {
+84
View File
@@ -0,0 +1,84 @@
package service
import (
"atsflare/model"
"errors"
"time"
"gorm.io/gorm"
)
const (
defaultObservabilityWindow = 24 * time.Hour
defaultObservabilityLimit = 120
maxObservabilityLimit = 500
)
type NodeObservabilityQuery struct {
Hours int `json:"hours"`
Limit int `json:"limit"`
}
type NodeObservabilityView struct {
NodeID string `json:"node_id"`
Profile *model.NodeSystemProfile `json:"profile"`
MetricSnapshots []*model.NodeMetricSnapshot `json:"metric_snapshots"`
TrafficReports []*model.NodeRequestReport `json:"traffic_reports"`
HealthEvents []*model.NodeHealthEvent `json:"health_events"`
}
func GetNodeObservability(id uint, query NodeObservabilityQuery) (*NodeObservabilityView, error) {
node, err := model.GetNodeByID(id)
if err != nil {
return nil, err
}
limit := normalizeObservabilityLimit(query.Limit)
since := time.Now().Add(-normalizeObservabilityWindow(query.Hours))
profile, err := model.GetNodeSystemProfile(node.NodeID)
if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
return nil, err
}
if errors.Is(err, gorm.ErrRecordNotFound) {
profile = nil
}
snapshots, err := model.ListNodeMetricSnapshots(node.NodeID, since, limit)
if err != nil {
return nil, err
}
reports, err := model.ListNodeRequestReports(node.NodeID, since, limit)
if err != nil {
return nil, err
}
events, err := model.ListNodeHealthEvents(node.NodeID, false, limit)
if err != nil {
return nil, err
}
return &NodeObservabilityView{
NodeID: node.NodeID,
Profile: profile,
MetricSnapshots: snapshots,
TrafficReports: reports,
HealthEvents: events,
}, nil
}
func normalizeObservabilityLimit(limit int) int {
if limit <= 0 {
return defaultObservabilityLimit
}
if limit > maxObservabilityLimit {
return maxObservabilityLimit
}
return limit
}
func normalizeObservabilityWindow(hours int) time.Duration {
if hours <= 0 {
return defaultObservabilityWindow
}
return time.Duration(hours) * time.Hour
}
+97
View File
@@ -493,3 +493,100 @@ func TestHeartbeatNodeResolvesMissingHealthEvents(t *testing.T) {
t.Fatalf("expected resolved health event record, got %+v", allEvents)
}
}
func TestGetNodeObservability(t *testing.T) {
setupServiceTestDB(t)
node := &model.Node{
NodeID: "node-observability-query",
Name: "query-edge",
IP: "10.0.0.61",
AgentToken: "token-query",
AgentVersion: "v0.6.0",
NginxVersion: "1.27.1.2",
Status: NodeStatusOnline,
}
if err := node.Insert(); err != nil {
t.Fatalf("failed to insert node: %v", err)
}
if err := model.UpsertNodeSystemProfile(&model.NodeSystemProfile{
NodeID: node.NodeID,
Hostname: "query-edge",
OSName: "Ubuntu",
Architecture: "amd64",
ReportedAt: time.Now(),
}); err != nil {
t.Fatalf("failed to insert node system profile: %v", err)
}
if err := (&model.NodeMetricSnapshot{
NodeID: node.NodeID,
CapturedAt: time.Now(),
}).Insert(); err != nil {
t.Fatalf("failed to insert node metric snapshot: %v", err)
}
if err := (&model.NodeRequestReport{
NodeID: node.NodeID,
WindowStartedAt: time.Now().Add(-time.Minute),
WindowEndedAt: time.Now(),
RequestCount: 123,
}).Insert(); err != nil {
t.Fatalf("failed to insert node request report: %v", err)
}
if err := model.DB.Create(&model.NodeHealthEvent{
NodeID: node.NodeID,
EventType: "sync_error",
Severity: NodeHealthSeverityWarning,
Status: NodeHealthEventStatusActive,
Message: "checksum mismatch",
FirstTriggeredAt: time.Now().Add(-time.Minute),
LastTriggeredAt: time.Now(),
ReportedAt: time.Now(),
}).Error; err != nil {
t.Fatalf("failed to insert node health event: %v", err)
}
view, err := GetNodeObservability(node.ID, NodeObservabilityQuery{Hours: 24, Limit: 10})
if err != nil {
t.Fatalf("GetNodeObservability failed: %v", err)
}
if view.NodeID != node.NodeID {
t.Fatalf("unexpected node id: %s", view.NodeID)
}
if view.Profile == nil || view.Profile.OSName != "Ubuntu" {
t.Fatalf("unexpected profile: %+v", view.Profile)
}
if len(view.MetricSnapshots) != 1 {
t.Fatalf("expected 1 metric snapshot, got %d", len(view.MetricSnapshots))
}
if len(view.TrafficReports) != 1 || view.TrafficReports[0].RequestCount != 123 {
t.Fatalf("unexpected traffic reports: %+v", view.TrafficReports)
}
if len(view.HealthEvents) != 1 || view.HealthEvents[0].EventType != "sync_error" {
t.Fatalf("unexpected health events: %+v", view.HealthEvents)
}
}
func TestGetNodeObservabilityAllowsMissingProfile(t *testing.T) {
setupServiceTestDB(t)
node := &model.Node{
NodeID: "node-observability-empty",
Name: "empty-edge",
IP: "10.0.0.62",
AgentToken: "token-empty",
AgentVersion: "v0.6.0",
NginxVersion: "1.27.1.2",
Status: NodeStatusOnline,
}
if err := node.Insert(); err != nil {
t.Fatalf("failed to insert node: %v", err)
}
view, err := GetNodeObservability(node.ID, NodeObservabilityQuery{})
if err != nil {
t.Fatalf("GetNodeObservability failed: %v", err)
}
if view.Profile != nil {
t.Fatalf("expected nil profile when profile not reported, got %+v", view.Profile)
}
}
@@ -6,6 +6,7 @@ import type {
NodeBootstrapToken,
NodeItem,
NodeMutationPayload,
NodeObservability,
} from '@/features/nodes/types';
import type { ReleaseChannel } from '@/features/update/types';
@@ -67,3 +68,20 @@ export function requestNodeOpenrestyRestart(id: number) {
method: 'POST',
});
}
export function getNodeObservability(
id: number,
options?: { hours?: number; limit?: number },
) {
const params = new URLSearchParams();
if (options?.hours) {
params.set('hours', String(options.hours));
}
if (options?.limit) {
params.set('limit', String(options.limit));
}
const query = params.toString();
return apiRequest<NodeObservability>(
`/nodes/${id}/observability${query ? `?${query}` : ''}`,
);
}
+60
View File
@@ -57,3 +57,63 @@ export interface NodeAgentUpdatePayload {
channel?: ReleaseChannel;
tag_name?: string;
}
export interface NodeSystemProfile {
hostname: string;
os_name: string;
os_version: string;
kernel_version: string;
architecture: string;
cpu_model: string;
cpu_cores: number;
total_memory_bytes: number;
total_disk_bytes: number;
uptime_seconds: number;
reported_at: string;
}
export interface NodeMetricSnapshot {
captured_at: string;
cpu_usage_percent: number;
memory_used_bytes: number;
memory_total_bytes: number;
storage_used_bytes: number;
storage_total_bytes: number;
disk_read_bytes: number;
disk_write_bytes: number;
network_rx_bytes: number;
network_tx_bytes: number;
openresty_rx_bytes: number;
openresty_tx_bytes: number;
openresty_connections: number;
}
export interface NodeTrafficReport {
window_started_at: string;
window_ended_at: string;
request_count: number;
error_count: number;
unique_visitor_count: number;
status_codes_json: string;
top_domains_json: string;
source_countries_json: string;
}
export interface NodeHealthEvent {
event_type: string;
severity: string;
status: string;
message: string;
first_triggered_at: string;
last_triggered_at: string;
reported_at: string;
resolved_at?: string | null;
}
export interface NodeObservability {
node_id: string;
profile: NodeSystemProfile | null;
metric_snapshots: NodeMetricSnapshot[];
traffic_reports: NodeTrafficReport[];
health_events: NodeHealthEvent[];
}