syncalert

package
v1.4.0 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Nov 8, 2026 License: Apache-2.0 Imports: 6 Imported by: 0

Documentation

Overview

Package syncalert 定义数据同步作业告警在 sync(产生信号)与 monitoring(判定、投递)之间的契约,两边都只依赖本包。 Package syncalert defines the contract for sync job alerts between sync (which produces signals) and monitoring (which evaluates and delivers them); both sides depend only on this package.

Index

Constants

View Source
const (
	// RuleSubmitFailed 是提交到引擎被明确拒绝,或者找不到可用的引擎端点(A1)。
	// RuleSubmitFailed fires when the engine definitely rejects a submit or no engine endpoint is usable (A1).
	RuleSubmitFailed = "sync_submit_failed"
	// RuleSubmitTimeout 是提交结果一直未知,过了截止时间被判失败(A2)。
	// RuleSubmitTimeout fires when an unknown submit outcome passes its deadline and is failed (A2).
	RuleSubmitTimeout = "sync_submit_timeout"
	// RuleJobFailed 是作业转为失败,包括引擎已无此作业(B1)。
	// RuleJobFailed fires when a job turns failed, including when the engine no longer has it (B1).
	RuleJobFailed = "sync_job_failed"
	// RuleStopTimeout 是停止请求过了截止时间作业仍未停下(C1)。
	// RuleStopTimeout fires when a job is still running after its stop deadline (C1).
	RuleStopTimeout = "sync_stop_timeout"
	// RuleCDCDelayHigh 是 CDC 实时作业读取延迟持续高于阈值(B4);阈值与持续时间按任务配置,判定在 sync 里完成。
	// RuleCDCDelayHigh fires when a CDC streaming job's read delay stays above its threshold (B4); the threshold and
	// duration are configured per task and evaluated in sync.
	RuleCDCDelayHigh = "sync_cdc_delay_high"
)

作业级规则:用户在任务上配置了才生效。 Job-level rules: they only apply when the user configures alerts on the task.

View Source
const (
	// RuleStatusUnavailable 是集群的作业状态持续获取不到(D1)。
	// RuleStatusUnavailable fires when job status of a cluster keeps being unavailable (D1).
	RuleStatusUnavailable = "sync_status_unavailable"
	// RuleEventChannelDegraded 是集群事件通道从健康变为降级(D2)。
	// RuleEventChannelDegraded fires when the cluster event channel turns from healthy to degraded (D2).
	RuleEventChannelDegraded = "sync_event_channel_degraded"
)

集群级规则:每个集群默认开启,与"节点离线"一样作为默认规则。 Cluster-level rules: enabled by default on every cluster, like "node offline".

View Source
const (
	ResolveReasonRecovered = "recovered"
	// ResolveReasonJobEnded 是作业已停止、失败或完成。
	// ResolveReasonJobEnded means the job stopped, failed or finished.
	ResolveReasonJobEnded = "job_ended"
	// ResolveReasonChannelOff 是事件通道已关闭、待重启,或集群不支持事件通道。
	// ResolveReasonChannelOff means the event channel is off, pending restart, or unsupported by the cluster.
	ResolveReasonChannelOff = "channel_unavailable"
	// ResolveReasonRuleRemoved 是任务告警配置去掉了 B4,或策略被停用。
	// ResolveReasonRuleRemoved means B4 was removed from the task alert config or the policy was disabled.
	ResolveReasonRuleRemoved = "rule_removed"
	// ResolveReasonRuleChanged 是告警中修改了阈值或持续时间、按新参数重新计时后延迟已不再超标;不算恢复,不发恢复通知。
	// ResolveReasonRuleChanged means the threshold or duration changed while alerting and, re-timed with the new
	// values, the delay is no longer above it; this is not a recovery and sends no recovery notification.
	ResolveReasonRuleChanged = "rule_changed"
)

Resolve 的原因:空或 recovered 表示恢复,按策略发恢复通知;其余原因只关闭阶段、不发通知,告警中心显示原因。 Resolve reasons: empty or recovered means recovery and sends a recovery notification per policy; every other reason only closes the episode silently and the alert center shows the reason.

View Source
const (
	// CDCDelayStateHigh 是本轮延迟高于阈值。
	// CDCDelayStateHigh means the delay of this round is above the threshold.
	CDCDelayStateHigh = "high"
	// CDCDelayStateUnknown 是收不到延迟数据,无法确认是否恢复。
	// CDCDelayStateUnknown means no delay data arrives and recovery cannot be confirmed.
	CDCDelayStateUnknown = "unknown"
	// CDCDelayStateRecovered 是延迟已回落。
	// CDCDelayStateRecovered means the delay dropped back.
	CDCDelayStateRecovered = "recovered"
	// CDCDelayStateRecovering 是告警中出现一轮低于阈值:还没连续两轮,阶段保留,只刷新为当前值。
	// CDCDelayStateRecovering means one round below the threshold while alerting: not yet two in a row, so the
	// episode stays open and is only refreshed with the current values.
	CDCDelayStateRecovering = "recovering"
	// CDCDelayStateClosed 是阶段因恢复以外的原因关闭(作业结束、去掉规则、改参数、事件通道不可用、配置删除、人工关闭),
	// 原因见明细的 resolve_reason;不再沿用关闭前的 high / recovering / unknown。
	// CDCDelayStateClosed means the episode closed for a reason other than recovery (job ended, rule removed, rule
	// changed, channel unavailable, config deleted, manual close); the reason is the detail's resolve_reason. The
	// pre-close high / recovering / unknown is no longer kept.
	CDCDelayStateClosed = "closed"
	// CDCDelayCloseManual 是人工关闭时明细 resolve_reason 的取值(阶段本身记为 closed)。closed 状态只存在于已关闭
	// 阶段的明细里,不会出现在 webhook 的 delay_state 中(静默关闭不发通知)。
	// CDCDelayCloseManual is the detail resolve_reason of a manual close (the episode itself records closed). The
	// closed state only lives in a closed episode's detail and never appears in a webhook delay_state (silent closes
	// send nothing).
	CDCDelayCloseManual = "manual"

	// CDCDelayUnknownNoProgress 是没有延迟数据、读取条数也没有增长。
	// CDCDelayUnknownNoProgress means no delay data and no growth in the read count.
	CDCDelayUnknownNoProgress = "no_progress"
	// CDCDelayUnknownChannelDegraded 是没有延迟数据、事件通道降级。
	// CDCDelayUnknownChannelDegraded means no delay data while the event channel is degraded.
	CDCDelayUnknownChannelDegraded = "channel_degraded"
	// CDCDelayUnknownStaleSamples 是收到了延迟样本,但全部按时间差判为重投的旧样本(例如引擎时钟回拨),没有可计入的数据。
	// CDCDelayUnknownStaleSamples means delay samples arrived but all were judged replayed old samples by their offset
	// (e.g. the engine clock stepped back), so there is no data to count.
	CDCDelayUnknownStaleSamples = "stale_samples"

	// CDCDelayRecoveryBySamples 是延迟样本连续两轮低于阈值。
	// CDCDelayRecoveryBySamples means delay samples stayed below the threshold for two rounds.
	CDCDelayRecoveryBySamples = "delay_samples"
	// CDCDelayRecoveryByReadCount 是没有延迟事件、读取条数在增长,据此推断延迟已低于 1 秒。
	// CDCDelayRecoveryByReadCount means no delay events while the read count grows, so the delay is inferred below 1s.
	CDCDelayRecoveryByReadCount = "read_count"
)

CDC 延迟明细里的状态与依据。 States and bases used in the CDC delay detail.

View Source
const (
	// DefaultCooldownMinutes 是未填写时的提醒间隔。
	// DefaultCooldownMinutes is the reminder interval when none is given.
	DefaultCooldownMinutes = 60
	// MaxCooldownMinutes 是提醒间隔上限(一天)。
	// MaxCooldownMinutes caps the reminder interval (one day).
	MaxCooldownMinutes = 1440

	// DefaultCDCDelayThresholdSeconds 是 B4 未填写阈值(或填 0)时的默认值。
	// DefaultCDCDelayThresholdSeconds is the B4 threshold when none (or 0) is given.
	DefaultCDCDelayThresholdSeconds = 60
	// MinCDCDelayThresholdSeconds 是阈值下限:引擎每 10 秒投递一批,数据库与 worker 的时钟差也会算进延迟。
	// MinCDCDelayThresholdSeconds is the lower bound: the engine delivers every 10s and DB/worker clock skew adds up.
	MinCDCDelayThresholdSeconds = 10
	// MaxCDCDelayThresholdSeconds 是阈值上限(24 小时)。
	// MaxCDCDelayThresholdSeconds is the upper bound (24 hours).
	MaxCDCDelayThresholdSeconds = 86400
	// DefaultCDCDelayDurationMinutes 是 B4 未填写持续时间(或填 0)时的默认值。
	// DefaultCDCDelayDurationMinutes is the B4 duration when none (or 0) is given.
	DefaultCDCDelayDurationMinutes = 5
	// MinCDCDelayDurationMinutes、MaxCDCDelayDurationMinutes 是持续时间范围。
	// MinCDCDelayDurationMinutes and MaxCDCDelayDurationMinutes bound the duration.
	MinCDCDelayDurationMinutes = 1
	MaxCDCDelayDurationMinutes = 60
)

默认值与范围。 Defaults and ranges.

View Source
const MaxMessageLength = 500

MaxMessageLength 是信号里错误信息的最大长度(字符),也是告警中心与通知里展示的上限:存多少就展示多少, 避免存 2000、展示 500 两套口径。500 字符在企业微信、钉钉单条文本(约 2048 字节)里仍能放下其余字段。 MaxMessageLength is the maximum length (in characters) of a signal's message and also the display cap in the alert center and notifications: what is stored is what is shown, instead of storing 2000 and showing 500. 500 characters still leave room for the other fields within a WeCom / DingTalk text message (about 2048 bytes).

Variables

View Source
var ErrConflict = errors.New("task already has a job alert policy")

ErrConflict 表示同一个任务已经有一条作业告警策略(HTTP 409)。 ErrConflict means the task already has a job alert policy (HTTP 409).

View Source
var ErrInvalidConfig = errors.New("invalid task alert config")

ErrInvalidConfig 表示任务告警配置不合法(HTTP 400)。 ErrInvalidConfig means the task alert configuration is invalid (HTTP 400).

Functions

func ClusterRules

func ClusterRules() []string

ClusterRules 返回集群级默认规则。 ClusterRules returns the cluster-level default rules.

func FormatDelayZH

func FormatDelayZH(ms int64) string

FormatDelayZH 把毫秒延迟格式化为"3分12秒"这样的中文时长(不足 1 分钟为"45秒",超过 1 小时为"1小时2分")。 FormatDelayZH formats a delay in milliseconds as a Chinese duration such as "3分12秒" ("45秒" under a minute, "1小时2分" over an hour).

func InvalidConfig

func InvalidConfig(format string, args ...any) error

InvalidConfig 包装一条具体的校验错误,保留 ErrInvalidConfig 以便映射为 400。 InvalidConfig wraps one concrete validation error while keeping ErrInvalidConfig for the 400 mapping.

func IsClusterRule

func IsClusterRule(rule string) bool

IsClusterRule 判断规则键是否属于集群级规则。 IsClusterRule reports whether the rule key is a cluster-level rule.

func IsTaskRule

func IsTaskRule(rule string) bool

IsTaskRule 判断规则键是否属于作业级规则。 IsTaskRule reports whether the rule key is a job-level rule.

func TaskRules

func TaskRules() []string

TaskRules 返回任务上可以勾选的规则,顺序固定。 TaskRules returns the rules a task can select, in a fixed order.

func TruncateMessage

func TruncateMessage(message string, limit int) string

TruncateMessage 把错误信息截到 limit 个字符,超出部分用省略号表示。 TruncateMessage cuts a message to limit characters and marks the cut with an ellipsis.

func ValidateCDCDelayParams

func ValidateCDCDelayParams(thresholdSeconds, durationMinutes int) error

ValidateCDCDelayParams 校验 B4 的阈值与持续时间;调用前应已补默认值(0 视为非法)。 ValidateCDCDelayParams checks the B4 threshold and duration; defaults must already be applied (0 is invalid).

Types

type AcceptingReporter

type AcceptingReporter interface {
	TryReportSyncAlert(signal Signal) bool
}

AcceptingReporter 是可选接口:TryReportSyncAlert 与 ReportSyncAlert 相同(非阻塞),但返回信号是否进了队列; 被丢弃的 B4 Refresh 由 sync 在下一轮重发。 AcceptingReporter is optional: TryReportSyncAlert works like ReportSyncAlert (non-blocking) but reports whether the signal was queued, so sync can resend a dropped B4 Refresh next round.

type CDCDelayDetail

type CDCDelayDetail struct {
	State          string `json:"state"`
	CurrentDelayMs int64  `json:"current_delay_ms"`
	// LastDelayMs 是最后一次有效延迟样本;delay_state=unknown 时 current_delay_ms 为 null,用本字段保留旧值。
	// LastDelayMs is the last valid delay sample; when delay_state=unknown, current_delay_ms is null and this keeps the old value.
	LastDelayMs      int64     `json:"last_delay_ms,omitempty"`
	WindowMaxDelayMs int64     `json:"window_max_delay_ms"`
	ThresholdSeconds int       `json:"threshold_seconds"`
	DurationMinutes  int       `json:"duration_minutes"`
	AboveSince       time.Time `json:"above_since,omitempty"`
	LastSampleAt     time.Time `json:"last_sample_at,omitempty"`
	UnknownReason    string    `json:"unknown_reason,omitempty"`
	UnknownSince     time.Time `json:"unknown_since,omitempty"`
	// RemindersPaused 为 true 时 monitoring 不再发冷却提醒("无法确认"超过 24 小时);阶段保留。
	// RemindersPaused stops monitoring's cooldown reminders ("unconfirmable" for over 24h); the episode stays open.
	RemindersPaused bool   `json:"reminders_paused,omitempty"`
	RecoveryBasis   string `json:"recovery_basis,omitempty"`
	// DelayInferred 为 true 时没有可用的延迟样本,状态是根据读取条数推断的:CurrentDelayMs 不是当前值(为 0),
	// webhook 的 current_delay_ms 输出 null。
	// DelayInferred means no delay sample backs this state and it was inferred from the read count: CurrentDelayMs is
	// not a current value (it is 0) and the webhook sends current_delay_ms as null.
	DelayInferred bool `json:"delay_inferred,omitempty"`
	// ResolveReason 只在 State 为 closed 时填写,与阶段的 resolve_reason 相同;人工关闭记为 manual(阶段为 closed)。
	// ResolveReason is only set when State is closed and equals the episode's resolve_reason; a manual close is
	// recorded as manual (the episode says closed).
	ResolveReason string `json:"resolve_reason,omitempty"`
}

CDCDelayDetail 是 B4 信号携带的结构化延迟明细;通知模板和 webhook 用它展示当前延迟等数值,而不是塞进错误摘要。 CDCDelayDetail is the structured delay detail carried by B4 signals; notification templates and webhooks render the current delay and friends from it instead of stuffing them into the error summary.

type Reporter

type Reporter interface {
	ReportSyncAlert(signal Signal)
}

Reporter 接收 sync 产生的告警信号;实现必须非阻塞。 Reporter receives alert signals produced by sync; implementations must not block.

type ResolveScope

type ResolveScope string

ResolveScope 指定 Resolve 关闭哪些告警阶段。 ResolveScope selects which alert episodes a Resolve closes.

const (
	// ResolveTaskRunHealthy 关闭任务的 A1、A2、B1:批作业运行成功,或实时作业进入运行中。
	// ResolveTaskRunHealthy closes the task's A1, A2 and B1: a batch job succeeded or a streaming job is running.
	ResolveTaskRunHealthy ResolveScope = "task_run_healthy"
	// ResolveJobTerminal 关闭这个作业的 C1:作业最终停下。
	// ResolveJobTerminal closes this job's C1: the job finally stopped.
	ResolveJobTerminal ResolveScope = "job_terminal"
	// ResolveClusterRule 关闭集群这条规则的阶段。
	// ResolveClusterRule closes the cluster's episode for this rule.
	ResolveClusterRule ResolveScope = "cluster_rule"
	// ResolveCDCDelay 关闭任务的 B4 阶段;JobID > 0 时只关闭最近作业为该作业的阶段。按 ResolveReason 决定是否发恢复通知。
	// ResolveCDCDelay closes the task's B4 episode; with JobID > 0 only an episode whose last job is that job. Whether
	// a recovery notification goes out depends on ResolveReason.
	ResolveCDCDelay ResolveScope = "cdc_delay"
)

type Signal

type Signal struct {
	Kind          SignalKind
	RuleKey       string
	ResolveScope  ResolveScope
	ClusterID     uint
	ClusterName   string
	TaskID        uint
	TaskName      string
	TaskCreatorID uint
	JobID         uint
	EngineJobID   string
	// TriggerSource 是作业的触发来源:manual、schedule、dolphin、cli 等。
	// TriggerSource is how the job was started: manual, schedule, dolphin, cli and so on.
	TriggerSource string
	RunType       string
	// Message 已脱敏,最长 MaxMessageLength 个字符。
	// Message is already redacted and at most MaxMessageLength characters.
	Message string
	// Since 只用于 Observe:这一段连续失败从何时开始。
	// Since is only used by Observe: when the current run of failures started.
	Since      time.Time
	OccurredAt time.Time
	// ResolveReason 只用于 Resolve:空或 recovered 发恢复通知,其余原因静默关闭。
	// ResolveReason is only used by Resolve: empty or recovered sends recovery, other reasons close silently.
	ResolveReason string
	// CDCDelay 只用于 B4 的 Fire、Refresh、Resolve。
	// CDCDelay is only used by B4 Fire, Refresh and Resolve.
	CDCDelay *CDCDelayDetail
	// StateChange 只用于 B4 Refresh:这条 Refresh 进入"无法确认"或停止提醒(high ↔ 回落待确认的切换不算),接收方不按
	// Refresh 的排队上限丢弃它(缓冲区真满时仍可能丢)。
	// StateChange is only used by B4 Refresh: the Refresh enters "unconfirmable" or pauses reminders (switching between
	// high and recovering does not count), so the receiver does not drop it under the Refresh queue limit (it can
	// still be dropped when the buffer is completely full).
	StateChange bool
}

Signal 是 sync 发给告警模块的一条信号;所有字段都是快照,告警模块不回查 sync。 Signal is one signal sync sends to the alert module; every field is a snapshot and monitoring never calls back.

type SignalKind

type SignalKind string

SignalKind 是信号类型。 SignalKind is the kind of a signal.

const (
	// KindFire 表示发生一次(作业级)或状态成立(集群级 D2)。
	// KindFire means one occurrence (job level) or a condition that now holds (cluster level D2).
	KindFire SignalKind = "fire"
	// KindResolve 表示按 ResolveScope 解除。
	// KindResolve clears episodes according to ResolveScope.
	KindResolve SignalKind = "resolve"
	// KindObserve 表示 D1 的持续失败观测,带 Since;由 monitoring 按规则窗口判断是否成熟。
	// KindObserve is a D1 sustained-failure observation carrying Since; monitoring decides maturity by the rule window.
	KindObserve SignalKind = "observe"
	// KindRefresh 只更新告警中阶段的展示内容(消息、延迟明细、最近发生时间),不计次、不通知;没有对应阶段时忽略。
	// KindRefresh only refreshes a firing episode's display content (message, delay detail, last occurrence); it
	// neither counts nor notifies and is ignored when there is no matching episode.
	KindRefresh SignalKind = "refresh"
)

type TaskConfig

type TaskConfig struct {
	Configured      bool     `json:"configured"`
	PolicyID        uint     `json:"policy_id,omitempty"`
	Enabled         bool     `json:"enabled"`
	Rules           []string `json:"rules"`
	ChannelIDs      []uint   `json:"channel_ids"`
	ReceiverUserIDs []uint   `json:"receiver_user_ids"`
	NotifyTaskOwner bool     `json:"notify_task_owner"`
	CooldownMinutes int      `json:"cooldown_minutes"`
	SendRecovery    bool     `json:"send_recovery"`
	// CDCDelayThresholdSeconds 是 B4 阈值(秒),0 或不填按默认 60,范围 10~86400;不论是否勾选 B4 都会返回。
	// CDCDelayThresholdSeconds is the B4 threshold in seconds; 0 or omitted means the default 60, range 10..86400.
	// It is always returned, whether or not B4 is selected.
	CDCDelayThresholdSeconds int `json:"cdc_delay_threshold_seconds"`
	// CDCDelayDurationMinutes 是 B4 持续时间(分钟),0 或不填按默认 5,范围 1~60。
	// CDCDelayDurationMinutes is the B4 duration in minutes; 0 or omitted means the default 5, range 1..60.
	CDCDelayDurationMinutes int `json:"cdc_delay_duration_minutes"`
}

TaskConfig 是任务告警配置,也是 GET/PUT /api/v1/sync/tasks/:id/alert 的数据结构。 TaskConfig is a task's alert configuration and the payload of GET/PUT /api/v1/sync/tasks/:id/alert.

func (*TaskConfig) HasRule

func (c *TaskConfig) HasRule(rule string) bool

HasRule 判断配置是否勾选了规则。 HasRule reports whether the config selects the rule.

func (*TaskConfig) Normalize

func (c *TaskConfig) Normalize() *TaskConfig

Normalize 去重、排序规则与 ID,并补齐默认值;不做校验。 Normalize de-duplicates and sorts rules and IDs and fills defaults; it does not validate.

func (*TaskConfig) Validate

func (c *TaskConfig) Validate() error

Validate 校验规则和数值范围;渠道和用户是否存在由告警模块校验。 Validate checks rules and value ranges; whether channels and users exist is checked by the alert module.

type TaskPolicyStore

type TaskPolicyStore interface {
	GetTaskAlertConfig(ctx context.Context, taskID uint) (*TaskConfig, error)
	SaveTaskAlertConfig(ctx context.Context, task TaskRef, cfg *TaskConfig) (*TaskConfig, error)
	DeleteTaskAlertConfig(ctx context.Context, taskID uint) error
}

TaskPolicyStore 读写以任务为范围的作业告警策略,由告警模块实现。 TaskPolicyStore reads and writes task-scoped job alert policies; the alert module implements it.

type TaskRef

type TaskRef struct {
	ID        uint
	Name      string
	ClusterID uint
	CreatorID uint
}

TaskRef 是告警模块保存任务策略时需要的任务快照。 TaskRef is the task snapshot the alert module needs to store a task policy.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL