feat: 实现边缘节点资产与离线状态管理 (#76)
This commit is contained in:
QiuSW
@@ -0,0 +1,19 @@
|
||||
package router
|
||||
|
||||
import (
|
||||
"github.com/gin-gonic/gin"
|
||||
jwt "github.com/go-admin-team/go-admin-core/sdk/pkg/jwtauth"
|
||||
|
||||
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/edge_node"
|
||||
"git.ilapage.cn/ila/yovision/Sense/server/common/actions"
|
||||
"git.ilapage.cn/ila/yovision/Sense/server/common/middleware"
|
||||
)
|
||||
|
||||
func init() { routerCheckRole = append(routerCheckRole, registerSenseEdgeNodeRouter) }
|
||||
|
||||
func registerSenseEdgeNodeRouter(v1 *gin.RouterGroup, auth *jwt.GinJWTMiddleware) {
|
||||
api := &edge_node.API{}
|
||||
r := v1.Group("/edge-nodes").Use(auth.MiddlewareFunc()).Use(middleware.AuthCheckRole()).Use(actions.PermissionAction())
|
||||
r.GET("", api.List)
|
||||
r.GET("/:id", api.Get)
|
||||
}
|
||||
@@ -0,0 +1,81 @@
|
||||
package edge_node
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/go-admin-team/go-admin-core/sdk/api"
|
||||
"github.com/go-admin-team/go-admin-core/sdk/pkg/jwtauth/user"
|
||||
coreService "github.com/go-admin-team/go-admin-core/sdk/service"
|
||||
|
||||
"git.ilapage.cn/ila/yovision/Sense/server/common"
|
||||
)
|
||||
|
||||
const auditSuccess, auditFailure = "1", "2"
|
||||
|
||||
type API struct{ api.Api }
|
||||
|
||||
func (e *API) service(c *gin.Context) (*Service, error) {
|
||||
base := coreService.Service{}
|
||||
if err := e.MakeContext(c).MakeOrm().MakeService(&base).Errors; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return NewService(base.Orm), nil
|
||||
}
|
||||
|
||||
func (e *API) List(c *gin.Context) {
|
||||
service, err := e.service(c)
|
||||
if err != nil {
|
||||
e.writeError(err)
|
||||
return
|
||||
}
|
||||
request := PageRequest{}
|
||||
if err = e.MakeContext(c).Bind(&request).Errors; err != nil {
|
||||
e.audit(c, service, "List", auditFailure, "边缘节点查询条件格式不正确")
|
||||
e.Error(http.StatusBadRequest, err, "查询条件格式不正确")
|
||||
return
|
||||
}
|
||||
response, err := service.List(request)
|
||||
if err != nil {
|
||||
e.audit(c, service, "List", auditFailure, "边缘节点列表查询失败")
|
||||
e.writeError(err)
|
||||
return
|
||||
}
|
||||
e.audit(c, service, "List", auditSuccess, "读取边缘节点列表")
|
||||
e.OK(response, "查询成功")
|
||||
}
|
||||
|
||||
func (e *API) Get(c *gin.Context) {
|
||||
service, err := e.service(c)
|
||||
if err != nil {
|
||||
e.writeError(err)
|
||||
return
|
||||
}
|
||||
item, err := service.Get(c.Param("id"))
|
||||
if err != nil {
|
||||
e.audit(c, service, "Get", auditFailure, "边缘节点详情查询失败")
|
||||
e.writeError(err)
|
||||
return
|
||||
}
|
||||
e.audit(c, service, "Get", auditSuccess, "读取边缘节点详情 "+item.ID)
|
||||
e.OK(item, "查询成功")
|
||||
}
|
||||
|
||||
func (e *API) audit(c *gin.Context, service *Service, action, status, remark string) {
|
||||
if err := WriteAudit(service.Orm, Audit{Action: action, Method: c.Request.Method, Status: status, Username: user.GetUserName(c), UserID: user.GetUserId(c), ClientIP: common.GetClientIP(c), Route: c.FullPath(), Remark: remark, At: time.Now()}); err != nil {
|
||||
api.GetRequestLogger(c).Errorf("edge node audit failed: %s", err.Error())
|
||||
}
|
||||
}
|
||||
|
||||
func (e *API) writeError(err error) {
|
||||
switch {
|
||||
case errors.Is(err, ErrInvalidQuery):
|
||||
e.Error(http.StatusBadRequest, err, err.Error())
|
||||
case errors.Is(err, ErrNotFound):
|
||||
e.Error(http.StatusNotFound, err, err.Error())
|
||||
default:
|
||||
e.Error(http.StatusInternalServerError, err, "边缘节点查询失败")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,21 @@
|
||||
package edge_node
|
||||
|
||||
import (
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
|
||||
adminModels "git.ilapage.cn/ila/yovision/Sense/server/app/admin/models"
|
||||
)
|
||||
|
||||
type Audit struct {
|
||||
Action, Method, Status, Username, ClientIP, Route, Remark string
|
||||
UserID int
|
||||
At time.Time
|
||||
}
|
||||
|
||||
func WriteAudit(db *gorm.DB, input Audit) error {
|
||||
model := adminModels.SysOperaLog{Title: "边缘节点", BusinessType: "other", Method: "edge_node.API." + input.Action, RequestMethod: input.Method, OperatorType: "1", OperName: input.Username, OperUrl: input.Route, OperIp: input.ClientIP, Status: input.Status, OperTime: input.At.UTC(), Remark: input.Remark, CreatedAt: input.At.UTC(), UpdatedAt: input.At.UTC()}
|
||||
model.CreateBy, model.UpdateBy = input.UserID, input.UserID
|
||||
return db.Create(&model).Error
|
||||
}
|
||||
@@ -0,0 +1,65 @@
|
||||
package edge_node
|
||||
|
||||
import (
|
||||
"time"
|
||||
|
||||
commonDto "git.ilapage.cn/ila/yovision/Sense/server/common/dto"
|
||||
)
|
||||
|
||||
const HeartbeatTimeout = 90 * time.Second
|
||||
|
||||
type PageRequest struct {
|
||||
commonDto.Pagination `search:"-"`
|
||||
}
|
||||
|
||||
type Summary struct {
|
||||
Total int `json:"total"`
|
||||
Online int `json:"online"`
|
||||
Offline int `json:"offline"`
|
||||
Recovering int `json:"recovering"`
|
||||
LoadUsed int `json:"loadUsed"`
|
||||
LoadCapacity int `json:"loadCapacity"`
|
||||
BackfillQueueDepth int `json:"backfillQueueDepth"`
|
||||
HeartbeatTimeout int64 `json:"heartbeatTimeoutSeconds"`
|
||||
}
|
||||
|
||||
type Response struct {
|
||||
ID string `json:"id"`
|
||||
Name string `json:"name"`
|
||||
Location string `json:"location"`
|
||||
Status string `json:"status"`
|
||||
RuntimeVersion string `json:"runtimeVersion"`
|
||||
ModelVersion string `json:"modelVersion"`
|
||||
StartedAt time.Time `json:"startedAt"`
|
||||
UptimeSeconds int64 `json:"uptimeSeconds"`
|
||||
LastHeartbeatAt time.Time `json:"lastHeartbeatAt"`
|
||||
LastCollectedAt time.Time `json:"lastCollectedAt"`
|
||||
FreshnessSeconds int64 `json:"freshnessSeconds"`
|
||||
Stale bool `json:"stale"`
|
||||
LoadUsed int `json:"loadUsed"`
|
||||
LoadCapacity int `json:"loadCapacity"`
|
||||
ControlTunnelStatus string `json:"controlTunnelStatus"`
|
||||
ControlTunnelDetail string `json:"controlTunnelDetail"`
|
||||
ControlLatencyMs int `json:"controlLatencyMs"`
|
||||
VideoPlaneStatus string `json:"videoPlaneStatus"`
|
||||
VideoPlaneDetail string `json:"videoPlaneDetail"`
|
||||
ActiveStreamCount int `json:"activeStreamCount"`
|
||||
BackfillStatus string `json:"backfillStatus"`
|
||||
BackfillQueueDepth int `json:"backfillQueueDepth"`
|
||||
RecoveryPhase string `json:"recoveryPhase"`
|
||||
ProjectionVersion int64 `json:"projectionVersion"`
|
||||
Events []Event `json:"events,omitempty"`
|
||||
}
|
||||
|
||||
type PageResponse struct {
|
||||
Summary Summary `json:"summary"`
|
||||
List []Response `json:"list"`
|
||||
Count int64 `json:"count"`
|
||||
}
|
||||
|
||||
// HeartbeatSample is an internal Sense adapter input. It is intentionally not
|
||||
// exposed as an HTTP contract by this task.
|
||||
type HeartbeatSample struct {
|
||||
Node
|
||||
CollectedAt time.Time
|
||||
}
|
||||
@@ -0,0 +1,20 @@
|
||||
package edge_node
|
||||
|
||||
import (
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
"gorm.io/gorm/clause"
|
||||
)
|
||||
|
||||
// SeedSyntheticFixture is explicit test/development data. Production startup
|
||||
// and migrations never call it.
|
||||
func SeedSyntheticFixture(db *gorm.DB, now time.Time) error {
|
||||
now = now.UTC().Truncate(time.Second)
|
||||
items := []Node{
|
||||
{ID: "SEN-EDGE-01", Name: "主楼边缘节点", Location: "主楼机房", RuntimeVersion: "sense-edge 0.1.0", ModelVersion: "detector-v3", StartedAt: now.Add(-7 * 24 * time.Hour), LastHeartbeatAt: now.Add(-18 * time.Second), LastCollectedAt: now.Add(-18 * time.Second), LoadUsed: 6, LoadCapacity: 16, ControlTunnelStatus: ChannelReady, ControlTunnelDetail: "控制隧道正常", ControlLatencyMs: 24, VideoPlaneStatus: ChannelReady, VideoPlaneDetail: "6 路视频正常", ActiveStreamCount: 6, BackfillStatus: BackfillIdle, ProjectionVersion: 1},
|
||||
{ID: "SEN-EDGE-02", Name: "仓库边缘节点", Location: "仓库弱电间", RuntimeVersion: "sense-edge 0.1.0", ModelVersion: "detector-v3", StartedAt: now.Add(-3 * 24 * time.Hour), LastHeartbeatAt: now.Add(-4 * time.Minute), LastCollectedAt: now.Add(-4 * time.Minute), LoadUsed: 4, LoadCapacity: 16, ControlTunnelStatus: ChannelReady, ControlTunnelDetail: "最后已知：控制隧道正常", ControlLatencyMs: 31, VideoPlaneStatus: ChannelReady, VideoPlaneDetail: "最后已知：4 路视频正常", ActiveStreamCount: 4, BackfillStatus: BackfillPending, BackfillQueueDepth: 12, ProjectionVersion: 1},
|
||||
{ID: "SEN-EDGE-03", Name: "南门边缘节点", Location: "南门岗亭", RuntimeVersion: "sense-edge 0.1.0", ModelVersion: "detector-v3", StartedAt: now.Add(-10 * time.Hour), LastHeartbeatAt: now.Add(-12 * time.Second), LastCollectedAt: now.Add(-12 * time.Second), LoadUsed: 2, LoadCapacity: 8, ControlTunnelStatus: ChannelReady, ControlTunnelDetail: "控制隧道已恢复", ControlLatencyMs: 46, VideoPlaneStatus: ChannelUnavailable, VideoPlaneDetail: "视频数据面恢复中", ActiveStreamCount: 0, BackfillStatus: BackfillPending, BackfillQueueDepth: 5, RecoveryPhase: "video_reconnecting", ProjectionVersion: 2},
|
||||
}
|
||||
return db.Clauses(clause.OnConflict{DoNothing: true}).Create(&items).Error
|
||||
}
|
||||
@@ -0,0 +1,61 @@
|
||||
package edge_node
|
||||
|
||||
import (
|
||||
"time"
|
||||
|
||||
common "git.ilapage.cn/ila/yovision/Sense/server/common/models"
|
||||
)
|
||||
|
||||
const (
|
||||
StatusOnline = "online"
|
||||
StatusOffline = "offline"
|
||||
StatusRecovering = "recovering"
|
||||
|
||||
ChannelReady = "ready"
|
||||
ChannelUnavailable = "unavailable"
|
||||
BackfillIdle = "idle"
|
||||
BackfillPending = "pending"
|
||||
)
|
||||
|
||||
// Node is the Sense-owned last-known projection of one edge runtime. It is
|
||||
// deliberately not a shared identity or a cross-product protocol model.
|
||||
type Node struct {
|
||||
ID string `gorm:"size:64;primaryKey" json:"id"`
|
||||
Name string `gorm:"size:128;not null" json:"name"`
|
||||
Location string `gorm:"size:256;not null;default:''" json:"location"`
|
||||
RuntimeVersion string `gorm:"size:64;not null;default:''" json:"runtimeVersion"`
|
||||
ModelVersion string `gorm:"size:64;not null;default:''" json:"modelVersion"`
|
||||
StartedAt time.Time `gorm:"not null" json:"startedAt"`
|
||||
LastHeartbeatAt time.Time `gorm:"not null;index" json:"lastHeartbeatAt"`
|
||||
LastCollectedAt time.Time `gorm:"not null" json:"lastCollectedAt"`
|
||||
LoadUsed int `gorm:"not null;default:0" json:"loadUsed"`
|
||||
LoadCapacity int `gorm:"not null;default:0" json:"loadCapacity"`
|
||||
ControlTunnelStatus string `gorm:"size:32;not null;default:'unavailable'" json:"controlTunnelStatus"`
|
||||
ControlTunnelDetail string `gorm:"size:256;not null;default:''" json:"controlTunnelDetail"`
|
||||
ControlLatencyMs int `gorm:"not null;default:0" json:"controlLatencyMs"`
|
||||
VideoPlaneStatus string `gorm:"size:32;not null;default:'unavailable'" json:"videoPlaneStatus"`
|
||||
VideoPlaneDetail string `gorm:"size:256;not null;default:''" json:"videoPlaneDetail"`
|
||||
ActiveStreamCount int `gorm:"not null;default:0" json:"activeStreamCount"`
|
||||
BackfillStatus string `gorm:"size:32;not null;default:'idle'" json:"backfillStatus"`
|
||||
BackfillQueueDepth int `gorm:"not null;default:0" json:"backfillQueueDepth"`
|
||||
RecoveryPhase string `gorm:"size:32;not null;default:''" json:"recoveryPhase"`
|
||||
ProjectionVersion int64 `gorm:"not null;default:1" json:"projectionVersion"`
|
||||
common.ControlBy
|
||||
common.ModelTime
|
||||
}
|
||||
|
||||
func (Node) TableName() string { return "sense_edge_nodes" }
|
||||
|
||||
// Event records projection state transitions without storing credentials or
|
||||
// raw heartbeat payloads.
|
||||
type Event struct {
|
||||
ID uint64 `gorm:"primaryKey;autoIncrement" json:"id"`
|
||||
NodeID string `gorm:"size:64;not null;index" json:"nodeId"`
|
||||
EventType string `gorm:"size:32;not null;index" json:"eventType"`
|
||||
FromStatus string `gorm:"size:32;not null;default:''" json:"fromStatus"`
|
||||
ToStatus string `gorm:"size:32;not null" json:"toStatus"`
|
||||
Detail string `gorm:"size:256;not null;default:''" json:"detail"`
|
||||
OccurredAt time.Time `gorm:"not null;index" json:"occurredAt"`
|
||||
}
|
||||
|
||||
func (Event) TableName() string { return "sense_edge_node_events" }
|
||||
@@ -0,0 +1,167 @@
|
||||
package edge_node
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
coreService "github.com/go-admin-team/go-admin-core/sdk/service"
|
||||
"gorm.io/gorm"
|
||||
"gorm.io/gorm/clause"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrInvalidQuery = errors.New("边缘节点查询条件不符合要求")
|
||||
ErrNotFound = errors.New("边缘节点不存在")
|
||||
ErrInvalidSample = errors.New("边缘节点心跳样本不符合要求")
|
||||
)
|
||||
|
||||
type Service struct {
|
||||
coreService.Service
|
||||
Now func() time.Time
|
||||
}
|
||||
|
||||
func NewService(db *gorm.DB) *Service {
|
||||
return &Service{Service: coreService.Service{Orm: db}, Now: time.Now}
|
||||
}
|
||||
|
||||
func (s *Service) List(request PageRequest) (PageResponse, error) {
|
||||
if request.GetPageSize() > 100 {
|
||||
return PageResponse{}, ErrInvalidQuery
|
||||
}
|
||||
var count int64
|
||||
if err := s.Orm.Model(&Node{}).Count(&count).Error; err != nil {
|
||||
return PageResponse{}, fmt.Errorf("count edge nodes: %w", err)
|
||||
}
|
||||
var models []Node
|
||||
if err := s.Orm.Order("name ASC, id ASC").Limit(request.GetPageSize()).Offset((request.GetPageIndex() - 1) * request.GetPageSize()).Find(&models).Error; err != nil {
|
||||
return PageResponse{}, fmt.Errorf("list edge nodes: %w", err)
|
||||
}
|
||||
result := PageResponse{List: make([]Response, 0, len(models)), Count: count}
|
||||
result.Summary.HeartbeatTimeout = int64(HeartbeatTimeout / time.Second)
|
||||
var all []Node
|
||||
if err := s.Orm.Find(&all).Error; err != nil {
|
||||
return PageResponse{}, fmt.Errorf("summarize edge nodes: %w", err)
|
||||
}
|
||||
result.Summary.Total = len(all)
|
||||
for _, item := range all {
|
||||
status := s.status(item)
|
||||
switch status {
|
||||
case StatusOffline:
|
||||
result.Summary.Offline++
|
||||
case StatusRecovering:
|
||||
result.Summary.Recovering++
|
||||
default:
|
||||
result.Summary.Online++
|
||||
}
|
||||
result.Summary.LoadUsed += item.LoadUsed
|
||||
result.Summary.LoadCapacity += item.LoadCapacity
|
||||
result.Summary.BackfillQueueDepth += item.BackfillQueueDepth
|
||||
}
|
||||
for _, item := range models {
|
||||
result.List = append(result.List, s.response(item, nil))
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (s *Service) Get(id string) (Response, error) {
|
||||
id = strings.TrimSpace(id)
|
||||
if id == "" || len(id) > 64 {
|
||||
return Response{}, ErrInvalidQuery
|
||||
}
|
||||
var model Node
|
||||
if err := s.Orm.First(&model, "id = ?", id).Error; err != nil {
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return Response{}, ErrNotFound
|
||||
}
|
||||
return Response{}, fmt.Errorf("get edge node: %w", err)
|
||||
}
|
||||
var events []Event
|
||||
if err := s.Orm.Where("node_id = ?", id).Order("occurred_at DESC, id DESC").Limit(20).Find(&events).Error; err != nil {
|
||||
return Response{}, fmt.Errorf("list edge node events: %w", err)
|
||||
}
|
||||
return s.response(model, events), nil
|
||||
}
|
||||
|
||||
// ApplyHeartbeat updates the Sense-owned projection atomically and records
|
||||
// only state transitions. Callers are internal adapters, not public APIs.
|
||||
func (s *Service) ApplyHeartbeat(ctx context.Context, sample HeartbeatSample) error {
|
||||
now := sample.CollectedAt.UTC()
|
||||
if now.IsZero() {
|
||||
now = s.now()
|
||||
}
|
||||
sample.ID = strings.TrimSpace(sample.ID)
|
||||
if sample.ID == "" || len(sample.ID) > 64 || strings.TrimSpace(sample.Name) == "" || sample.LastHeartbeatAt.IsZero() || sample.StartedAt.IsZero() || sample.LoadUsed < 0 || sample.LoadCapacity < 0 || sample.LoadUsed > sample.LoadCapacity || sample.BackfillQueueDepth < 0 {
|
||||
return ErrInvalidSample
|
||||
}
|
||||
sample.LastHeartbeatAt = sample.LastHeartbeatAt.UTC()
|
||||
sample.LastCollectedAt = now
|
||||
return s.Orm.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||||
var previous Node
|
||||
err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&previous, "id = ?", sample.ID).Error
|
||||
if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return fmt.Errorf("lock edge node projection: %w", err)
|
||||
}
|
||||
from := ""
|
||||
if err == nil {
|
||||
from = statusAt(previous, now)
|
||||
sample.CreatedAt = previous.CreatedAt
|
||||
sample.ProjectionVersion = previous.ProjectionVersion + 1
|
||||
} else {
|
||||
sample.ProjectionVersion = 1
|
||||
}
|
||||
if err = tx.Save(&sample.Node).Error; err != nil {
|
||||
return fmt.Errorf("save edge node projection: %w", err)
|
||||
}
|
||||
to := statusAt(sample.Node, now)
|
||||
if from == to && from != "" {
|
||||
return nil
|
||||
}
|
||||
if from == StatusOffline {
|
||||
timeoutAt := previous.LastHeartbeatAt.UTC().Add(HeartbeatTimeout)
|
||||
if err = tx.Create(&Event{NodeID: sample.ID, EventType: "heartbeat_timeout", FromStatus: StatusOnline, ToStatus: StatusOffline, Detail: "节点心跳超过 90 秒未更新", OccurredAt: timeoutAt}).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
eventType, detail := "heartbeat_received", "收到节点首次心跳"
|
||||
if from == StatusOffline {
|
||||
eventType, detail = "node_recovered", "节点心跳恢复，通道状态继续独立收敛"
|
||||
} else if to == StatusRecovering {
|
||||
eventType, detail = "recovery_started", "节点进入恢复阶段"
|
||||
}
|
||||
return tx.Create(&Event{NodeID: sample.ID, EventType: eventType, FromStatus: from, ToStatus: to, Detail: detail, OccurredAt: now}).Error
|
||||
})
|
||||
}
|
||||
|
||||
func (s *Service) response(model Node, events []Event) Response {
|
||||
now := s.now()
|
||||
freshness := int64(now.Sub(model.LastHeartbeatAt.UTC()).Seconds())
|
||||
if freshness < 0 {
|
||||
freshness = 0
|
||||
}
|
||||
uptime := int64(model.LastHeartbeatAt.UTC().Sub(model.StartedAt.UTC()).Seconds())
|
||||
if uptime < 0 {
|
||||
uptime = 0
|
||||
}
|
||||
return Response{ID: model.ID, Name: model.Name, Location: model.Location, Status: statusAt(model, now), RuntimeVersion: model.RuntimeVersion, ModelVersion: model.ModelVersion, StartedAt: model.StartedAt, UptimeSeconds: uptime, LastHeartbeatAt: model.LastHeartbeatAt, LastCollectedAt: model.LastCollectedAt, FreshnessSeconds: freshness, Stale: freshness > int64(HeartbeatTimeout/time.Second), LoadUsed: model.LoadUsed, LoadCapacity: model.LoadCapacity, ControlTunnelStatus: model.ControlTunnelStatus, ControlTunnelDetail: model.ControlTunnelDetail, ControlLatencyMs: model.ControlLatencyMs, VideoPlaneStatus: model.VideoPlaneStatus, VideoPlaneDetail: model.VideoPlaneDetail, ActiveStreamCount: model.ActiveStreamCount, BackfillStatus: model.BackfillStatus, BackfillQueueDepth: model.BackfillQueueDepth, RecoveryPhase: model.RecoveryPhase, ProjectionVersion: model.ProjectionVersion, Events: events}
|
||||
}
|
||||
|
||||
func (s *Service) status(model Node) string { return statusAt(model, s.now()) }
|
||||
func (s *Service) now() time.Time {
|
||||
if s.Now != nil {
|
||||
return s.Now().UTC()
|
||||
}
|
||||
return time.Now().UTC()
|
||||
}
|
||||
|
||||
func statusAt(model Node, now time.Time) string {
|
||||
if now.UTC().Sub(model.LastHeartbeatAt.UTC()) > HeartbeatTimeout {
|
||||
return StatusOffline
|
||||
}
|
||||
if strings.TrimSpace(model.RecoveryPhase) != "" && model.RecoveryPhase != "converged" {
|
||||
return StatusRecovering
|
||||
}
|
||||
return StatusOnline
|
||||
}
|
||||
@@ -0,0 +1,103 @@
|
||||
package edge_node
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"gorm.io/driver/sqlite"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
func edgeNodeTestService(t *testing.T) (*Service, *gorm.DB, time.Time) {
|
||||
t.Helper()
|
||||
db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err = db.AutoMigrate(&Node{}, &Event{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
now := time.Date(2026, 8, 28, 4, 0, 0, 0, time.UTC)
|
||||
service := NewService(db)
|
||||
service.Now = func() time.Time { return now }
|
||||
return service, db, now
|
||||
}
|
||||
|
||||
func TestListClassifiesOnlineOfflineAndRecoveringWithoutErasingLastKnownState(t *testing.T) {
|
||||
service, db, now := edgeNodeTestService(t)
|
||||
if err := SeedSyntheticFixture(db, now); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
page, err := service.List(PageRequest{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if page.Count != 3 || page.Summary.Online != 1 || page.Summary.Offline != 1 || page.Summary.Recovering != 1 || page.Summary.BackfillQueueDepth != 17 {
|
||||
t.Fatalf("unexpected summary: %+v", page.Summary)
|
||||
}
|
||||
var offline Response
|
||||
for _, item := range page.List {
|
||||
if item.Status == StatusOffline {
|
||||
offline = item
|
||||
}
|
||||
}
|
||||
if !offline.Stale || offline.ControlTunnelStatus != ChannelReady || offline.VideoPlaneStatus != ChannelReady || offline.BackfillQueueDepth != 12 {
|
||||
t.Fatalf("offline projection lost last-known state: %+v", offline)
|
||||
}
|
||||
}
|
||||
|
||||
func TestApplyHeartbeatRecordsRecoveryAndPreservesIndependentChannelConvergence(t *testing.T) {
|
||||
service, db, now := edgeNodeTestService(t)
|
||||
old := Node{ID: "EDGE-1", Name: "旧节点", StartedAt: now.Add(-time.Hour), LastHeartbeatAt: now.Add(-5 * time.Minute), LastCollectedAt: now.Add(-5 * time.Minute), LoadCapacity: 16, ControlTunnelStatus: ChannelReady, VideoPlaneStatus: ChannelReady, BackfillStatus: BackfillPending, BackfillQueueDepth: 8, ProjectionVersion: 3}
|
||||
if err := db.Create(&old).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
sample := HeartbeatSample{Node: Node{ID: "EDGE-1", Name: "旧节点", StartedAt: old.StartedAt, LastHeartbeatAt: now.Add(-5 * time.Second), LoadUsed: 4, LoadCapacity: 16, ControlTunnelStatus: ChannelReady, VideoPlaneStatus: ChannelUnavailable, BackfillStatus: BackfillPending, BackfillQueueDepth: 3, RecoveryPhase: "video_reconnecting"}, CollectedAt: now}
|
||||
if err := service.ApplyHeartbeat(context.Background(), sample); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
item, err := service.Get("EDGE-1")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if item.Status != StatusRecovering || item.ProjectionVersion != 4 || item.VideoPlaneStatus != ChannelUnavailable || item.BackfillQueueDepth != 3 {
|
||||
t.Fatalf("unexpected recovery projection: %+v", item)
|
||||
}
|
||||
if len(item.Events) != 2 || item.Events[0].EventType != "node_recovered" || item.Events[0].FromStatus != StatusOffline || item.Events[0].ToStatus != StatusRecovering || item.Events[1].EventType != "heartbeat_timeout" || item.Events[1].ToStatus != StatusOffline {
|
||||
t.Fatalf("unexpected events: %+v", item.Events)
|
||||
}
|
||||
}
|
||||
|
||||
func TestApplyHeartbeatRejectsInvalidSampleAndDoesNotPersist(t *testing.T) {
|
||||
service, db, now := edgeNodeTestService(t)
|
||||
err := service.ApplyHeartbeat(context.Background(), HeartbeatSample{Node: Node{ID: "EDGE-2", Name: "节点", StartedAt: now.Add(-time.Hour), LastHeartbeatAt: now, LoadUsed: 17, LoadCapacity: 16}, CollectedAt: now})
|
||||
if !errors.Is(err, ErrInvalidSample) {
|
||||
t.Fatalf("err=%v", err)
|
||||
}
|
||||
var count int64
|
||||
db.Model(&Node{}).Count(&count)
|
||||
if count != 0 {
|
||||
t.Fatalf("count=%d", count)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetRejectsMissingOrMalformedID(t *testing.T) {
|
||||
service, _, _ := edgeNodeTestService(t)
|
||||
if _, err := service.Get(""); !errors.Is(err, ErrInvalidQuery) {
|
||||
t.Fatalf("err=%v", err)
|
||||
}
|
||||
if _, err := service.Get("missing"); !errors.Is(err, ErrNotFound) {
|
||||
t.Fatalf("err=%v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestListRejectsOversizedPage(t *testing.T) {
|
||||
service, _, _ := edgeNodeTestService(t)
|
||||
request := PageRequest{}
|
||||
request.PageSize = 101
|
||||
if _, err := service.List(request); !errors.Is(err, ErrInvalidQuery) {
|
||||
t.Fatalf("err=%v", err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,53 @@
|
||||
package version
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"runtime"
|
||||
|
||||
"gorm.io/gorm"
|
||||
"gorm.io/gorm/clause"
|
||||
|
||||
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/edge_node"
|
||||
"git.ilapage.cn/ila/yovision/Sense/server/cmd/migrate/migration"
|
||||
migrationModels "git.ilapage.cn/ila/yovision/Sense/server/cmd/migrate/migration/models"
|
||||
common "git.ilapage.cn/ila/yovision/Sense/server/common/models"
|
||||
)
|
||||
|
||||
func init() {
|
||||
_, fileName, _, _ := runtime.Caller(0)
|
||||
migration.Migrate.SetVersion(migration.GetFilename(fileName), migrateSenseEdgeNode)
|
||||
}
|
||||
|
||||
func migrateSenseEdgeNode(db *gorm.DB, version string) error {
|
||||
return db.Transaction(func(tx *gorm.DB) error {
|
||||
if err := tx.AutoMigrate(&edge_node.Node{}, &edge_node.Event{}); err != nil {
|
||||
return err
|
||||
}
|
||||
root, err := ensureDeviceMenu(tx, migrationModels.SysMenu{MenuName: senseLayoutMenuName, Title: "视频感知", Icon: "video-camera", Path: "/sense", MenuType: "M", Component: "Layout", Sort: 5, Visible: "0", IsFrame: "1"})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
page, err := ensureDeviceMenu(tx, migrationModels.SysMenu{MenuName: "SenseEdgeNode", Title: "边缘节点", Icon: "monitor", Path: "edge-node", Paths: fmt.Sprintf("/0/%d", root.MenuId), MenuType: "C", Permission: "sense:edge-node:list", ParentId: root.MenuId, Component: "/sense/edge-node/index", Sort: 9, Visible: "0", IsFrame: "1"})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
detail, err := ensureDeviceMenu(tx, migrationModels.SysMenu{MenuName: "SenseEdgeNodeDetail", Title: "查看节点详情", MenuType: "F", Action: "GET", Permission: "sense:edge-node:detail", ParentId: page.MenuId, Paths: fmt.Sprintf("/0/%d/%d", root.MenuId, page.MenuId), Sort: 1, Visible: "1", IsFrame: "1"})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, role := range []string{"implementation_operator", "site_admin", "viewer"} {
|
||||
if err = attachDeviceRole(tx, role, []migrationModels.SysMenu{page, detail}); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, policy := range [][2]string{{"/api/v1/edge-nodes", "GET"}, {"/api/v1/edge-nodes/:id", "GET"}} {
|
||||
if err = tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&deviceCasbinRule{Ptype: "p", V0: role, V1: policy[0], V2: policy[1]}).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
if err = rebuildSenseMenuPaths(tx, root.MenuId, "/0"); err != nil {
|
||||
return err
|
||||
}
|
||||
return tx.Create(&common.Migration{Version: version}).Error
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,90 @@
|
||||
package version
|
||||
|
||||
import (
|
||||
"os"
|
||||
"testing"
|
||||
|
||||
"gorm.io/driver/postgres"
|
||||
"gorm.io/driver/sqlite"
|
||||
"gorm.io/gorm"
|
||||
|
||||
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/edge_node"
|
||||
migrationModels "git.ilapage.cn/ila/yovision/Sense/server/cmd/migrate/migration/models"
|
||||
common "git.ilapage.cn/ila/yovision/Sense/server/common/models"
|
||||
)
|
||||
|
||||
func TestEdgeNodeMigrationAddsTablesAndReadOnlyPermissions(t *testing.T) {
|
||||
db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err = db.AutoMigrate(&migrationModels.SysRole{}, &migrationModels.SysMenu{}, &deviceCasbinRule{}, &common.Migration{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, role := range []string{"implementation_operator", "site_admin", "viewer"} {
|
||||
if err = db.Create(&migrationModels.SysRole{RoleName: role, RoleKey: role, Status: "2"}).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
const version = "2026082813000_edge_node.go"
|
||||
if err = migrateSenseEdgeNode(db, version); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !db.Migrator().HasTable(&edge_node.Node{}) || !db.Migrator().HasTable(&edge_node.Event{}) {
|
||||
t.Fatal("edge node tables missing")
|
||||
}
|
||||
var menus, reads, writes, applied int64
|
||||
db.Model(&migrationModels.SysMenu{}).Where("menu_name LIKE ?", "SenseEdgeNode%").Count(&menus)
|
||||
db.Model(&deviceCasbinRule{}).Where("v1 LIKE ? AND v2 = ?", "/api/v1/edge-nodes%", "GET").Count(&reads)
|
||||
db.Model(&deviceCasbinRule{}).Where("v1 LIKE ? AND v2 <> ?", "/api/v1/edge-nodes%", "GET").Count(&writes)
|
||||
db.Model(&common.Migration{}).Where("version = ?", version).Count(&applied)
|
||||
if menus != 2 || reads != 6 || writes != 0 || applied != 1 {
|
||||
t.Fatalf("menus=%d reads=%d writes=%d applied=%d", menus, reads, writes, applied)
|
||||
}
|
||||
var fixtures int64
|
||||
db.Model(&edge_node.Node{}).Count(&fixtures)
|
||||
if fixtures != 0 {
|
||||
t.Fatalf("migration seeded %d fixtures", fixtures)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEdgeNodeMigrationOnPostgres(t *testing.T) {
|
||||
dsn := os.Getenv("SENSE_EDGE_NODE_MIGRATION_TEST_DATABASE_URL")
|
||||
if dsn == "" {
|
||||
t.Skip("set SENSE_EDGE_NODE_MIGRATION_TEST_DATABASE_URL to run the PostgreSQL migration test")
|
||||
}
|
||||
db, err := gorm.Open(postgres.Open(dsn), &gorm.Config{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
const schema = "sense_edge_node_76_test"
|
||||
if err = db.Exec("DROP SCHEMA IF EXISTS " + schema + " CASCADE").Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err = db.Exec("CREATE SCHEMA " + schema).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = db.Exec("DROP SCHEMA IF EXISTS " + schema + " CASCADE").Error })
|
||||
sqlDB, err := db.DB()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
sqlDB.SetMaxOpenConns(1)
|
||||
if err = db.Exec("SET search_path TO " + schema).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err = db.AutoMigrate(&migrationModels.SysRole{}, &migrationModels.SysMenu{}, &deviceCasbinRule{}, &common.Migration{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, role := range []string{"implementation_operator", "site_admin", "viewer"} {
|
||||
if err = db.Create(&migrationModels.SysRole{RoleName: role, RoleKey: role, Status: "2"}).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if err = migrateSenseEdgeNode(db, "2026082813000_edge_node.go"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !db.Migrator().HasTable(&edge_node.Node{}) || !db.Migrator().HasTable(&edge_node.Event{}) {
|
||||
t.Fatal("edge node tables missing")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user