feat: 按运营商账户调度的后台接口与重算到期
- /polling-carrier-params 列表（含默认标记与消费能力）、详情、保存（执行上限不得大于订阅进程 max_in_flight 之和）、删除；/polling-check-scenes 列表与修改（检查类型、递增的确认延迟、冷却）；/polling-failure-alerts 列表、按账户与错误类别汇总、详情、标记已处理；GET /iot-cards/:id/polling-state 单卡轮询状态；GET /polling-stats/carrier-queues 按账户队列与执行监控 - 读取仅超级管理员与平台账号，修改仅超级管理员（告警处理平台账号可用），均写统一审计；新增审计动作与资源 - carrier 模式下重算到期时间写入卡到期信息与账户队列，只提前不推后
This commit is contained in:
邹进
@@ -33,6 +33,7 @@ func generateOpenAPIDocs(outputPath string, logger *zap.Logger) {
|
||||
handlers.PhoneAssetAssociation = admin.NewPhoneAssetAssociationHandler(nil, nil)
|
||||
// 网关限流 Handler 必须同时进入文档工厂，避免新增管理接口遗漏文档注册。
|
||||
handlers.GatewayRateLimit = admin.NewGatewayRateLimitHandler(nil, nil, nil)
|
||||
handlers.CarrierScheduler = admin.NewCarrierSchedulerHandler(nil, nil)
|
||||
// 套餐真流量预警 Handler 必须同时进入文档工厂，避免新增管理接口遗漏文档注册。
|
||||
handlers.PackageTrafficAlert = admin.NewPackageTrafficAlertHandler(nil, nil, nil, nil)
|
||||
// 资产钱包自动续费配置 Handler 必须同时进入文档工厂，避免新增管理接口遗漏文档注册。
|
||||
|
||||
@@ -42,6 +42,7 @@ func generateAdminDocs(outputPath string) error {
|
||||
handlers.PhoneAssetAssociation = admin.NewPhoneAssetAssociationHandler(nil, nil)
|
||||
// 网关限流 Handler 必须同时进入文档工厂，避免新增管理接口遗漏文档注册。
|
||||
handlers.GatewayRateLimit = admin.NewGatewayRateLimitHandler(nil, nil, nil)
|
||||
handlers.CarrierScheduler = admin.NewCarrierSchedulerHandler(nil, nil)
|
||||
// 套餐真流量预警 Handler 必须同时进入文档工厂，避免新增管理接口遗漏文档注册。
|
||||
handlers.PackageTrafficAlert = admin.NewPackageTrafficAlertHandler(nil, nil, nil, nil)
|
||||
// 资产钱包自动续费配置 Handler 必须同时进入文档工厂，避免新增管理接口遗漏文档注册。
|
||||
|
||||
@@ -196,9 +196,9 @@ func runWorker(cfg *config.Config) {
|
||||
taskHandler.GetMux().HandleFunc(constants.TaskTypeOutboxDeliver, outboxHandler.Handle)
|
||||
partitionMaintainer := gatewaylimit.NewPartitionMaintainer(postgres.NewCardStopResumeRecordStore(runtime.db), runtime.gatewayLimit.Settings, appLogger)
|
||||
taskHandler.GetMux().HandleFunc(constants.TaskTypeStopResumeRecordPartition, partitionMaintainer.Handle)
|
||||
recalcHandler := polling.NewRecalculateTaskHandler(
|
||||
polling.NewDueRecalculator(runtime.db, runtime.redisClient, runtime.pollingConfigMgr, runtime.pollingQueueMgr, appLogger),
|
||||
runtime.redisClient, appLogger)
|
||||
dueRecalculator := polling.NewDueRecalculator(runtime.db, runtime.redisClient, runtime.pollingConfigMgr, runtime.pollingQueueMgr, appLogger)
|
||||
dueRecalculator.SetCarrierScheduling(runtime.carrierScheduling)
|
||||
recalcHandler := polling.NewRecalculateTaskHandler(dueRecalculator, runtime.redisClient, appLogger)
|
||||
taskHandler.GetMux().HandleFunc(constants.TaskTypePollingRecalculateDue, recalcHandler.Handle)
|
||||
startOutboxRelay(ctx, runtime, cfg.Worker.InstanceName, appLogger)
|
||||
rescuePendingImportTasks(ctx, runtime, appLogger)
|
||||
|
||||
@@ -337,10 +337,12 @@ func initHandlers(svc *services, deps *Dependencies) *Handlers {
|
||||
PollingConfig: admin.NewPollingConfigHandler(svc.PollingConfig),
|
||||
GatewayRateLimit: admin.NewGatewayRateLimitHandler(pollingSvcPkg.NewGatewayRateLimitService(deps.DB, auditInfra.NewWriter(auditInfra.NewRegistry(), nil), deps.GatewayLimit), deps.Redis, validate),
|
||||
PollingConcurrency: admin.NewPollingConcurrencyHandler(svc.PollingConcurrency),
|
||||
PollingMonitoring: admin.NewPollingMonitoringHandler(svc.PollingMonitoring),
|
||||
PollingAlert: admin.NewPollingAlertHandler(svc.PollingAlert),
|
||||
PollingCleanup: admin.NewPollingCleanupHandler(svc.PollingCleanup),
|
||||
PollingManualTrigger: admin.NewPollingManualTriggerHandler(svc.PollingManualTrigger),
|
||||
CarrierScheduler: admin.NewCarrierSchedulerHandler(pollingSvcPkg.NewCarrierSchedulerService(deps.DB, deps.Redis,
|
||||
auditInfra.NewWriter(auditInfra.NewRegistry(), nil), carrierSchedulerConfig()), validate),
|
||||
PollingMonitoring: admin.NewPollingMonitoringHandler(svc.PollingMonitoring),
|
||||
PollingAlert: admin.NewPollingAlertHandler(svc.PollingAlert),
|
||||
PollingCleanup: admin.NewPollingCleanupHandler(svc.PollingCleanup),
|
||||
PollingManualTrigger: admin.NewPollingManualTriggerHandler(svc.PollingManualTrigger),
|
||||
PollingPriority: admin.NewPriorityPollingHandler(
|
||||
svc.PriorityPolling, priorityPollingQuery.NewQuery(deps.DB, pollingPriorityItemStore), validate,
|
||||
),
|
||||
@@ -408,3 +410,11 @@ func initHandlers(svc *services, deps *Dependencies) *Handlers {
|
||||
AgentOpenAPI: openapiHandler.NewHandler(svc.AgentOpenAPI, validate),
|
||||
}
|
||||
}
|
||||
|
||||
// carrierSchedulerConfig 返回按运营商账户调度的 yaml 参数（供后台校验执行上限与展示默认值）。
|
||||
func carrierSchedulerConfig() config.CarrierSchedulerConfig {
|
||||
if cfg := config.Get(); cfg != nil {
|
||||
return cfg.Polling.Carrier
|
||||
}
|
||||
return config.CarrierSchedulerConfig{}
|
||||
}
|
||||
|
||||
@@ -65,6 +65,7 @@ type Handlers struct {
|
||||
WeComApprovalCallback *callback.WeComApprovalHandler
|
||||
PollingConfig *admin.PollingConfigHandler
|
||||
GatewayRateLimit *admin.GatewayRateLimitHandler
|
||||
CarrierScheduler *admin.CarrierSchedulerHandler
|
||||
PollingConcurrency *admin.PollingConcurrencyHandler
|
||||
PollingMonitoring *admin.PollingMonitoringHandler
|
||||
PollingAlert *admin.PollingAlertHandler
|
||||
|
||||
195
internal/handler/admin/carrier_scheduler.go
Normal file
195
internal/handler/admin/carrier_scheduler.go
Normal file
@@ -0,0 +1,195 @@
|
||||
package admin
|
||||
|
||||
import (
|
||||
"strings"
|
||||
|
||||
"github.com/go-playground/validator/v10"
|
||||
"github.com/gofiber/fiber/v2"
|
||||
|
||||
"github.com/break/junhong_cmp_fiber/internal/model/dto"
|
||||
pollingSvc "github.com/break/junhong_cmp_fiber/internal/service/polling"
|
||||
"github.com/break/junhong_cmp_fiber/pkg/errors"
|
||||
"github.com/break/junhong_cmp_fiber/pkg/response"
|
||||
)
|
||||
|
||||
// CarrierSchedulerHandler 按运营商账户调度的后台接口：调度参数、请求检查场景、失败告警、单卡轮询状态、按账户监控。
|
||||
type CarrierSchedulerHandler struct {
|
||||
service *pollingSvc.CarrierSchedulerService
|
||||
validator *validator.Validate
|
||||
}
|
||||
|
||||
// NewCarrierSchedulerHandler 创建按运营商账户调度 Handler。
|
||||
func NewCarrierSchedulerHandler(service *pollingSvc.CarrierSchedulerService, validator *validator.Validate) *CarrierSchedulerHandler {
|
||||
return &CarrierSchedulerHandler{service: service, validator: validator}
|
||||
}
|
||||
|
||||
func (h *CarrierSchedulerHandler) validate(prefix string, req any) error {
|
||||
if err := h.validator.Struct(req); err != nil {
|
||||
return errors.New(errors.CodeInvalidParam, validationMessage(prefix, req, err))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ListParams 查询运营商账户调度参数列表。GET /api/admin/polling-carrier-params
|
||||
func (h *CarrierSchedulerHandler) ListParams(c *fiber.Ctx) error {
|
||||
result, err := h.service.ListParams(c.UserContext())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return response.Success(c, result)
|
||||
}
|
||||
|
||||
// GetParam 查询运营商账户调度参数。GET /api/admin/polling-carrier-params/:carrier_id
|
||||
func (h *CarrierSchedulerHandler) GetParam(c *fiber.Ctx) error {
|
||||
carrierID, err := rateLimitPathID(c, "carrier_id", "运营商账户 ID")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
result, err := h.service.GetParam(c.UserContext(), carrierID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return response.Success(c, result)
|
||||
}
|
||||
|
||||
// SaveParam 保存运营商账户调度参数。PUT /api/admin/polling-carrier-params/:carrier_id
|
||||
func (h *CarrierSchedulerHandler) SaveParam(c *fiber.Ctx) error {
|
||||
var req dto.SaveCarrierPollingParamRequest
|
||||
if err := c.BodyParser(&req); err != nil {
|
||||
return errors.New(errors.CodeInvalidParam, "请求参数解析失败")
|
||||
}
|
||||
carrierID, err := rateLimitPathID(c, "carrier_id", "运营商账户 ID")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
req.CarrierID = carrierID
|
||||
if err := h.validate("运营商账户调度参数不合法", &req); err != nil {
|
||||
return err
|
||||
}
|
||||
result, err := h.service.SaveParam(c.UserContext(), &req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return response.Success(c, result)
|
||||
}
|
||||
|
||||
// DeleteParam 删除运营商账户调度参数（恢复全局默认）。DELETE /api/admin/polling-carrier-params/:carrier_id
|
||||
func (h *CarrierSchedulerHandler) DeleteParam(c *fiber.Ctx) error {
|
||||
carrierID, err := rateLimitPathID(c, "carrier_id", "运营商账户 ID")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := h.service.DeleteParam(c.UserContext(), carrierID); err != nil {
|
||||
return err
|
||||
}
|
||||
return response.Success(c, nil)
|
||||
}
|
||||
|
||||
// ListScenes 查询请求检查场景。GET /api/admin/polling-check-scenes
|
||||
func (h *CarrierSchedulerHandler) ListScenes(c *fiber.Ctx) error {
|
||||
result, err := h.service.ListScenes(c.UserContext())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return response.Success(c, result)
|
||||
}
|
||||
|
||||
// UpdateScene 修改请求检查场景。PUT /api/admin/polling-check-scenes/:scene
|
||||
func (h *CarrierSchedulerHandler) UpdateScene(c *fiber.Ctx) error {
|
||||
var req dto.UpdatePollingCheckSceneRequest
|
||||
if err := c.BodyParser(&req); err != nil {
|
||||
return errors.New(errors.CodeInvalidParam, "请求参数解析失败")
|
||||
}
|
||||
req.Scene = strings.TrimSpace(c.Params("scene"))
|
||||
if req.Scene == "" {
|
||||
return errors.New(errors.CodeInvalidParam, "无效的场景编码")
|
||||
}
|
||||
if err := h.validate("请求检查场景参数不合法", &req); err != nil {
|
||||
return err
|
||||
}
|
||||
result, err := h.service.UpdateScene(c.UserContext(), &req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return response.Success(c, result)
|
||||
}
|
||||
|
||||
// ListAlerts 查询卡轮询失败告警列表。GET /api/admin/polling-failure-alerts
|
||||
func (h *CarrierSchedulerHandler) ListAlerts(c *fiber.Ctx) error {
|
||||
var req dto.PollingFailureAlertListRequest
|
||||
if err := c.QueryParser(&req); err != nil {
|
||||
return errors.New(errors.CodeInvalidParam, "请求参数解析失败")
|
||||
}
|
||||
if err := h.validate("卡轮询失败告警查询参数不合法", &req); err != nil {
|
||||
return err
|
||||
}
|
||||
result, err := h.service.ListAlerts(c.UserContext(), &req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return response.Success(c, result)
|
||||
}
|
||||
|
||||
// AlertSummary 未处理告警按运营商账户 × 错误类别汇总。GET /api/admin/polling-failure-alerts/summary
|
||||
func (h *CarrierSchedulerHandler) AlertSummary(c *fiber.Ctx) error {
|
||||
result, err := h.service.AlertSummary(c.UserContext())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return response.Success(c, result)
|
||||
}
|
||||
|
||||
// GetAlert 查询卡轮询失败告警详情。GET /api/admin/polling-failure-alerts/:id
|
||||
func (h *CarrierSchedulerHandler) GetAlert(c *fiber.Ctx) error {
|
||||
id, err := rateLimitPathID(c, "id", "告警 ID")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
result, err := h.service.GetAlert(c.UserContext(), id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return response.Success(c, result)
|
||||
}
|
||||
|
||||
// HandleAlert 标记卡轮询失败告警已处理。PUT /api/admin/polling-failure-alerts/:id/status
|
||||
func (h *CarrierSchedulerHandler) HandleAlert(c *fiber.Ctx) error {
|
||||
var req dto.HandlePollingFailureAlertRequest
|
||||
if err := c.BodyParser(&req); err != nil {
|
||||
return errors.New(errors.CodeInvalidParam, "请求参数解析失败")
|
||||
}
|
||||
id, err := rateLimitPathID(c, "id", "告警 ID")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
req.ID = id
|
||||
if err := h.validate("卡轮询失败告警处理参数不合法", &req); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := h.service.HandleAlert(c.UserContext(), &req); err != nil {
|
||||
return err
|
||||
}
|
||||
return response.Success(c, nil)
|
||||
}
|
||||
|
||||
// CardState 查询单卡轮询状态。GET /api/admin/iot-cards/:id/polling-state
|
||||
func (h *CarrierSchedulerHandler) CardState(c *fiber.Ctx) error {
|
||||
id, err := rateLimitPathID(c, "id", "卡 ID")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
result, err := h.service.CardState(c.UserContext(), id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return response.Success(c, result)
|
||||
}
|
||||
|
||||
// CarrierStats 按运营商账户的轮询队列与执行监控。GET /api/admin/polling-stats/carrier-queues
|
||||
func (h *CarrierSchedulerHandler) CarrierStats(c *fiber.Ctx) error {
|
||||
result, err := h.service.CarrierStats(c.UserContext())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return response.Success(c, result)
|
||||
}
|
||||
@@ -277,6 +277,10 @@ func NewRegistry() *Registry {
|
||||
gatewayErrorRuleCreated := pollingAction(constants.AuditActionGatewayErrorRuleCreated, "创建上游错误分类规则", constants.AuditResourceGatewayErrorRule, constants.AuditRiskNormal)
|
||||
gatewayErrorRuleUpdated := pollingAction(constants.AuditActionGatewayErrorRuleUpdated, "更新上游错误分类规则", constants.AuditResourceGatewayErrorRule, constants.AuditRiskNormal)
|
||||
gatewayErrorRuleStatusUpdated := pollingAction(constants.AuditActionGatewayErrorRuleStatusUpdated, "更新上游错误分类规则状态", constants.AuditResourceGatewayErrorRule, constants.AuditRiskNormal)
|
||||
carrierPollingParamSaved := pollingAction(constants.AuditActionCarrierPollingParamSaved, "保存运营商账户调度参数", constants.AuditResourceCarrierPollingParam, constants.AuditRiskHigh)
|
||||
carrierPollingParamDeleted := pollingAction(constants.AuditActionCarrierPollingParamDeleted, "删除运营商账户调度参数", constants.AuditResourceCarrierPollingParam, constants.AuditRiskHigh)
|
||||
pollingCheckSceneUpdated := pollingAction(constants.AuditActionPollingCheckSceneUpdated, "修改请求检查场景", constants.AuditResourcePollingCheckScene, constants.AuditRiskNormal)
|
||||
pollingFailureAlertHandled := pollingAction(constants.AuditActionPollingFailureAlertHandled, "处理卡轮询失败告警", constants.AuditResourcePollingFailureAlert, constants.AuditRiskNormal)
|
||||
gatewayErrorRuleDeleted := pollingAction(constants.AuditActionGatewayErrorRuleDeleted, "删除上游错误分类规则", constants.AuditResourceGatewayErrorRule, constants.AuditRiskNormal)
|
||||
pollingConcurrencyUpdated := pollingAction(constants.AuditActionPollingConcurrencyUpdated, "更新轮询并发配置", constants.AuditResourcePollingConcurrencyConfig, constants.AuditRiskNormal)
|
||||
pollingConcurrencyReset := pollingAction(constants.AuditActionPollingConcurrencyReset, "重置轮询并发计数", constants.AuditResourcePollingConcurrencyConfig, constants.AuditRiskNormal)
|
||||
@@ -678,6 +682,10 @@ func NewRegistry() *Registry {
|
||||
constants.AuditActionGatewayErrorRuleUpdated: gatewayErrorRuleUpdated,
|
||||
constants.AuditActionGatewayErrorRuleStatusUpdated: gatewayErrorRuleStatusUpdated,
|
||||
constants.AuditActionGatewayErrorRuleDeleted: gatewayErrorRuleDeleted,
|
||||
constants.AuditActionCarrierPollingParamSaved: carrierPollingParamSaved,
|
||||
constants.AuditActionCarrierPollingParamDeleted: carrierPollingParamDeleted,
|
||||
constants.AuditActionPollingCheckSceneUpdated: pollingCheckSceneUpdated,
|
||||
constants.AuditActionPollingFailureAlertHandled: pollingFailureAlertHandled,
|
||||
constants.AuditActionPollingConcurrencyUpdated: pollingConcurrencyUpdated,
|
||||
constants.AuditActionPollingConcurrencyReset: pollingConcurrencyReset,
|
||||
constants.AuditActionPollingAlertRuleCreated: pollingAlertRuleCreated,
|
||||
@@ -966,6 +974,18 @@ func NewRegistry() *Registry {
|
||||
Type: constants.AuditResourceGatewayErrorRule, Name: "上游错误分类规则",
|
||||
IdentityFields: []string{"id", "keyword", "match_type", "category", "priority", "status"},
|
||||
},
|
||||
constants.AuditResourceCarrierPollingParam: {
|
||||
Type: constants.AuditResourceCarrierPollingParam, Name: "运营商账户调度参数",
|
||||
IdentityFields: []string{"carrier_id", "inflight_limit", "card_max_in_flight", "priority_max_ratio"},
|
||||
},
|
||||
constants.AuditResourcePollingCheckScene: {
|
||||
Type: constants.AuditResourcePollingCheckScene, Name: "请求检查场景",
|
||||
IdentityFields: []string{"scene", "scene_name"},
|
||||
},
|
||||
constants.AuditResourcePollingFailureAlert: {
|
||||
Type: constants.AuditResourcePollingFailureAlert, Name: "卡轮询失败告警",
|
||||
IdentityFields: []string{"id", "alert_level", "card_id", "carrier_id", "check_type", "status"},
|
||||
},
|
||||
constants.AuditResourcePollingAlertRule: {
|
||||
Type: constants.AuditResourcePollingAlertRule, Name: "轮询告警规则",
|
||||
IdentityFields: []string{"id", "rule_name", "task_type", "metric_type", "operator", "threshold", "alert_level", "status"},
|
||||
|
||||
185
internal/model/dto/carrier_polling_dto.go
Normal file
185
internal/model/dto/carrier_polling_dto.go
Normal file
@@ -0,0 +1,185 @@
|
||||
package dto
|
||||
|
||||
import "time"
|
||||
|
||||
// CarrierPollingParamItem 运营商账户调度参数（含全局默认合并后的生效值）。
|
||||
type CarrierPollingParamItem struct {
|
||||
CarrierID uint `json:"carrier_id" description:"运营商账户ID"`
|
||||
CarrierName string `json:"carrier_name" description:"运营商账户名称"`
|
||||
IsDefault bool `json:"is_default" description:"是否未单独配置、全部使用全局默认"`
|
||||
InflightLimit int `json:"inflight_limit" description:"生效的执行上限（跨 Worker 实例同时执行卡任务数）"`
|
||||
CardMaxInFlight int `json:"card_max_in_flight" description:"生效的每进程卡任务 max_in_flight"`
|
||||
PriorityMaxRatio *float64 `json:"priority_max_ratio" description:"优先队列每轮最大占比（0~1），空表示不限"`
|
||||
StopResumeMaxInFlight *int `json:"stop_resume_max_in_flight" description:"每进程停复机 max_in_flight（停复机指令队列使用），空表示默认"`
|
||||
SRInflightLimit *int `json:"sr_inflight_limit" description:"停复机执行上限（停复机指令队列使用），空表示默认"`
|
||||
StopResumeMinInterval *int `json:"stop_resume_min_interval" description:"单卡两次停复机调用最小间隔（秒），空表示默认"`
|
||||
StopResumeConfirmDelay *int `json:"stop_resume_confirm_delay" description:"停复机结果未知后确认延迟（秒），空表示默认"`
|
||||
ConfiguredInflight *int `json:"configured_inflight_limit" description:"单独配置的执行上限，空表示使用默认"`
|
||||
ConfiguredMaxInFlight *int `json:"configured_card_max_in_flight" description:"单独配置的每进程 max_in_flight，空表示使用默认"`
|
||||
ConsumeCapacity int `json:"consume_capacity" description:"当前订阅进程 max_in_flight 之和（执行上限不应超过该值）"`
|
||||
}
|
||||
|
||||
// CarrierPollingParamListResponse 运营商账户调度参数列表。
|
||||
type CarrierPollingParamListResponse struct {
|
||||
Items []CarrierPollingParamItem `json:"items" description:"各运营商账户调度参数"`
|
||||
SubscriberProcesses int `json:"subscriber_processes" description:"订阅卡任务的 Worker 进程数"`
|
||||
DefaultInflight int `json:"default_inflight_limit" description:"全局默认执行上限"`
|
||||
DefaultMaxInFlight int `json:"default_card_max_in_flight" description:"全局默认每进程 max_in_flight"`
|
||||
}
|
||||
|
||||
// CarrierPollingParamPath 运营商账户路径参数。
|
||||
type CarrierPollingParamPath struct {
|
||||
CarrierID uint `json:"-" path:"carrier_id" required:"true" description:"运营商账户ID"`
|
||||
}
|
||||
|
||||
// SaveCarrierPollingParamRequest 保存运营商账户调度参数；字段为空表示该项使用全局默认。
|
||||
type SaveCarrierPollingParamRequest struct {
|
||||
CarrierID uint `json:"-" path:"carrier_id" required:"true" description:"运营商账户ID"`
|
||||
InflightLimit *int `json:"inflight_limit" validate:"omitempty,min=1,max=10000" minimum:"1" maximum:"10000" description:"执行上限，空表示默认"`
|
||||
CardMaxInFlight *int `json:"card_max_in_flight" validate:"omitempty,min=1,max=2000" minimum:"1" maximum:"2000" description:"每进程卡任务 max_in_flight，空表示默认"`
|
||||
StopResumeMaxInFlight *int `json:"stop_resume_max_in_flight" validate:"omitempty,min=1,max=1000" minimum:"1" maximum:"1000" description:"每进程停复机 max_in_flight，空表示默认"`
|
||||
SRInflightLimit *int `json:"sr_inflight_limit" validate:"omitempty,min=1,max=10000" minimum:"1" maximum:"10000" description:"停复机执行上限，空表示默认"`
|
||||
PriorityMaxRatio *float64 `json:"priority_max_ratio" validate:"omitempty,gt=0,lte=1" description:"优先队列每轮最大占比（0~1]，空表示不限"`
|
||||
StopResumeMinInterval *int `json:"stop_resume_min_interval" validate:"omitempty,min=1,max=86400" minimum:"1" maximum:"86400" description:"单卡停复机最小间隔（秒），空表示默认"`
|
||||
StopResumeConfirmDelay *int `json:"stop_resume_confirm_delay" validate:"omitempty,min=1,max=86400" minimum:"1" maximum:"86400" description:"停复机结果未知确认延迟（秒），空表示默认"`
|
||||
}
|
||||
|
||||
// PollingCheckSceneItem 请求检查场景。
|
||||
type PollingCheckSceneItem struct {
|
||||
Scene string `json:"scene" description:"场景编码 (order_paid:订单支付, package_activated:套餐激活, stop_resume_succeeded:停复机成功, device_operation:设备操作, auto_renewal_purchase:自动续费购包, client_asset_detail:C 端资产详情, admin_asset_realtime:后台资产实时状态, realname_link:实名链接)"`
|
||||
SceneName string `json:"scene_name" description:"场景名称"`
|
||||
CheckTypes []string `json:"check_types" description:"检查类型 (card_status:卡状态, carddata:流量, realname:实名)"`
|
||||
ConfirmDelays []int `json:"confirm_delays" description:"延迟确认序列（秒），空表示只立即检查一次"`
|
||||
Cooldown *int `json:"cooldown" description:"冷却时长（秒），最近一次成功检查未超过该时长则不请求；空表示不冷却"`
|
||||
UpdatedAt time.Time `json:"updated_at" description:"更新时间"`
|
||||
}
|
||||
|
||||
// UpdatePollingCheckSceneRequest 修改请求检查场景（场景由代码预置，只可改检查类型、确认延迟与冷却）。
|
||||
type UpdatePollingCheckSceneRequest struct {
|
||||
Scene string `json:"-" path:"scene" required:"true" description:"场景编码"`
|
||||
CheckTypes []string `json:"check_types" validate:"required,min=1,max=3,dive,oneof=card_status carddata realname" required:"true" description:"检查类型 (card_status:卡状态, carddata:流量, realname:实名)"`
|
||||
ConfirmDelays []int `json:"confirm_delays" validate:"omitempty,max=5,dive,min=1,max=86400" description:"延迟确认序列（秒，递增），空表示只立即检查一次"`
|
||||
Cooldown *int `json:"cooldown" validate:"omitempty,min=1,max=86400" minimum:"1" maximum:"86400" description:"冷却时长（秒），空表示不冷却"`
|
||||
}
|
||||
|
||||
// PollingFailureAlertListRequest 卡轮询失败告警列表请求。
|
||||
type PollingFailureAlertListRequest struct {
|
||||
Page int `json:"page" query:"page" validate:"omitempty,min=1" minimum:"1" description:"页码"`
|
||||
PageSize int `json:"page_size" query:"page_size" validate:"omitempty,min=1,max=100" minimum:"1" maximum:"100" description:"每页数量"`
|
||||
CarrierID *uint `json:"carrier_id" query:"carrier_id" validate:"omitempty,min=1" description:"运营商账户ID"`
|
||||
CheckType string `json:"check_type" query:"check_type" validate:"omitempty,oneof=card_status carddata realname package all" description:"检查类型 (card_status:卡状态, carddata:流量, realname:实名, package:套餐, all:账户级)"`
|
||||
ErrorCategory string `json:"error_category" query:"error_category" validate:"omitempty,max=50" description:"错误类别"`
|
||||
Status *int `json:"status" query:"status" validate:"omitempty,oneof=1 2 3 4" description:"状态 (1:未处理, 2:已恢复, 3:已关闭轮询, 4:已处理)"`
|
||||
AlertLevel string `json:"alert_level" query:"alert_level" validate:"omitempty,oneof=card carrier" description:"告警级别 (card:卡级, carrier:账户级)"`
|
||||
ICCID string `json:"iccid" query:"iccid" validate:"omitempty,max=30" description:"ICCID（精确）"`
|
||||
StartTime string `json:"start_time" query:"start_time" description:"最近失败时间起始（带时区的 RFC3339 秒级时间）"`
|
||||
EndTime string `json:"end_time" query:"end_time" description:"最近失败时间结束（带时区的 RFC3339 秒级时间）"`
|
||||
}
|
||||
|
||||
// PollingFailureAlertItem 卡轮询失败告警。
|
||||
type PollingFailureAlertItem struct {
|
||||
ID uint `json:"id" description:"告警ID"`
|
||||
AlertLevel string `json:"alert_level" description:"告警级别 (card:卡级, carrier:账户级)"`
|
||||
CardID uint `json:"card_id" description:"卡ID，账户级为 0"`
|
||||
ICCID string `json:"iccid" description:"ICCID"`
|
||||
CarrierID uint `json:"carrier_id" description:"运营商账户ID"`
|
||||
CarrierName string `json:"carrier_name" description:"运营商账户名称"`
|
||||
CheckType string `json:"check_type" description:"检查类型"`
|
||||
CheckTypeName string `json:"check_type_name" description:"检查类型名称"`
|
||||
ErrorCategory string `json:"error_category" description:"最近错误类别"`
|
||||
FailCount int `json:"fail_count" description:"连续失败次数；账户级为窗口内告警数"`
|
||||
FirstFailedAt *time.Time `json:"first_failed_at" description:"首次失败时间"`
|
||||
LastFailedAt *time.Time `json:"last_failed_at" description:"最近失败时间"`
|
||||
LastError string `json:"last_error" description:"最近错误"`
|
||||
Status int `json:"status" description:"状态 (1:未处理, 2:已恢复, 3:已关闭轮询, 4:已处理)"`
|
||||
StatusName string `json:"status_name" description:"状态名称"`
|
||||
HandledBy uint `json:"handled_by" description:"处理人"`
|
||||
HandledAt *time.Time `json:"handled_at" description:"处理时间"`
|
||||
HandleRemark string `json:"handle_remark" description:"处理备注"`
|
||||
RecoveredAt *time.Time `json:"recovered_at" description:"自动恢复时间"`
|
||||
CreatedAt time.Time `json:"created_at" description:"创建时间"`
|
||||
}
|
||||
|
||||
// PollingFailureAlertPageResult 卡轮询失败告警分页结果。
|
||||
type PollingFailureAlertPageResult struct {
|
||||
List []PollingFailureAlertItem `json:"items" description:"告警列表"`
|
||||
Total int64 `json:"total" description:"总数"`
|
||||
Page int `json:"page" description:"当前页"`
|
||||
PageSize int `json:"size" description:"每页数量"`
|
||||
}
|
||||
|
||||
// PollingFailureAlertSummaryItem 未处理告警按运营商账户 × 错误类别汇总。
|
||||
type PollingFailureAlertSummaryItem struct {
|
||||
CarrierID uint `json:"carrier_id" description:"运营商账户ID"`
|
||||
CarrierName string `json:"carrier_name" description:"运营商账户名称"`
|
||||
ErrorCategory string `json:"error_category" description:"错误类别"`
|
||||
OpenCount int64 `json:"open_count" description:"未处理卡级告警数"`
|
||||
}
|
||||
|
||||
// PollingFailureAlertIDPath 告警路径参数。
|
||||
type PollingFailureAlertIDPath struct {
|
||||
ID uint `json:"-" path:"id" required:"true" description:"告警ID"`
|
||||
}
|
||||
|
||||
// HandlePollingFailureAlertRequest 标记告警已处理。
|
||||
type HandlePollingFailureAlertRequest struct {
|
||||
ID uint `json:"-" path:"id" required:"true" description:"告警ID"`
|
||||
Status int `json:"status" validate:"required,oneof=4" required:"true" description:"目标状态 (4:已处理)；关闭轮询请使用资产轮询开关接口"`
|
||||
Remark string `json:"remark" validate:"omitempty,max=500" maxLength:"500" description:"处理备注"`
|
||||
}
|
||||
|
||||
// CardPollingStatePath 卡ID路径参数。
|
||||
type CardPollingStatePath struct {
|
||||
ID uint `json:"-" path:"id" required:"true" description:"卡ID"`
|
||||
}
|
||||
|
||||
// CardPollingTypeState 卡某检查类型的轮询状态。
|
||||
type CardPollingTypeState struct {
|
||||
CheckType string `json:"check_type" description:"检查类型"`
|
||||
CheckTypeName string `json:"check_type_name" description:"检查类型名称"`
|
||||
NextDueAt *time.Time `json:"next_due_at" description:"下次到期时间"`
|
||||
LastSuccessAt *time.Time `json:"last_success_at" description:"最近一次成功时间"`
|
||||
FailCount int `json:"fail_count" description:"连续失败次数"`
|
||||
FirstFailedAt *time.Time `json:"first_failed_at" description:"首次失败时间"`
|
||||
LastError string `json:"last_error" description:"最近错误（类别|信息）"`
|
||||
PendingConfirm []int `json:"pending_confirm" description:"待执行的延迟确认（秒）"`
|
||||
}
|
||||
|
||||
// CardPollingStateResponse 单卡轮询状态（只读）。
|
||||
type CardPollingStateResponse struct {
|
||||
CardID uint `json:"card_id" description:"卡ID"`
|
||||
ICCID string `json:"iccid" description:"ICCID"`
|
||||
CarrierID uint `json:"carrier_id" description:"运营商账户ID"`
|
||||
SchedulerMode string `json:"scheduler_mode" description:"当前调度模式 (legacy:旧分片调度, carrier:按运营商账户调度)"`
|
||||
EnablePolling bool `json:"enable_polling" description:"卡轮询开关"`
|
||||
Queue string `json:"queue" description:"所在队列 (priority:优先, queue:普通, none:不在轮询中)"`
|
||||
QueueScoreAt *time.Time `json:"queue_score_at" description:"队列中下次可领取时间"`
|
||||
Running bool `json:"running" description:"是否执行中"`
|
||||
ClaimExpireAt *time.Time `json:"claim_expire_at" description:"处理中标记到期时间"`
|
||||
PriorityMark bool `json:"priority_mark" description:"执行中收到请求检查，收尾后进入优先队列"`
|
||||
Types []CardPollingTypeState `json:"types" description:"各检查类型状态"`
|
||||
}
|
||||
|
||||
// CarrierPollingStatsItem 按运营商账户的轮询监控。
|
||||
type CarrierPollingStatsItem struct {
|
||||
CarrierID uint `json:"carrier_id" description:"运营商账户ID"`
|
||||
CarrierName string `json:"carrier_name" description:"运营商账户名称"`
|
||||
PriorityLength int64 `json:"priority_length" description:"优先队列长度"`
|
||||
QueueLength int64 `json:"queue_length" description:"普通队列长度"`
|
||||
DueCount int64 `json:"due_count" description:"已到期卡数（优先 + 普通）"`
|
||||
Running int64 `json:"running" description:"执行中卡任务数"`
|
||||
InflightLimit int `json:"inflight_limit" description:"执行上限"`
|
||||
QPSLastMinute float64 `json:"qps_last_minute" description:"上一分钟实际网关调用 QPS（查询类）"`
|
||||
RateLimitWaits int64 `json:"rate_limit_waits" description:"上一分钟限流等待次数"`
|
||||
RateLimited int64 `json:"rate_limited" description:"上一分钟限流拒绝或等待超时次数"`
|
||||
ExpiredClaims1h int64 `json:"expired_claims_1h" description:"近 1 小时处理中标记到期次数（执行崩溃或超时）"`
|
||||
OpenAlerts int64 `json:"open_alerts" description:"未处理卡级失败告警数"`
|
||||
OpenAlertsByType map[string]int64 `json:"open_alerts_by_category" description:"未处理卡级失败告警数（按错误类别）"`
|
||||
}
|
||||
|
||||
// CarrierPollingStatsResponse 按运营商账户的轮询监控。
|
||||
type CarrierPollingStatsResponse struct {
|
||||
SchedulerMode string `json:"scheduler_mode" description:"当前调度模式"`
|
||||
InitState string `json:"init_state" description:"新结构初始化状态 (running / done，空表示未初始化)"`
|
||||
Items []CarrierPollingStatsItem `json:"items" description:"各运营商账户监控"`
|
||||
}
|
||||
@@ -2,10 +2,12 @@ package polling
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strconv"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/redis/go-redis/v9"
|
||||
"go.uber.org/zap"
|
||||
|
||||
"github.com/break/junhong_cmp_fiber/internal/model"
|
||||
@@ -197,3 +199,28 @@ func (i *CarrierInitializer) setProgress(update func(*InitProgress)) {
|
||||
update(&i.progress)
|
||||
i.mu.Unlock()
|
||||
}
|
||||
|
||||
// AdvanceDue 把卡某检查类型的到期时间提前到 due（只提前不推后），并同步把账户队列中的 score 提前；返回是否提前。
|
||||
// 卡不在轮询中（无到期信息）时不处理。
|
||||
func (c *CarrierScheduling) AdvanceDue(ctx context.Context, card *model.IotCard, checkType string, due time.Time) (bool, error) {
|
||||
if c == nil || c.store == nil || card == nil {
|
||||
return false, nil
|
||||
}
|
||||
current, err := c.store.ReadDue(ctx, card.ID)
|
||||
if err != nil || current.CarrierID == 0 {
|
||||
return false, err
|
||||
}
|
||||
item, ok := current.Types[checkType]
|
||||
if !ok || !item.HasDue || !due.Before(item.Due) {
|
||||
return false, nil
|
||||
}
|
||||
client := c.store.Client()
|
||||
member := strconv.FormatUint(uint64(card.ID), 10)
|
||||
pipe := client.TxPipeline()
|
||||
pipe.HSet(ctx, constants.RedisCardDueKey(card.ID), carrierqueue.FieldDue(checkType), due.UnixMilli())
|
||||
score := redis.ZAddArgs{XX: true, LT: true, Members: []redis.Z{{Score: float64(due.UnixMilli()), Member: member}}}
|
||||
pipe.ZAddArgs(ctx, constants.RedisCarrierQueueKey(current.CarrierID), score)
|
||||
pipe.ZAddArgs(ctx, constants.RedisCarrierPriorityQueueKey(current.CarrierID), score)
|
||||
_, err = pipe.Exec(ctx)
|
||||
return err == nil, err
|
||||
}
|
||||
|
||||
@@ -32,6 +32,39 @@ type DueRecalculator struct {
|
||||
configMgr *PollingConfigManager
|
||||
queueMgr *PollingQueueManager
|
||||
logger *zap.Logger
|
||||
// carrier 按运营商账户调度接缝：carrier 模式下重算写入卡到期信息与账户队列。
|
||||
carrier *CarrierScheduling
|
||||
}
|
||||
|
||||
// SetCarrierScheduling 注入按运营商账户调度接缝。
|
||||
func (r *DueRecalculator) SetCarrierScheduling(carrier *CarrierScheduling) {
|
||||
r.carrier = carrier
|
||||
}
|
||||
|
||||
// recalculateCarrierBatch 按运营商账户调度时重算：各检查类型候选到期同旧结构口径，只提前不推后。
|
||||
func (r *DueRecalculator) recalculateCarrierBatch(ctx context.Context, cards []*model.IotCard) (int, error) {
|
||||
now := time.Now()
|
||||
advanced := 0
|
||||
for _, card := range cards {
|
||||
for taskType, info := range r.configMgr.MergedTaskIntervals(card) {
|
||||
checkType := constants.CheckTypeFromTaskType(taskType)
|
||||
if checkType == "" || info.Interval <= 0 {
|
||||
continue
|
||||
}
|
||||
due := recalculatedDue(lastCheckAtByTaskType(card, taskType), info.Interval, now)
|
||||
if windowDue := r.configMgr.NextCheckAt(ctx, card, taskType, info, now); windowDue.Before(due) && windowDue.After(now) {
|
||||
due = windowDue
|
||||
}
|
||||
ok, err := r.carrier.AdvanceDue(ctx, card, checkType, due)
|
||||
if err != nil {
|
||||
return advanced, err
|
||||
}
|
||||
if ok {
|
||||
advanced++
|
||||
}
|
||||
}
|
||||
}
|
||||
return advanced, nil
|
||||
}
|
||||
|
||||
// NewDueRecalculator 创建重算器。
|
||||
@@ -97,6 +130,9 @@ func (r *DueRecalculator) recalculateBatch(ctx context.Context, cards []*model.I
|
||||
if tier := r.configMgr.TierResolver(); tier != nil {
|
||||
tier.Prefill(ctx, cards)
|
||||
}
|
||||
if r.carrier.Active(ctx) {
|
||||
return r.recalculateCarrierBatch(ctx, cards)
|
||||
}
|
||||
now := time.Now()
|
||||
pipe := r.redis.Pipeline()
|
||||
var cmds []*redis.IntCmd
|
||||
|
||||
@@ -136,6 +136,9 @@ func RegisterAdminRoutes(router fiber.Router, handlers *bootstrap.Handlers, midd
|
||||
if handlers.PollingConfig != nil {
|
||||
registerPollingConfigRoutes(authGroup, handlers.PollingConfig, doc, basePath)
|
||||
}
|
||||
if handlers.CarrierScheduler != nil {
|
||||
registerCarrierSchedulerRoutes(authGroup, handlers.CarrierScheduler, doc, basePath)
|
||||
}
|
||||
if handlers.GatewayRateLimit != nil {
|
||||
registerGatewayRateLimitRoutes(authGroup, handlers.GatewayRateLimit, doc, basePath)
|
||||
}
|
||||
|
||||
68
internal/routes/carrier_scheduler.go
Normal file
68
internal/routes/carrier_scheduler.go
Normal file
@@ -0,0 +1,68 @@
|
||||
package routes
|
||||
|
||||
import (
|
||||
"github.com/gofiber/fiber/v2"
|
||||
|
||||
"github.com/break/junhong_cmp_fiber/internal/handler/admin"
|
||||
"github.com/break/junhong_cmp_fiber/internal/model/dto"
|
||||
"github.com/break/junhong_cmp_fiber/pkg/openapi"
|
||||
)
|
||||
|
||||
// registerCarrierSchedulerRoutes 注册按运营商账户调度的后台接口。读取仅超级管理员与平台账号，修改仅超级管理员（服务层校验）。
|
||||
func registerCarrierSchedulerRoutes(router fiber.Router, handler *admin.CarrierSchedulerHandler, doc *openapi.Generator, basePath string) {
|
||||
tags := []string{"轮询调度"}
|
||||
|
||||
params := router.Group("/polling-carrier-params")
|
||||
paramsPath := basePath + "/polling-carrier-params"
|
||||
Register(params, doc, paramsPath, "GET", "", handler.ListParams, RouteSpec{
|
||||
Summary: "查询运营商账户调度参数列表", Description: "未单独配置的账户返回全局默认并标记 is_default。", Tags: tags,
|
||||
Input: nil, Output: new(dto.CarrierPollingParamListResponse), Auth: true,
|
||||
})
|
||||
Register(params, doc, paramsPath, "GET", "/:carrier_id", handler.GetParam, RouteSpec{
|
||||
Summary: "查询运营商账户调度参数", Tags: tags, Input: new(dto.CarrierPollingParamPath), Output: new(dto.CarrierPollingParamItem), Auth: true,
|
||||
})
|
||||
Register(params, doc, paramsPath, "PUT", "/:carrier_id", handler.SaveParam, RouteSpec{
|
||||
Summary: "保存运营商账户调度参数（仅超级管理员）",
|
||||
Description: "字段为空表示该项使用全局默认；执行上限不得大于当前订阅进程 max_in_flight 之和；下一轮调度生效并写统一审计。",
|
||||
Tags: tags, Input: new(dto.SaveCarrierPollingParamRequest), Output: new(dto.CarrierPollingParamItem), Auth: true,
|
||||
})
|
||||
Register(params, doc, paramsPath, "DELETE", "/:carrier_id", handler.DeleteParam, RouteSpec{
|
||||
Summary: "删除运营商账户调度参数（恢复全局默认，仅超级管理员）", Tags: tags, Input: new(dto.CarrierPollingParamPath), Output: nil, Auth: true,
|
||||
})
|
||||
|
||||
scenes := router.Group("/polling-check-scenes")
|
||||
scenesPath := basePath + "/polling-check-scenes"
|
||||
Register(scenes, doc, scenesPath, "GET", "", handler.ListScenes, RouteSpec{
|
||||
Summary: "查询请求检查场景", Description: "业务事件与读入口在按运营商账户调度时按场景请求检查；场景由代码预置。", Tags: tags,
|
||||
Input: nil, Output: new([]dto.PollingCheckSceneItem), Auth: true,
|
||||
})
|
||||
Register(scenes, doc, scenesPath, "PUT", "/:scene", handler.UpdateScene, RouteSpec{
|
||||
Summary: "修改请求检查场景（仅超级管理员）", Description: "只可修改检查类型、延迟确认序列（递增）与冷却时长。", Tags: tags,
|
||||
Input: new(dto.UpdatePollingCheckSceneRequest), Output: new(dto.PollingCheckSceneItem), Auth: true,
|
||||
})
|
||||
|
||||
alerts := router.Group("/polling-failure-alerts")
|
||||
alertsPath := basePath + "/polling-failure-alerts"
|
||||
Register(alerts, doc, alertsPath, "GET", "", handler.ListAlerts, RouteSpec{
|
||||
Summary: "查询卡轮询失败告警列表", Tags: tags, Input: new(dto.PollingFailureAlertListRequest), Output: new(dto.PollingFailureAlertPageResult), Auth: true,
|
||||
})
|
||||
Register(alerts, doc, alertsPath, "GET", "/summary", handler.AlertSummary, RouteSpec{
|
||||
Summary: "未处理卡轮询失败告警按运营商账户与错误类别汇总", Tags: tags, Input: nil, Output: new([]dto.PollingFailureAlertSummaryItem), Auth: true,
|
||||
})
|
||||
Register(alerts, doc, alertsPath, "GET", "/:id", handler.GetAlert, RouteSpec{
|
||||
Summary: "查询卡轮询失败告警详情", Tags: tags, Input: new(dto.PollingFailureAlertIDPath), Output: new(dto.PollingFailureAlertItem), Auth: true,
|
||||
})
|
||||
Register(alerts, doc, alertsPath, "PUT", "/:id/status", handler.HandleAlert, RouteSpec{
|
||||
Summary: "标记卡轮询失败告警已处理", Description: "只有未处理告警可标记；关闭该卡轮询请使用资产轮询开关接口。", Tags: tags,
|
||||
Input: new(dto.HandlePollingFailureAlertRequest), Output: nil, Auth: true,
|
||||
})
|
||||
|
||||
Register(router, doc, basePath, "GET", "/iot-cards/:id/polling-state", handler.CardState, RouteSpec{
|
||||
Summary: "查询单卡轮询状态", Description: "各检查类型下次到期、最近成功、连续失败与最近错误、是否执行中；只读。", Tags: tags,
|
||||
Input: new(dto.CardPollingStatePath), Output: new(dto.CardPollingStateResponse), Auth: true,
|
||||
})
|
||||
Register(router, doc, basePath, "GET", "/polling-stats/carrier-queues", handler.CarrierStats, RouteSpec{
|
||||
Summary: "按运营商账户的轮询队列与执行监控", Description: "按运营商账户调度的各账户队列长度、已到期数、执行中、上一分钟 QPS、限流等待与拒绝、处理中标记到期、未处理告警。", Tags: tags,
|
||||
Input: nil, Output: new(dto.CarrierPollingStatsResponse), Auth: true,
|
||||
})
|
||||
}
|
||||
622
internal/service/polling/carrier_scheduler_service.go
Normal file
622
internal/service/polling/carrier_scheduler_service.go
Normal file
@@ -0,0 +1,622 @@
|
||||
package polling
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
stderrors "errors"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"github.com/redis/go-redis/v9"
|
||||
"gorm.io/datatypes"
|
||||
"gorm.io/gorm"
|
||||
|
||||
auditinfra "github.com/break/junhong_cmp_fiber/internal/infrastructure/audit"
|
||||
"github.com/break/junhong_cmp_fiber/internal/model"
|
||||
"github.com/break/junhong_cmp_fiber/internal/model/dto"
|
||||
"github.com/break/junhong_cmp_fiber/internal/polling/carrierqueue"
|
||||
"github.com/break/junhong_cmp_fiber/pkg/config"
|
||||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||||
"github.com/break/junhong_cmp_fiber/pkg/errors"
|
||||
"github.com/break/junhong_cmp_fiber/pkg/middleware"
|
||||
"github.com/break/junhong_cmp_fiber/pkg/utils"
|
||||
)
|
||||
|
||||
// CarrierSchedulerService 按运营商账户调度的后台维护与监控：调度参数、请求检查场景、失败告警、单卡轮询状态与按账户监控。
|
||||
// 读取仅超级管理员与平台账号；修改仅超级管理员并写统一审计，下一轮调度生效。
|
||||
type CarrierSchedulerService struct {
|
||||
db *gorm.DB
|
||||
redis *redis.Client
|
||||
auditWriter *auditinfra.Writer
|
||||
store *carrierqueue.Store
|
||||
settings *carrierqueue.SettingsProvider
|
||||
cfg config.CarrierSchedulerConfig
|
||||
}
|
||||
|
||||
// NewCarrierSchedulerService 创建按运营商账户调度后台服务。
|
||||
func NewCarrierSchedulerService(db *gorm.DB, redisClient *redis.Client, auditWriter *auditinfra.Writer, cfg config.CarrierSchedulerConfig) *CarrierSchedulerService {
|
||||
return &CarrierSchedulerService{db: db, redis: redisClient, auditWriter: auditWriter, store: carrierqueue.NewStore(redisClient),
|
||||
settings: carrierqueue.NewDefaultSettingsProvider(db, redisClient, nil), cfg: cfg}
|
||||
}
|
||||
|
||||
func (s *CarrierSchedulerService) processes() int {
|
||||
if s.cfg.SubscriberProcesses > 0 {
|
||||
return s.cfg.SubscriberProcesses
|
||||
}
|
||||
return 1
|
||||
}
|
||||
|
||||
// ListParams 列出全部运营商账户的调度参数（未配置的账户返回全局默认并标记默认）。
|
||||
func (s *CarrierSchedulerService) ListParams(ctx context.Context) (*dto.CarrierPollingParamListResponse, error) {
|
||||
if err := requirePlatformReader(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
names, err := s.carrierNameMap(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var rows []model.CarrierPollingParam
|
||||
if err := s.db.WithContext(ctx).Find(&rows).Error; err != nil {
|
||||
return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询运营商账户调度参数失败")
|
||||
}
|
||||
byCarrier := make(map[uint]model.CarrierPollingParam, len(rows))
|
||||
for _, row := range rows {
|
||||
byCarrier[row.CarrierID] = row
|
||||
}
|
||||
defaults := carrierqueue.MergeParams(model.CarrierPollingParam{}, false, s.cfg)
|
||||
resp := &dto.CarrierPollingParamListResponse{SubscriberProcesses: s.processes(),
|
||||
DefaultInflight: defaults.InflightLimit, DefaultMaxInFlight: defaults.CardMaxInFlight, Items: []dto.CarrierPollingParamItem{}}
|
||||
for _, id := range sortedCarrierIDs(names) {
|
||||
row, configured := byCarrier[id]
|
||||
resp.Items = append(resp.Items, s.paramItem(id, names[id], row, configured))
|
||||
}
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
func (s *CarrierSchedulerService) paramItem(carrierID uint, name string, row model.CarrierPollingParam, configured bool) dto.CarrierPollingParamItem {
|
||||
merged := carrierqueue.MergeParams(row, configured, s.cfg)
|
||||
item := dto.CarrierPollingParamItem{CarrierID: carrierID, CarrierName: name, IsDefault: !configured,
|
||||
InflightLimit: merged.InflightLimit, CardMaxInFlight: merged.CardMaxInFlight, ConsumeCapacity: merged.CardMaxInFlight * s.processes()}
|
||||
if configured {
|
||||
item.PriorityMaxRatio, item.StopResumeMaxInFlight, item.SRInflightLimit = row.PriorityMaxRatio, row.StopResumeMaxInFlight, row.SRInflightLimit
|
||||
item.StopResumeMinInterval, item.StopResumeConfirmDelay = row.StopResumeMinInterval, row.StopResumeConfirmDelay
|
||||
item.ConfiguredInflight, item.ConfiguredMaxInFlight = row.InflightLimit, row.CardMaxInFlight
|
||||
}
|
||||
return item
|
||||
}
|
||||
|
||||
// GetParam 查询单个运营商账户调度参数。
|
||||
func (s *CarrierSchedulerService) GetParam(ctx context.Context, carrierID uint) (*dto.CarrierPollingParamItem, error) {
|
||||
if err := requirePlatformReader(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
carrier, err := s.loadCarrier(ctx, carrierID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
row, configured, err := s.loadParam(ctx, carrierID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
item := s.paramItem(carrierID, carrier.CarrierName, row, configured)
|
||||
return &item, nil
|
||||
}
|
||||
|
||||
// SaveParam 保存运营商账户调度参数；执行上限不得大于当前订阅进程 max_in_flight 之和。
|
||||
func (s *CarrierSchedulerService) SaveParam(ctx context.Context, req *dto.SaveCarrierPollingParamRequest) (*dto.CarrierPollingParamItem, error) {
|
||||
operatorID, err := requireSuperAdminWriter(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
carrier, err := s.loadCarrier(ctx, req.CarrierID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
row, configured, err := s.loadParam(ctx, req.CarrierID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
before := paramState(row, configured)
|
||||
row.CarrierID = req.CarrierID
|
||||
row.InflightLimit, row.CardMaxInFlight, row.StopResumeMaxInFlight = req.InflightLimit, req.CardMaxInFlight, req.StopResumeMaxInFlight
|
||||
row.SRInflightLimit, row.PriorityMaxRatio = req.SRInflightLimit, req.PriorityMaxRatio
|
||||
row.StopResumeMinInterval, row.StopResumeConfirmDelay = req.StopResumeMinInterval, req.StopResumeConfirmDelay
|
||||
merged := carrierqueue.MergeParams(row, true, s.cfg)
|
||||
if capacity := merged.CardMaxInFlight * s.processes(); merged.InflightLimit > capacity {
|
||||
return nil, errors.New(errors.CodeInvalidParam, "执行上限 "+strconv.Itoa(merged.InflightLimit)+" 大于当前订阅进程 max_in_flight 之和 "+
|
||||
strconv.Itoa(capacity)+"（"+strconv.Itoa(s.processes())+" 个进程 × "+strconv.Itoa(merged.CardMaxInFlight)+"），请调低执行上限或调高 max_in_flight")
|
||||
}
|
||||
if !configured {
|
||||
row.Creator = operatorID
|
||||
}
|
||||
row.Updater = operatorID
|
||||
err = runPollingTransaction(ctx, s.db, s.auditWriter, func(tx *gorm.DB) error {
|
||||
if err := tx.WithContext(ctx).Save(&row).Error; err != nil {
|
||||
return errors.Wrap(errors.CodeDatabaseError, err, "保存运营商账户调度参数失败")
|
||||
}
|
||||
return writePollingAudit(ctx, tx, s.auditWriter, auditinfra.PollingInput{
|
||||
ActionCode: constants.AuditActionCarrierPollingParamSaved, Summary: "保存运营商账户调度参数",
|
||||
ResourceType: constants.AuditResourceCarrierPollingParam, ResourceID: req.CarrierID,
|
||||
ResourceKey: "carrier_polling_param:" + uintString(req.CarrierID), DisplayName: carrier.CarrierName, OperatorID: operatorID,
|
||||
IdentitySnapshot: map[string]any{"carrier_id": req.CarrierID, "inflight_limit": row.InflightLimit,
|
||||
"card_max_in_flight": row.CardMaxInFlight, "priority_max_ratio": row.PriorityMaxRatio},
|
||||
BeforeData: before, AfterData: paramState(row, true),
|
||||
})
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
item := s.paramItem(req.CarrierID, carrier.CarrierName, row, true)
|
||||
return &item, nil
|
||||
}
|
||||
|
||||
// DeleteParam 删除运营商账户调度参数，该账户恢复使用全局默认。
|
||||
func (s *CarrierSchedulerService) DeleteParam(ctx context.Context, carrierID uint) error {
|
||||
operatorID, err := requireSuperAdminWriter(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
row, configured, err := s.loadParam(ctx, carrierID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !configured {
|
||||
return errors.New(errors.CodeNotFound, "该运营商账户未单独配置调度参数")
|
||||
}
|
||||
return runPollingTransaction(ctx, s.db, s.auditWriter, func(tx *gorm.DB) error {
|
||||
if err := tx.WithContext(ctx).Delete(&model.CarrierPollingParam{}, row.ID).Error; err != nil {
|
||||
return errors.Wrap(errors.CodeDatabaseError, err, "删除运营商账户调度参数失败")
|
||||
}
|
||||
return writePollingAudit(ctx, tx, s.auditWriter, auditinfra.PollingInput{
|
||||
ActionCode: constants.AuditActionCarrierPollingParamDeleted, Summary: "删除运营商账户调度参数（恢复全局默认）",
|
||||
ResourceType: constants.AuditResourceCarrierPollingParam, ResourceID: carrierID,
|
||||
ResourceKey: "carrier_polling_param:" + uintString(carrierID), DisplayName: uintString(carrierID), OperatorID: operatorID,
|
||||
IdentitySnapshot: map[string]any{"carrier_id": carrierID}, BeforeData: paramState(row, true),
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
func paramState(row model.CarrierPollingParam, configured bool) map[string]any {
|
||||
if !configured {
|
||||
return map[string]any{"configured": false}
|
||||
}
|
||||
return map[string]any{"configured": true, "inflight_limit": row.InflightLimit, "card_max_in_flight": row.CardMaxInFlight,
|
||||
"stop_resume_max_in_flight": row.StopResumeMaxInFlight, "sr_inflight_limit": row.SRInflightLimit,
|
||||
"priority_max_ratio": row.PriorityMaxRatio, "stop_resume_min_interval": row.StopResumeMinInterval,
|
||||
"stop_resume_confirm_delay": row.StopResumeConfirmDelay}
|
||||
}
|
||||
|
||||
func (s *CarrierSchedulerService) loadParam(ctx context.Context, carrierID uint) (model.CarrierPollingParam, bool, error) {
|
||||
var row model.CarrierPollingParam
|
||||
err := s.db.WithContext(ctx).Where("carrier_id = ?", carrierID).First(&row).Error
|
||||
if stderrors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return model.CarrierPollingParam{}, false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return row, false, errors.Wrap(errors.CodeDatabaseError, err, "查询运营商账户调度参数失败")
|
||||
}
|
||||
return row, true, nil
|
||||
}
|
||||
|
||||
func (s *CarrierSchedulerService) loadCarrier(ctx context.Context, carrierID uint) (*model.Carrier, error) {
|
||||
var carrier model.Carrier
|
||||
if err := s.db.WithContext(ctx).Select("id", "carrier_name").First(&carrier, carrierID).Error; err != nil {
|
||||
if stderrors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return nil, errors.New(errors.CodeNotFound, "运营商账户不存在")
|
||||
}
|
||||
return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询运营商账户失败")
|
||||
}
|
||||
return &carrier, nil
|
||||
}
|
||||
|
||||
func (s *CarrierSchedulerService) carrierNameMap(ctx context.Context) (map[uint]string, error) {
|
||||
var carriers []model.Carrier
|
||||
if err := s.db.WithContext(ctx).Select("id", "carrier_name").Find(&carriers).Error; err != nil {
|
||||
return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询运营商账户失败")
|
||||
}
|
||||
names := make(map[uint]string, len(carriers))
|
||||
for _, carrier := range carriers {
|
||||
names[carrier.ID] = carrier.CarrierName
|
||||
}
|
||||
return names, nil
|
||||
}
|
||||
|
||||
// ListScenes 列出请求检查场景。
|
||||
func (s *CarrierSchedulerService) ListScenes(ctx context.Context) ([]dto.PollingCheckSceneItem, error) {
|
||||
if err := requirePlatformReader(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var rows []model.PollingCheckScene
|
||||
if err := s.db.WithContext(ctx).Order("id").Find(&rows).Error; err != nil {
|
||||
return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询请求检查场景失败")
|
||||
}
|
||||
items := make([]dto.PollingCheckSceneItem, 0, len(rows))
|
||||
for _, row := range rows {
|
||||
items = append(items, sceneItem(row))
|
||||
}
|
||||
return items, nil
|
||||
}
|
||||
|
||||
func sceneItem(row model.PollingCheckScene) dto.PollingCheckSceneItem {
|
||||
item := dto.PollingCheckSceneItem{Scene: row.Scene, SceneName: row.SceneName, Cooldown: row.Cooldown, UpdatedAt: row.UpdatedAt,
|
||||
CheckTypes: []string{}, ConfirmDelays: []int{}}
|
||||
_ = json.Unmarshal(row.CheckTypes, &item.CheckTypes)
|
||||
_ = json.Unmarshal(row.ConfirmDelays, &item.ConfirmDelays)
|
||||
return item
|
||||
}
|
||||
|
||||
// UpdateScene 修改请求检查场景的检查类型、确认延迟与冷却；场景由代码预置，不可新增或删除。
|
||||
func (s *CarrierSchedulerService) UpdateScene(ctx context.Context, req *dto.UpdatePollingCheckSceneRequest) (*dto.PollingCheckSceneItem, error) {
|
||||
operatorID, err := requireSuperAdminWriter(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for index := 1; index < len(req.ConfirmDelays); index++ {
|
||||
if req.ConfirmDelays[index] <= req.ConfirmDelays[index-1] {
|
||||
return nil, errors.New(errors.CodeInvalidParam, "延迟确认序列必须递增")
|
||||
}
|
||||
}
|
||||
types := uniqueStrings(req.CheckTypes)
|
||||
var row model.PollingCheckScene
|
||||
if err := s.db.WithContext(ctx).Where("scene = ?", req.Scene).First(&row).Error; err != nil {
|
||||
if stderrors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return nil, errors.New(errors.CodeNotFound, "请求检查场景不存在")
|
||||
}
|
||||
return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询请求检查场景失败")
|
||||
}
|
||||
before := sceneItem(row)
|
||||
typesJSON, _ := json.Marshal(types)
|
||||
delays := req.ConfirmDelays
|
||||
if delays == nil {
|
||||
delays = []int{}
|
||||
}
|
||||
delaysJSON, _ := json.Marshal(delays)
|
||||
row.CheckTypes, row.ConfirmDelays, row.Cooldown, row.Updater = datatypes.JSON(typesJSON), datatypes.JSON(delaysJSON), req.Cooldown, operatorID
|
||||
err = runPollingTransaction(ctx, s.db, s.auditWriter, func(tx *gorm.DB) error {
|
||||
if err := tx.WithContext(ctx).Model(&model.PollingCheckScene{}).Where("id = ?", row.ID).Updates(map[string]any{
|
||||
"check_types": row.CheckTypes, "confirm_delays": row.ConfirmDelays, "cooldown": row.Cooldown, "updater": operatorID, "updated_at": time.Now(),
|
||||
}).Error; err != nil {
|
||||
return errors.Wrap(errors.CodeDatabaseError, err, "修改请求检查场景失败")
|
||||
}
|
||||
return writePollingAudit(ctx, tx, s.auditWriter, auditinfra.PollingInput{
|
||||
ActionCode: constants.AuditActionPollingCheckSceneUpdated, Summary: "修改请求检查场景",
|
||||
ResourceType: constants.AuditResourcePollingCheckScene, ResourceID: row.ID,
|
||||
ResourceKey: "polling_check_scene:" + row.Scene, DisplayName: row.SceneName, OperatorID: operatorID,
|
||||
IdentitySnapshot: map[string]any{"scene": row.Scene, "scene_name": row.SceneName},
|
||||
BeforeData: map[string]any{"check_types": before.CheckTypes, "confirm_delays": before.ConfirmDelays, "cooldown": before.Cooldown},
|
||||
AfterData: map[string]any{"check_types": types, "confirm_delays": delays, "cooldown": req.Cooldown},
|
||||
})
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
item := sceneItem(row)
|
||||
return &item, nil
|
||||
}
|
||||
|
||||
// ListAlerts 分页查询卡轮询失败告警。
|
||||
func (s *CarrierSchedulerService) ListAlerts(ctx context.Context, req *dto.PollingFailureAlertListRequest) (*dto.PollingFailureAlertPageResult, error) {
|
||||
if err := requirePlatformReader(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
page, size := normalizePage(req.Page, req.PageSize)
|
||||
query := s.db.WithContext(ctx).Model(&model.PollingFailureAlert{})
|
||||
if req.CarrierID != nil {
|
||||
query = query.Where("carrier_id = ?", *req.CarrierID)
|
||||
}
|
||||
if req.CheckType != "" {
|
||||
query = query.Where("check_type = ?", req.CheckType)
|
||||
}
|
||||
if req.ErrorCategory != "" {
|
||||
query = query.Where("error_category = ?", req.ErrorCategory)
|
||||
}
|
||||
if req.Status != nil {
|
||||
query = query.Where("status = ?", *req.Status)
|
||||
}
|
||||
if req.AlertLevel != "" {
|
||||
query = query.Where("alert_level = ?", req.AlertLevel)
|
||||
}
|
||||
if req.ICCID != "" {
|
||||
query = query.Where("iccid = ?", req.ICCID)
|
||||
}
|
||||
start, end, err := utils.ParseTimeRange(req.StartTime, req.EndTime)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if start != nil {
|
||||
query = query.Where("last_failed_at >= ?", *start)
|
||||
}
|
||||
if end != nil {
|
||||
query = query.Where("last_failed_at <= ?", *end)
|
||||
}
|
||||
var total int64
|
||||
if err := query.Count(&total).Error; err != nil {
|
||||
return nil, errors.Wrap(errors.CodeDatabaseError, err, "统计卡轮询失败告警失败")
|
||||
}
|
||||
var rows []model.PollingFailureAlert
|
||||
if err := query.Order("last_failed_at DESC NULLS LAST, id DESC").Offset((page - 1) * size).Limit(size).Find(&rows).Error; err != nil {
|
||||
return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询卡轮询失败告警失败")
|
||||
}
|
||||
names, err := s.carrierNameMap(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
result := &dto.PollingFailureAlertPageResult{Total: total, Page: page, PageSize: size, List: make([]dto.PollingFailureAlertItem, 0, len(rows))}
|
||||
for _, row := range rows {
|
||||
result.List = append(result.List, alertItem(row, names))
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func alertItem(row model.PollingFailureAlert, names map[uint]string) dto.PollingFailureAlertItem {
|
||||
return dto.PollingFailureAlertItem{
|
||||
ID: row.ID, AlertLevel: row.AlertLevel, CardID: row.CardID, ICCID: row.ICCID, CarrierID: row.CarrierID,
|
||||
CarrierName: names[row.CarrierID], CheckType: row.CheckType, CheckTypeName: constants.CheckTypeName(row.CheckType),
|
||||
ErrorCategory: row.ErrorCategory, FailCount: row.FailCount, FirstFailedAt: row.FirstFailedAt, LastFailedAt: row.LastFailedAt,
|
||||
LastError: row.LastError, Status: row.Status, StatusName: constants.PollingAlertStatusName(row.Status),
|
||||
HandledBy: row.HandledBy, HandledAt: row.HandledAt, HandleRemark: row.HandleRemark, RecoveredAt: row.RecoveredAt, CreatedAt: row.CreatedAt,
|
||||
}
|
||||
}
|
||||
|
||||
// AlertSummary 未处理卡级告警按运营商账户 × 错误类别汇总。
|
||||
func (s *CarrierSchedulerService) AlertSummary(ctx context.Context) ([]dto.PollingFailureAlertSummaryItem, error) {
|
||||
if err := requirePlatformReader(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var rows []struct {
|
||||
CarrierID uint
|
||||
ErrorCategory string
|
||||
OpenCount int64
|
||||
}
|
||||
if err := s.db.WithContext(ctx).Model(&model.PollingFailureAlert{}).
|
||||
Select("carrier_id, error_category, COUNT(*) AS open_count").
|
||||
Where("status = ? AND alert_level = ?", constants.PollingAlertStatusOpen, constants.PollingAlertLevelCard).
|
||||
Group("carrier_id, error_category").Order("open_count DESC").Scan(&rows).Error; err != nil {
|
||||
return nil, errors.Wrap(errors.CodeDatabaseError, err, "汇总卡轮询失败告警失败")
|
||||
}
|
||||
names, err := s.carrierNameMap(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
items := make([]dto.PollingFailureAlertSummaryItem, 0, len(rows))
|
||||
for _, row := range rows {
|
||||
items = append(items, dto.PollingFailureAlertSummaryItem{CarrierID: row.CarrierID, CarrierName: names[row.CarrierID],
|
||||
ErrorCategory: row.ErrorCategory, OpenCount: row.OpenCount})
|
||||
}
|
||||
return items, nil
|
||||
}
|
||||
|
||||
// GetAlert 查询告警详情。
|
||||
func (s *CarrierSchedulerService) GetAlert(ctx context.Context, id uint) (*dto.PollingFailureAlertItem, error) {
|
||||
if err := requirePlatformReader(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
row, err := s.loadAlert(ctx, id)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
names, err := s.carrierNameMap(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
item := alertItem(*row, names)
|
||||
return &item, nil
|
||||
}
|
||||
|
||||
// HandleAlert 把未处理告警标记为已处理。
|
||||
func (s *CarrierSchedulerService) HandleAlert(ctx context.Context, req *dto.HandlePollingFailureAlertRequest) error {
|
||||
operatorID, err := requireSuperAdminWriterOrPlatform(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
row, err := s.loadAlert(ctx, req.ID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if row.Status != constants.PollingAlertStatusOpen {
|
||||
return errors.New(errors.CodeInvalidStatus, "只有未处理的告警可以标记已处理")
|
||||
}
|
||||
now := time.Now()
|
||||
return runPollingTransaction(ctx, s.db, s.auditWriter, func(tx *gorm.DB) error {
|
||||
result := tx.WithContext(ctx).Model(&model.PollingFailureAlert{}).Where("id = ? AND status = ?", row.ID, constants.PollingAlertStatusOpen).
|
||||
Updates(map[string]any{"status": req.Status, "handled_by": operatorID, "handled_at": now, "handle_remark": req.Remark, "updated_at": now})
|
||||
if result.Error != nil {
|
||||
return errors.Wrap(errors.CodeDatabaseError, result.Error, "处理卡轮询失败告警失败")
|
||||
}
|
||||
if result.RowsAffected == 0 {
|
||||
return errors.New(errors.CodeConflict, "告警状态已变化，请刷新后重试")
|
||||
}
|
||||
return writePollingAudit(ctx, tx, s.auditWriter, auditinfra.PollingInput{
|
||||
ActionCode: constants.AuditActionPollingFailureAlertHandled, Summary: "标记卡轮询失败告警已处理",
|
||||
ResourceType: constants.AuditResourcePollingFailureAlert, ResourceID: row.ID,
|
||||
ResourceKey: "polling_failure_alert:" + uintString(row.ID), DisplayName: row.ICCID, OperatorID: operatorID,
|
||||
IdentitySnapshot: map[string]any{"id": row.ID, "alert_level": row.AlertLevel, "card_id": row.CardID,
|
||||
"carrier_id": row.CarrierID, "check_type": row.CheckType, "status": row.Status},
|
||||
BeforeData: map[string]any{"status": row.Status}, AfterData: map[string]any{"status": req.Status, "handle_remark": req.Remark},
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
func (s *CarrierSchedulerService) loadAlert(ctx context.Context, id uint) (*model.PollingFailureAlert, error) {
|
||||
var row model.PollingFailureAlert
|
||||
if err := s.db.WithContext(ctx).First(&row, id).Error; err != nil {
|
||||
if stderrors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return nil, errors.New(errors.CodeNotFound, "卡轮询失败告警不存在")
|
||||
}
|
||||
return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询卡轮询失败告警失败")
|
||||
}
|
||||
return &row, nil
|
||||
}
|
||||
|
||||
// CardState 查询单卡轮询状态（只读）。
|
||||
func (s *CarrierSchedulerService) CardState(ctx context.Context, cardID uint) (*dto.CardPollingStateResponse, error) {
|
||||
if err := requirePlatformReader(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var card model.IotCard
|
||||
if err := s.db.WithContext(ctx).Select("id", "iccid", "carrier_id", "enable_polling").First(&card, cardID).Error; err != nil {
|
||||
if stderrors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return nil, errors.New(errors.CodeNotFound, "IoT卡不存在")
|
||||
}
|
||||
return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询IoT卡失败")
|
||||
}
|
||||
resp := &dto.CardPollingStateResponse{CardID: card.ID, ICCID: card.ICCID, CarrierID: card.CarrierID,
|
||||
SchedulerMode: s.settings.Current(ctx).Mode, EnablePolling: card.EnablePolling, Queue: "none", Types: []dto.CardPollingTypeState{}}
|
||||
due, err := s.store.ReadDue(ctx, card.ID)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(errors.CodeRedisError, err, "读取卡轮询到期信息失败")
|
||||
}
|
||||
carrierID := due.CarrierID
|
||||
if carrierID == 0 {
|
||||
carrierID = card.CarrierID
|
||||
}
|
||||
member := strconv.FormatUint(uint64(card.ID), 10)
|
||||
if score, err := s.redis.ZScore(ctx, constants.RedisCarrierPriorityQueueKey(carrierID), member).Result(); err == nil {
|
||||
resp.Queue, resp.QueueScoreAt = "priority", msTime(int64(score))
|
||||
} else if score, err := s.redis.ZScore(ctx, constants.RedisCarrierQueueKey(carrierID), member).Result(); err == nil {
|
||||
resp.Queue, resp.QueueScoreAt = "queue", msTime(int64(score))
|
||||
}
|
||||
if ttl, err := s.redis.PTTL(ctx, constants.RedisCardClaimKey(card.ID)).Result(); err == nil && ttl > 0 {
|
||||
resp.Running = true
|
||||
expire := time.Now().Add(ttl)
|
||||
resp.ClaimExpireAt = &expire
|
||||
}
|
||||
resp.PriorityMark = due.Priority
|
||||
for _, checkType := range constants.AllCheckTypes() {
|
||||
item, ok := due.Types[checkType]
|
||||
if !ok || (!item.HasDue && item.FailCount == 0 && item.LastOK.IsZero()) {
|
||||
continue
|
||||
}
|
||||
state := dto.CardPollingTypeState{CheckType: checkType, CheckTypeName: constants.CheckTypeName(checkType),
|
||||
FailCount: item.FailCount, LastError: item.LastError, PendingConfirm: []int{}}
|
||||
if item.HasDue {
|
||||
state.NextDueAt = timePtr(item.Due)
|
||||
}
|
||||
if !item.LastOK.IsZero() {
|
||||
state.LastSuccessAt = timePtr(item.LastOK)
|
||||
}
|
||||
if !item.FirstFailed.IsZero() {
|
||||
state.FirstFailedAt = timePtr(item.FirstFailed)
|
||||
}
|
||||
for _, delay := range item.Confirm {
|
||||
state.PendingConfirm = append(state.PendingConfirm, int(delay/time.Second))
|
||||
}
|
||||
resp.Types = append(resp.Types, state)
|
||||
}
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
// CarrierStats 按运营商账户的轮询监控。
|
||||
func (s *CarrierSchedulerService) CarrierStats(ctx context.Context) (*dto.CarrierPollingStatsResponse, error) {
|
||||
if err := requirePlatformReader(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
names, err := s.carrierNameMap(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var params []model.CarrierPollingParam
|
||||
if err := s.db.WithContext(ctx).Find(¶ms).Error; err != nil {
|
||||
return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询运营商账户调度参数失败")
|
||||
}
|
||||
paramByCarrier := map[uint]model.CarrierPollingParam{}
|
||||
for _, row := range params {
|
||||
paramByCarrier[row.CarrierID] = row
|
||||
}
|
||||
var alertRows []struct {
|
||||
CarrierID uint
|
||||
ErrorCategory string
|
||||
Count int64
|
||||
}
|
||||
if err := s.db.WithContext(ctx).Model(&model.PollingFailureAlert{}).Select("carrier_id, error_category, COUNT(*) AS count").
|
||||
Where("status = ? AND alert_level = ?", constants.PollingAlertStatusOpen, constants.PollingAlertLevelCard).
|
||||
Group("carrier_id, error_category").Scan(&alertRows).Error; err != nil {
|
||||
return nil, errors.Wrap(errors.CodeDatabaseError, err, "统计卡轮询失败告警失败")
|
||||
}
|
||||
state, _ := s.redis.Get(ctx, constants.RedisCarrierSchedulerInitStateKey).Result()
|
||||
resp := &dto.CarrierPollingStatsResponse{SchedulerMode: s.settings.Current(ctx).Mode, InitState: state, Items: []dto.CarrierPollingStatsItem{}}
|
||||
now := time.Now()
|
||||
nowMS := strconv.FormatInt(now.UnixMilli(), 10)
|
||||
lastMinute := now.Add(-time.Minute).In(time.FixedZone("CST", 8*3600)).Format("200601021504")
|
||||
for _, carrierID := range sortedCarrierIDs(names) {
|
||||
row, configured := paramByCarrier[carrierID]
|
||||
item := dto.CarrierPollingStatsItem{CarrierID: carrierID, CarrierName: names[carrierID],
|
||||
InflightLimit: carrierqueue.MergeParams(row, configured, s.cfg).InflightLimit, OpenAlertsByType: map[string]int64{}}
|
||||
pipe := s.redis.Pipeline()
|
||||
priorityLen := pipe.ZCard(ctx, constants.RedisCarrierPriorityQueueKey(carrierID))
|
||||
queueLen := pipe.ZCard(ctx, constants.RedisCarrierQueueKey(carrierID))
|
||||
priorityDue := pipe.ZCount(ctx, constants.RedisCarrierPriorityQueueKey(carrierID), "-inf", nowMS)
|
||||
queueDue := pipe.ZCount(ctx, constants.RedisCarrierQueueKey(carrierID), "-inf", nowMS)
|
||||
running := pipe.ZCount(ctx, constants.RedisCarrierRunningKey(carrierID), nowMS, "+inf")
|
||||
expired := pipe.HGetAll(ctx, constants.RedisCarrierExpiredClaimsKey(carrierID))
|
||||
var statCmds []*redis.MapStringStringCmd
|
||||
for _, op := range []string{constants.GatewayOpQueryStatus, constants.GatewayOpQueryFlow, constants.GatewayOpQueryRealname} {
|
||||
statCmds = append(statCmds, pipe.HGetAll(ctx, constants.RedisGatewayCallStatKey(carrierID, op, lastMinute)))
|
||||
}
|
||||
_, _ = pipe.Exec(ctx)
|
||||
item.PriorityLength, item.QueueLength = priorityLen.Val(), queueLen.Val()
|
||||
item.DueCount, item.Running = priorityDue.Val()+queueDue.Val(), running.Val()
|
||||
cutoff := now.Add(-time.Hour).Unix() / 60
|
||||
for minute, value := range expired.Val() {
|
||||
if m, err := strconv.ParseInt(minute, 10, 64); err == nil && m >= cutoff {
|
||||
count, _ := strconv.ParseInt(value, 10, 64)
|
||||
item.ExpiredClaims1h += count
|
||||
}
|
||||
}
|
||||
var attempts int64
|
||||
for _, cmd := range statCmds {
|
||||
values := cmd.Val()
|
||||
attempts += atoi64(values["attempts"])
|
||||
item.RateLimitWaits += atoi64(values["rate_wait"])
|
||||
item.RateLimited += atoi64(values[constants.GatewayErrorRateLimited])
|
||||
}
|
||||
item.QPSLastMinute = float64(attempts) / 60
|
||||
for _, alert := range alertRows {
|
||||
if alert.CarrierID == carrierID {
|
||||
item.OpenAlerts += alert.Count
|
||||
item.OpenAlertsByType[alert.ErrorCategory] += alert.Count
|
||||
}
|
||||
}
|
||||
resp.Items = append(resp.Items, item)
|
||||
}
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
// requireSuperAdminWriterOrPlatform 校验告警处理权限：超级管理员与平台账号均可，返回操作者 ID。
|
||||
func requireSuperAdminWriterOrPlatform(ctx context.Context) (uint, error) {
|
||||
if err := requirePlatformReader(ctx); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
operatorID := middleware.GetUserIDFromContext(ctx)
|
||||
if operatorID == 0 {
|
||||
return 0, errors.New(errors.CodeUnauthorized, "未授权访问")
|
||||
}
|
||||
return operatorID, nil
|
||||
}
|
||||
|
||||
func sortedCarrierIDs(names map[uint]string) []uint {
|
||||
ids := make([]uint, 0, len(names))
|
||||
for id := range names {
|
||||
ids = append(ids, id)
|
||||
}
|
||||
for i := 1; i < len(ids); i++ {
|
||||
for j := i; j > 0 && ids[j] < ids[j-1]; j-- {
|
||||
ids[j], ids[j-1] = ids[j-1], ids[j]
|
||||
}
|
||||
}
|
||||
return ids
|
||||
}
|
||||
|
||||
func msTime(ms int64) *time.Time {
|
||||
t := time.UnixMilli(ms)
|
||||
return &t
|
||||
}
|
||||
|
||||
func timePtr(t time.Time) *time.Time {
|
||||
return &t
|
||||
}
|
||||
|
||||
func atoi64(s string) int64 {
|
||||
value, _ := strconv.ParseInt(s, 10, 64)
|
||||
return value
|
||||
}
|
||||
@@ -602,6 +602,14 @@ const (
|
||||
AuditActionGatewayErrorRuleStatusUpdated = "gateway_error_rule.update_status"
|
||||
// AuditActionGatewayErrorRuleDeleted 表示删除上游错误分类规则。
|
||||
AuditActionGatewayErrorRuleDeleted = "gateway_error_rule.delete"
|
||||
// AuditActionCarrierPollingParamSaved 表示保存运营商账户调度参数。
|
||||
AuditActionCarrierPollingParamSaved = "carrier_polling_param.save"
|
||||
// AuditActionCarrierPollingParamDeleted 表示删除运营商账户调度参数（恢复全局默认）。
|
||||
AuditActionCarrierPollingParamDeleted = "carrier_polling_param.delete"
|
||||
// AuditActionPollingCheckSceneUpdated 表示修改请求检查场景配置。
|
||||
AuditActionPollingCheckSceneUpdated = "polling_check_scene.update"
|
||||
// AuditActionPollingFailureAlertHandled 表示处理卡轮询失败告警。
|
||||
AuditActionPollingFailureAlertHandled = "polling_failure_alert.handle"
|
||||
// AuditActionPollingConcurrencyUpdated 表示更新轮询任务并发配置。
|
||||
AuditActionPollingConcurrencyUpdated = "polling_concurrency.update"
|
||||
// AuditActionPollingConcurrencyReset 表示人工重置轮询并发计数。
|
||||
@@ -781,6 +789,12 @@ const (
|
||||
AuditResourceRateLimitGroup = "rate_limit_group"
|
||||
// AuditResourceGatewayErrorRule 表示上游错误分类规则资源。
|
||||
AuditResourceGatewayErrorRule = "gateway_error_rule"
|
||||
// AuditResourceCarrierPollingParam 表示运营商账户调度参数资源。
|
||||
AuditResourceCarrierPollingParam = "carrier_polling_param"
|
||||
// AuditResourcePollingCheckScene 表示请求检查场景资源。
|
||||
AuditResourcePollingCheckScene = "polling_check_scene"
|
||||
// AuditResourcePollingFailureAlert 表示卡轮询失败告警资源。
|
||||
AuditResourcePollingFailureAlert = "polling_failure_alert"
|
||||
// AuditResourcePollingConcurrencyConfig 表示轮询并发配置资源。
|
||||
AuditResourcePollingConcurrencyConfig = "polling_concurrency"
|
||||
// AuditResourcePollingAlertRule 表示轮询告警规则资源。
|
||||
|
||||
Reference in New Issue
Block a user