360 lines
13 KiB
Go
360 lines
13 KiB
Go
package operations
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
"gorm.io/gorm"
|
|
|
|
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/admission"
|
|
deviceModels "git.ilapage.cn/ila/yovision/Sense/server/app/sense/device/models"
|
|
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/media"
|
|
coreService "github.com/go-admin-team/go-admin-core/sdk/service"
|
|
)
|
|
|
|
var (
|
|
ErrInvalidFilter = errors.New("运维中心查询条件不符合要求")
|
|
ErrProblemNotFound = errors.New("运维问题不存在或已经收敛")
|
|
ErrVersionConflict = errors.New("状态已经变化，请刷新后重试")
|
|
ErrRetryInProgress = errors.New("该对象已有重试任务")
|
|
ErrRetryNotAllowed = errors.New("该问题不允许重试")
|
|
)
|
|
|
|
type Service struct {
|
|
DB *gorm.DB
|
|
Now func() time.Time
|
|
LocalInferenceConfigured bool
|
|
DeviceRetry func(context.Context, admission.ProbeRequest) error
|
|
}
|
|
|
|
type admissionResult struct {
|
|
DeviceID string
|
|
Address string
|
|
Status string
|
|
Detail string
|
|
CheckedAt time.Time
|
|
}
|
|
|
|
func (admissionResult) TableName() string { return "sense_admission_results" }
|
|
|
|
func NewService(db *gorm.DB) *Service {
|
|
service := &Service{DB: db, Now: time.Now, LocalInferenceConfigured: false}
|
|
service.DeviceRetry = func(ctx context.Context, request admission.ProbeRequest) error {
|
|
base := coreService.Service{}
|
|
base.Orm = db
|
|
runtimeService, err := admission.NewRuntime(base)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = runtimeService.Probe(ctx, request)
|
|
if err == nil {
|
|
// Keep the existing admission boundary: route intent is best-effort and
|
|
// must not turn a successful device probe into a MediaMTX failure.
|
|
_ = media.EnsureDeviceRoutes(ctx, db, request.DeviceID)
|
|
}
|
|
return err
|
|
}
|
|
return service
|
|
}
|
|
|
|
func (s *Service) List(request PageRequest) (PageResponse, error) {
|
|
if err := validateFilter(request); err != nil {
|
|
return PageResponse{}, err
|
|
}
|
|
problems, summary, err := s.project()
|
|
if err != nil {
|
|
return PageResponse{}, err
|
|
}
|
|
filtered := make([]Problem, 0, len(problems))
|
|
keyword := strings.ToLower(strings.TrimSpace(request.Keyword))
|
|
for _, item := range problems {
|
|
if request.ObjectType != "" && item.ObjectType != request.ObjectType {
|
|
continue
|
|
}
|
|
if request.ProblemType != "" && item.ProblemType != request.ProblemType {
|
|
continue
|
|
}
|
|
if request.Severity != "" && item.Severity != request.Severity {
|
|
continue
|
|
}
|
|
searchable := strings.ToLower(strings.Join([]string{item.ObjectID, item.ObjectName, item.Location, item.Difference}, " "))
|
|
if keyword != "" && !strings.Contains(searchable, keyword) {
|
|
continue
|
|
}
|
|
filtered = append(filtered, item)
|
|
}
|
|
pageIndex, pageSize := request.GetPageIndex(), request.GetPageSize()
|
|
if pageSize > 100 {
|
|
pageSize = 100
|
|
}
|
|
start := (pageIndex - 1) * pageSize
|
|
if start > len(filtered) {
|
|
start = len(filtered)
|
|
}
|
|
end := start + pageSize
|
|
if end > len(filtered) {
|
|
end = len(filtered)
|
|
}
|
|
return PageResponse{Summary: summary, List: filtered[start:end], Count: int64(len(filtered))}, nil
|
|
}
|
|
|
|
func (s *Service) Get(id string) (Problem, error) {
|
|
problems, _, err := s.project()
|
|
if err != nil {
|
|
return Problem{}, err
|
|
}
|
|
for _, item := range problems {
|
|
if item.ID == id {
|
|
return item, nil
|
|
}
|
|
}
|
|
return Problem{}, ErrProblemNotFound
|
|
}
|
|
|
|
func (s *Service) Retry(ctx context.Context, id string, expectedVersion int64, userID int) (Problem, error) {
|
|
if expectedVersion < 1 {
|
|
return Problem{}, ErrVersionConflict
|
|
}
|
|
problem, err := s.Get(id)
|
|
if err != nil {
|
|
return Problem{}, err
|
|
}
|
|
if !problem.Retryable {
|
|
if problem.RetryInProgress {
|
|
return Problem{}, ErrRetryInProgress
|
|
}
|
|
return Problem{}, ErrRetryNotAllowed
|
|
}
|
|
now := s.now().UTC()
|
|
switch problem.ObjectType {
|
|
case ObjectDevice:
|
|
var admissionState admissionResult
|
|
if err := s.DB.First(&admissionState, "device_id = ?", problem.ObjectID).Error; err != nil || strings.TrimSpace(admissionState.Address) == "" {
|
|
return Problem{}, ErrRetryNotAllowed
|
|
}
|
|
result := s.DB.Model(&deviceModels.Device{}).
|
|
Where("id = ? AND version = ? AND retry_requested_at IS NULL", problem.ObjectID, expectedVersion).
|
|
Updates(map[string]any{"retry_requested_at": now, "version": expectedVersion + 1, "updated_at": now})
|
|
if result.Error != nil {
|
|
return Problem{}, fmt.Errorf("queue device retry: %w", result.Error)
|
|
}
|
|
if result.RowsAffected == 0 {
|
|
return Problem{}, s.retryConflict(ObjectDevice, problem.ObjectID, expectedVersion)
|
|
}
|
|
if s.DeviceRetry != nil {
|
|
err = s.DeviceRetry(ctx, admission.ProbeRequest{DeviceID: problem.ObjectID, Address: admissionState.Address, Version: expectedVersion + 1, UpdateBy: userID})
|
|
if err != nil {
|
|
clearErr := s.DB.Model(&deviceModels.Device{}).Where("id = ? AND version = ?", problem.ObjectID, expectedVersion+1).Update("retry_requested_at", nil).Error
|
|
if clearErr != nil {
|
|
return Problem{}, errors.Join(fmt.Errorf("execute device retry: %w", err), fmt.Errorf("clear device retry gate: %w", clearErr))
|
|
}
|
|
return Problem{}, fmt.Errorf("execute device retry: %w", err)
|
|
}
|
|
}
|
|
case ObjectMedia:
|
|
result := s.DB.Model(&media.Route{}).
|
|
Where("id = ? AND version = ? AND actual <> ?", problem.ObjectID, expectedVersion, "retry_pending").
|
|
Updates(map[string]any{"actual": "retry_pending", "next_retry_at": now, "detail": "已请求受控重试，等待视频服务执行", "version": expectedVersion + 1, "updated_at": now})
|
|
if result.Error != nil {
|
|
return Problem{}, fmt.Errorf("queue media retry: %w", result.Error)
|
|
}
|
|
if result.RowsAffected == 0 {
|
|
return Problem{}, s.retryConflict(ObjectMedia, problem.ObjectID, expectedVersion)
|
|
}
|
|
default:
|
|
return Problem{}, ErrRetryNotAllowed
|
|
}
|
|
updated, err := s.Get(id)
|
|
if errors.Is(err, ErrProblemNotFound) {
|
|
problem.Actual, problem.Difference, problem.NextAction = "converged", "重试完成，状态已经收敛", "无需处理"
|
|
problem.Retryable, problem.RetryInProgress, problem.Version, problem.UpdatedAt = false, false, expectedVersion+2, now
|
|
return problem, nil
|
|
}
|
|
return updated, err
|
|
}
|
|
|
|
func (s *Service) project() ([]Problem, Summary, error) {
|
|
var devices []deviceModels.Device
|
|
if err := s.DB.Order("updated_at DESC").Find(&devices).Error; err != nil {
|
|
return nil, Summary{}, fmt.Errorf("list operation devices: %w", err)
|
|
}
|
|
var admissions []admissionResult
|
|
if err := s.DB.Find(&admissions).Error; err != nil {
|
|
return nil, Summary{}, fmt.Errorf("list admission results: %w", err)
|
|
}
|
|
var routes []media.Route
|
|
if err := s.DB.Order("updated_at DESC").Find(&routes).Error; err != nil {
|
|
return nil, Summary{}, fmt.Errorf("list operation media routes: %w", err)
|
|
}
|
|
admissionByDevice := make(map[string]admissionResult, len(admissions))
|
|
for _, item := range admissions {
|
|
admissionByDevice[item.DeviceID] = item
|
|
}
|
|
deviceByID := make(map[string]deviceModels.Device, len(devices))
|
|
problems := make([]Problem, 0)
|
|
summary := Summary{ManagedCount: len(devices) + len(routes), LocalInference: "configured"}
|
|
for _, device := range devices {
|
|
deviceByID[device.ID] = device
|
|
if problem, ok := deviceProblem(device, admissionByDevice[device.ID]); ok {
|
|
problems = append(problems, problem)
|
|
summary.ActionableProblems++
|
|
} else {
|
|
summary.ConvergedCount++
|
|
}
|
|
}
|
|
for _, route := range routes {
|
|
if problem, ok := mediaProblem(route, deviceByID); ok {
|
|
problems = append(problems, problem)
|
|
summary.ActionableProblems++
|
|
} else {
|
|
summary.ConvergedCount++
|
|
}
|
|
}
|
|
if !s.LocalInferenceConfigured {
|
|
summary.LocalInference = "unavailable"
|
|
now := s.now().UTC()
|
|
problems = append(problems, Problem{
|
|
ID: "local_inference:adapter", ObjectType: ObjectLocalInference, ObjectID: "adapter",
|
|
ObjectName: "本地推理适配器", ProblemType: ProblemUnavailable, Severity: "info",
|
|
Expected: "可选", Actual: "unavailable", Difference: "未配置本地推理适配器；不影响设备和媒体运维",
|
|
NextAction: "无需处理", SafetyGate: "不读取 Brain 数据库，不阻断本页", OperationalOnly: true, UpdatedAt: now,
|
|
})
|
|
}
|
|
sort.SliceStable(problems, func(i, j int) bool {
|
|
rank := map[string]int{"high": 0, "medium": 1, "info": 2}
|
|
if rank[problems[i].Severity] != rank[problems[j].Severity] {
|
|
return rank[problems[i].Severity] < rank[problems[j].Severity]
|
|
}
|
|
return problems[i].UpdatedAt.After(problems[j].UpdatedAt)
|
|
})
|
|
return problems, summary, nil
|
|
}
|
|
|
|
func deviceProblem(device deviceModels.Device, admission admissionResult) (Problem, bool) {
|
|
if device.Status == deviceModels.StatusDisabled {
|
|
return Problem{}, false
|
|
}
|
|
actual, detail, updatedAt := device.AdapterStatus, "设备尚未完成接入验证", device.UpdatedAt
|
|
if admission.DeviceID != "" {
|
|
actual, detail, updatedAt = admission.Status, admission.Detail, admission.CheckedAt
|
|
}
|
|
if device.Status == deviceModels.StatusActive && (actual == "ready" || actual == "verified") {
|
|
return Problem{}, false
|
|
}
|
|
problemType, severity, nextAction := ProblemUnready, "medium", "检查设备和接入配置后重试"
|
|
text := strings.ToLower(actual + " " + detail)
|
|
if strings.Contains(text, "auth") || strings.Contains(text, "认证") || strings.Contains(text, "凭据") {
|
|
problemType, severity, nextAction = ProblemAuthentication, "high", "确认设备账号未变更后执行受控重试"
|
|
} else if strings.Contains(text, "clock") || strings.Contains(text, "time drift") || strings.Contains(text, "时间漂移") || strings.Contains(text, "时钟") {
|
|
problemType, nextAction = ProblemClockDrift, "检查设备 NTP 和时区后重新检测"
|
|
}
|
|
retrying := device.RetryRequestedAt != nil
|
|
retryable := !retrying && strings.TrimSpace(admission.Address) != ""
|
|
if admission.DeviceID == "" || strings.TrimSpace(admission.Address) == "" {
|
|
nextAction = "先到视频接入完成地址与凭据验证"
|
|
}
|
|
return Problem{
|
|
ID: "device:" + device.ID, ObjectType: ObjectDevice, ObjectID: device.ID, ObjectName: device.Name,
|
|
Location: device.Location, ProblemType: problemType, Severity: severity, Expected: "active / ready", Actual: actual,
|
|
Difference: detail, NextAction: nextAction, AttemptCount: boolInt(retrying), Retryable: retryable,
|
|
RetryInProgress: retrying, Version: device.Version, SafetyGate: "校验设备版本且同一设备仅允许一个在途重试",
|
|
OperationalOnly: true, UpdatedAt: updatedAt,
|
|
}, true
|
|
}
|
|
|
|
func mediaProblem(route media.Route, devices map[string]deviceModels.Device) (Problem, bool) {
|
|
device, exists := devices[route.DeviceID]
|
|
name, location := route.Path, ""
|
|
if exists {
|
|
name, location = device.Name+" / "+route.Path, device.Location
|
|
}
|
|
if !exists {
|
|
return Problem{
|
|
ID: "media:" + route.ID, ObjectType: ObjectMedia, ObjectID: route.ID, ObjectName: name,
|
|
ProblemType: ProblemOrphan, Severity: "medium", Expected: "路由关联有效设备", Actual: "已隔离待确认",
|
|
Difference: "媒体路由存在，但找不到有效设备归属", NextAction: "人工核对；不会自动删除", Version: route.Version,
|
|
SafetyGate: "孤儿资源仅隔离和提示，本接口没有删除动作", OperationalOnly: true, UpdatedAt: route.UpdatedAt,
|
|
}, true
|
|
}
|
|
converged := (route.Desired == media.DesiredRunning && (route.Actual == "ready" || route.Actual == "waiting")) || (route.Desired == media.DesiredStopped && route.Actual == "stopped")
|
|
if converged {
|
|
return Problem{}, false
|
|
}
|
|
problemType, nextAction := ProblemUnready, "检查视频服务状态后重试"
|
|
if route.NextRetryAt != nil {
|
|
problemType, nextAction = ProblemBackoff, "等待退避到期或执行受控提前重试"
|
|
}
|
|
retrying := route.Actual == "retry_pending"
|
|
return Problem{
|
|
ID: "media:" + route.ID, ObjectType: ObjectMedia, ObjectID: route.ID, ObjectName: name, Location: location,
|
|
ProblemType: problemType, Severity: "medium", Expected: route.Desired, Actual: route.Actual, Difference: route.Detail,
|
|
NextAction: nextAction, NextRetryAt: route.NextRetryAt, AttemptCount: route.FailureCount, Retryable: !retrying,
|
|
RetryInProgress: retrying, Version: route.Version, SafetyGate: "校验路由版本且只排队，不删除路径或修改凭据",
|
|
OperationalOnly: true, UpdatedAt: route.UpdatedAt,
|
|
}, true
|
|
}
|
|
|
|
func (s *Service) retryConflict(objectType, objectID string, expectedVersion int64) error {
|
|
var version int64
|
|
var inProgress bool
|
|
switch objectType {
|
|
case ObjectDevice:
|
|
var item deviceModels.Device
|
|
if err := s.DB.Select("version", "retry_requested_at").First(&item, "id = ?", objectID).Error; err != nil {
|
|
return ErrProblemNotFound
|
|
}
|
|
version, inProgress = item.Version, item.RetryRequestedAt != nil
|
|
case ObjectMedia:
|
|
var item media.Route
|
|
if err := s.DB.Select("version", "actual").First(&item, "id = ?", objectID).Error; err != nil {
|
|
return ErrProblemNotFound
|
|
}
|
|
version, inProgress = item.Version, item.Actual == "retry_pending"
|
|
}
|
|
if inProgress {
|
|
return ErrRetryInProgress
|
|
}
|
|
if version != expectedVersion {
|
|
return ErrVersionConflict
|
|
}
|
|
return ErrVersionConflict
|
|
}
|
|
|
|
func (s *Service) now() time.Time {
|
|
if s.Now != nil {
|
|
return s.Now()
|
|
}
|
|
return time.Now()
|
|
}
|
|
|
|
func validateFilter(request PageRequest) error {
|
|
valid := func(value string, values ...string) bool {
|
|
if value == "" {
|
|
return true
|
|
}
|
|
for _, candidate := range values {
|
|
if value == candidate {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
if !valid(request.ObjectType, ObjectDevice, ObjectMedia, ObjectLocalInference) ||
|
|
!valid(request.ProblemType, ProblemAuthentication, ProblemBackoff, ProblemClockDrift, ProblemOrphan, ProblemUnavailable, ProblemUnready) ||
|
|
!valid(request.Severity, "high", "medium", "info") || len([]rune(request.Keyword)) > 128 {
|
|
return ErrInvalidFilter
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func boolInt(value bool) int {
|
|
if value {
|
|
return 1
|
|
}
|
|
return 0
|
|
}
|