event

package

v0.2.0 Latest Latest Go to latest Published: Aug 26, 2026 License: MIT Imports: 8 Imported by: 7

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

github.com/EthanScriptOn/FlowLink

Links

Open Source Insights

Documentation ¶

Overview ¶

Package event 定义 workflow 运行过程中产生的事件模型和事件下沉接口。

runtime 使用这些事件描述工作流、节点、日志和统计信息；存储到哪里由外部 sink 决定。

Index ¶

type AsyncSink
- func NewAsyncSink(sink Sink, opts AsyncSinkOptions) *AsyncSink
- func (s *AsyncSink) Close(ctx context.Context) error
- func (s *AsyncSink) Dropped() int64
- func (s *AsyncSink) Emit(e *Event)
type AsyncSinkOptions
type BatchSink
type Event
- func CloneEvent(e *Event) *Event
type MemorySink
- func NewMemorySink() *MemorySink
- func (s *MemorySink) Emit(e *Event)
- func (s *MemorySink) Events() []*Event
type MultiSink
- func (ms MultiSink) Emit(e *Event)
type NopSink
- func (NopSink) Emit(_ *Event)
- func (NopSink) EmitBatch(_ []*Event)
type Sanitizer
- func NewSanitizer(opts SanitizerOptions) *Sanitizer
- func (s *Sanitizer) SanitizeEvent(e *Event) *Event
type SanitizerOptions
type SanitizingSink
- func (s *SanitizingSink) Emit(e *Event)
- func (s *SanitizingSink) EmitBatch(events []*Event)
type Sink
- func NewSanitizingSink(sink Sink, opts SanitizerOptions) Sink
type Type

Constants ¶

This section is empty.

Variables ¶

This section is empty.

Functions ¶

This section is empty.

Types ¶

type AsyncSink ¶

type AsyncSink struct {
	// contains filtered or unexported fields
}

AsyncSink 用后台 worker 异步转发事件，降低 runtime 写事件时的阻塞风险。

func NewAsyncSink ¶

func NewAsyncSink(sink Sink, opts AsyncSinkOptions) *AsyncSink

NewAsyncSink 创建异步 sink，并按需补齐安全默认值。

func (*AsyncSink) Close ¶

func (s *AsyncSink) Close(ctx context.Context) error

Close 停止接收新事件并等待后台 worker 尽量刷新已有事件。

func (*AsyncSink) Dropped ¶

func (s *AsyncSink) Dropped() int64

Dropped 返回因为队列满而丢弃的事件数量。

func (*AsyncSink) Emit ¶

func (s *AsyncSink) Emit(e *Event)

Emit 复制事件后放入异步队列；DropOnFull 开启时队列满会计数丢弃。

type AsyncSinkOptions ¶

type AsyncSinkOptions struct {
	BufferSize    int
	Workers       int
	BatchSize     int
	FlushInterval time.Duration
	DropOnFull    bool
}

AsyncSinkOptions 配置异步事件 sink 的缓冲、并发和批量刷新行为。

type BatchSink ¶

type BatchSink interface {
	EmitBatch(events []*Event)
}

BatchSink 是 sink 可选实现的批量导出接口。

type Event ¶

type Event struct {
	Type         Type                   `json:"type"`
	Env          string                 `json:"env"`
	Workspace    string                 `json:"workspace"`
	TraceID      string                 `json:"traceId"`
	RunID        string                 `json:"runId"`
	ParentRunID  string                 `json:"parentRunId"`
	WorkflowID   string                 `json:"workflowId"`
	WorkflowVer  string                 `json:"workflowVersion"`
	NodeID       string                 `json:"nodeId"`
	NodeName     string                 `json:"nodeName"`
	ParentNodeID string                 `json:"parentNodeId"`
	NodePath     string                 `json:"nodePath"`
	NodeDepth    *int                   `json:"nodeDepth"`
	NodeIndex    *int                   `json:"nodeIndex"`
	NodeType     string                 `json:"nodeType"`
	Status       string                 `json:"status"`
	NodeStatus   string                 `json:"nodeStatus"`
	Error        string                 `json:"error"`
	LogLevel     string                 `json:"logLevel"`
	Message      string                 `json:"message"`
	StartedAt    time.Time              `json:"startedAt"`
	EndedAt      time.Time              `json:"endedAt"`
	DurationMs   int64                  `json:"durationMs"`
	Attrs        map[string]interface{} `json:"attrs"`
	Fields       map[string]interface{} `json:"fields"`
}

Event 是 FlowLink runtime 执行过程和节点日志的统一事件记录。

func CloneEvent ¶

func CloneEvent(e *Event) *Event

CloneEvent 深拷贝事件中的可变 map 字段。

type MemorySink ¶

type MemorySink struct {
	// contains filtered or unexported fields
}

MemorySink 将事件保存在内存中，主要用于测试、debug 和一次性报告。

func NewMemorySink ¶

func NewMemorySink() *MemorySink

NewMemorySink 创建空的内存事件 sink。

func (*MemorySink) Emit ¶

func (s *MemorySink) Emit(e *Event)

Emit 保存事件副本。

func (*MemorySink) Events ¶

func (s *MemorySink) Events() []*Event

Events 返回当前已收集事件的副本。

type MultiSink ¶

type MultiSink []Sink

MultiSink 将事件广播给多个 sink。

func (MultiSink) Emit ¶

func (ms MultiSink) Emit(e *Event)

Emit 依次向每个非 nil sink 发送事件。

type NopSink ¶

type NopSink struct{}

NopSink 丢弃所有事件，适合作为未配置 sink 时的安全默认值。

func (NopSink) Emit ¶

func (NopSink) Emit(_ *Event)

Emit 丢弃单条事件。

func (NopSink) EmitBatch ¶

func (NopSink) EmitBatch(_ []*Event)

EmitBatch 丢弃批量事件。

type Sanitizer ¶

type Sanitizer struct {
	// contains filtered or unexported fields
}

Sanitizer 对事件中的错误、消息、attrs 和 fields 做安全清洗。

func NewSanitizer ¶

func NewSanitizer(opts SanitizerOptions) *Sanitizer

NewSanitizer 创建事件清洗器，并补齐默认限制和默认敏感 key。

func (*Sanitizer) SanitizeEvent ¶

func (s *Sanitizer) SanitizeEvent(e *Event) *Event

SanitizeEvent 返回清洗后的事件副本。

type SanitizerOptions ¶

type SanitizerOptions struct {
	Enabled            bool
	MaxStringLength    int
	MaxCollectionItems int
	MaxDepth           int
	MaskedKeys         []string
}

SanitizerOptions 配置事件脱敏、截断和嵌套深度限制。

type SanitizingSink ¶

type SanitizingSink struct {
	// contains filtered or unexported fields
}

SanitizingSink 在写入下游 sink 前先清洗事件内容。

func (*SanitizingSink) Emit ¶

func (s *SanitizingSink) Emit(e *Event)

Emit 清洗并转发单条事件。

func (*SanitizingSink) EmitBatch ¶

func (s *SanitizingSink) EmitBatch(events []*Event)

EmitBatch 清洗并转发批量事件。

type Sink ¶

type Sink interface {
	Emit(e *Event)
}

Sink 是 runtime 和 trace helper 使用的可插拔事件导出接口。

func NewSanitizingSink ¶

func NewSanitizingSink(sink Sink, opts SanitizerOptions) Sink

NewSanitizingSink 包装下游 sink，使事件写入前经过 Sanitizer。

type Type ¶

type Type string

Type 是 runtime 产生的标准执行事件类型。

const (
	// WorkflowStarted 表示 workflow 开始执行。
	WorkflowStarted Type = "WORKFLOW_STARTED"
	// WorkflowFinished 表示 workflow 完成执行。
	WorkflowFinished Type = "WORKFLOW_FINISHED"
	// NodeStarted 表示节点开始执行。
	NodeStarted Type = "NODE_STARTED"
	// NodeFinished 表示节点成功完成执行。
	NodeFinished Type = "NODE_FINISHED"
	// NodeFailed 表示节点执行失败。
	NodeFailed Type = "NODE_FAILED"
	// WorkflowLog 表示框架在 workflow 维度写出的运行日志。
	WorkflowLog Type = "WORKFLOW_LOG"
	// NodeLog 表示节点通过 FlowContext 写出的日志。
	NodeLog Type = "NODE_LOG"
)

Source Files ¶

View all Source files

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