logsink

package
v2.0.11 Latest Latest
Warning

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

 Go to latest Published: Oct 15, 2026 License: MIT Imports: 6 Imported by: 0

Documentation

Overview

Package logsink 提供 MGIN 接口访问日志(PostLog)和运行日志(Record)的异步分发能力。

设计目标：

  1. 中间件 middleware/postlog 只负责产生日志并调用 Emit，不关心日志落到哪里；
  2. 落库、发 Kafka、写 Elasticsearch 等动作由注册的 Handler 完成；
  3. 外部插件（mgkafka、elasticsearch 插件等）只需实现 Handler 并在启动时注册， 即可接收全量接口日志，框架无需为每种目标硬编码分支。

处理模型：每个 Handler 拥有独立的缓冲队列与消费协程，慢 Handler 互不影响； 投递是非阻塞的，队列满时丢弃并计数，绝不会拖慢接口响应。

Index

Constants

View Source 
const (
	// DefaultQueueSize 每个 Handler 默认的缓冲队列长度。
	DefaultQueueSize = 1024
	// DefaultWorkers 每个 Handler 默认的消费协程数，1 保证日志按到达顺序处理。
	DefaultWorkers = 1
	// DefaultShutdownTimeout 关闭时等待队列排空的默认超时。
	DefaultShutdownTimeout = 3 * time.Second
)

Variables

View Source 
var ErrDuplicateHandler = errors.New("logsink: 同名 handler 已注册")

ErrDuplicateHandler 同名 Handler 重复注册。

View Source 
var ErrInvalidHandler = errors.New("logsink: handler 不能为空且 Name 不能为空")

ErrInvalidHandler Handler 为空或 Name 为空。

Functions

func Close 

func Close(timeout time.Duration)

Close 停止全部 Handler：停止接收新日志、尽力排空队列、调用 Closer。 timeout 为等待每个 Handler 排空的时间预算，超时则放弃剩余日志。

func Emit 

func Emit(entry *Entry) bool

Emit 异步投递一条日志到所有 Handler。 非阻塞：队列满时该条日志对该 Handler 丢弃并计数。返回是否至少投递成功一个 Handler。

func EmitRecord added in v2.0.11 

func EmitRecord(entry *Record) bool

EmitRecord 异步投递一条日志到所有 Handler。 非阻塞：队列满时该条日志对该 Handler 丢弃并计数。返回是否至少投递成功一个 Handler。

func EmitRecordTo added in v2.0.11 

func EmitRecordTo(name string, entry *Record) bool

EmitRecordTo sends a record only to the named application log handler. The record and its Data must remain immutable after dispatch.

func HandlerNames 

func HandlerNames() []string

HandlerNames 返回已注册 Handler 的名称，按注册顺序排列。

func HasHandlers 

func HasHandlers() bool

HasHandlers 是否已注册任何 Handler。用于快速跳过日志投递。

func HasRecordHandlers added in v2.0.11 

func HasRecordHandlers() bool

HasRecordHandlers 是否已注册任何 Handler。用于快速跳过日志投递。

func MustRegister 

func MustRegister(h Handler, opts ...Option)

MustRegister 同 Register，注册失败时 panic。适用于启动阶段装配。

func MustRegisterRecord added in v2.0.11 

func MustRegisterRecord(h RecordHandler, opts ...Option)

MustRegisterRecord 同 RegisterRecord，注册失败时 panic。适用于启动阶段装配。

func RecordHandlerNames added in v2.0.11 

func RecordHandlerNames() []string

RecordHandlerNames 返回已注册 Handler 的名称，按注册顺序排列。

func Register 

func Register(h Handler, opts ...Option) error

Register 注册一个日志处理器并立即启动其消费协程。 同名 Handler 重复注册返回 ErrDuplicateHandler，原 Handler 保持不变。

func RegisterRecord added in v2.0.11 

func RegisterRecord(h RecordHandler, opts ...Option) error

RegisterRecord 注册一个日志处理器并立即启动其消费协程。 同名 Handler 重复注册返回 ErrDuplicateHandler，原 Handler 保持不变。

func Reset 

func Reset()

Reset 清空全部 Handler（不等待排空、不调用 Closer），仅用于测试。

func SetDropLogger 

func SetDropLogger(fn func(handler string, dropped uint64))

SetDropLogger 设置日志丢弃时的回调，参数为 handler 名称与该 handler 累计丢弃条数。 回调已做 60 秒限流。默认不输出任何内容，由 MGIN 内部接管为 logs 包输出。

func SetErrorLogger 

func SetErrorLogger(fn func(handler string, err error))

SetErrorLogger 设置 Handler 写失败或 panic 时的回调。 回调已做 60 秒限流。默认不输出任何内容。

func Unregister 

func Unregister(name string, timeout time.Duration) bool

Unregister 注销并停止指定 Handler，等待其队列排空（受 timeout 限制）， 若 Handler 实现了 Closer 则调用 Close。

func UnregisterRecord added in v2.0.11 

func UnregisterRecord(name string, timeout time.Duration) bool

UnregisterRecord 注销并停止指定 Handler，等待其队列排空（受 timeout 限制）， 若 Handler 实现了 Closer 则调用 Close。

Types

type Closer 

type Closer interface {
	Close() error
}

Closer 可选接口。实现了它的 Handler 在注销或全局关闭时会被调用一次。

type Entry 

type Entry = models.PostLog

Entry 一条接口访问日志。

type Handler 

type Handler interface {
	Name() string
	Write(entry *Entry) error
}

Handler 日志处理器。由内置实现或外部插件实现，注册后即可收到全量接口日志。

Name 用于注册去重、日志标识以及运行期注销，必须唯一且非空。 Write 在独立协程中被调用，实现必须是并发安全的；返回 error 仅用于统计与告警。

type HandlerFunc 

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

HandlerFunc 把普通函数适配成 Handler，方便外部插件一行挂接。 

sink := logsink.NewHandlerFunc("kafka", func(e *logsink.Entry) error {
    return kafka.Send(topic, e)
}).WithClose(kafka.Close)
logsink.Register(sink)

func NewHandlerFunc 

func NewHandlerFunc(name string, write func(entry *Entry) error) *HandlerFunc

NewHandlerFunc 用名称与写函数构造 Handler。

func (*HandlerFunc) Close 

func (f *HandlerFunc) Close() error

Close 实现可选接口 Closer。

func (*HandlerFunc) Name 

func (f *HandlerFunc) Name() string

Name 返回 Handler 名称。

func (*HandlerFunc) WithClose 

func (f *HandlerFunc) WithClose(fn func() error) *HandlerFunc

WithClose 设置关闭回调，返回自身以便链式调用。

func (*HandlerFunc) Write 

func (f *HandlerFunc) Write(entry *Entry) error

Write 执行日志写入。

type HandlerStats 

type HandlerStats struct {
	Name    string
	Queued  int
	Dropped uint64
	Written uint64
	Failed  uint64
}

HandlerStats 单个 Handler 的运行状态。

func RecordStats added in v2.0.11 

func RecordStats() []HandlerStats

RecordStats 返回全部 Handler 的运行状态，按注册顺序排列。

func Stats 

func Stats() []HandlerStats

Stats 返回全部 Handler 的运行状态，按注册顺序排列。

type Option 

type Option func(*Options)

Option 注册时的可选配置。

func WithQueueSize 

func WithQueueSize(n int) Option

WithQueueSize 设置该 Handler 的缓冲队列长度。

func WithWorkers 

func WithWorkers(n int) Option

WithWorkers 设置该 Handler 的消费协程数。

type Options 

type Options struct {
	// QueueSize 缓冲队列长度，<=0 使用 DefaultQueueSize。
	QueueSize int
	// Workers 消费协程数，<=0 使用 DefaultWorkers。大于 1 时不保证处理顺序。
	Workers int
}

Options 注册参数。

type Record added in v2.0.11 

type Record struct {
	// Data is the destination-specific formatted payload (without a trailing newline).
	Data        []byte `json:"-"`
	Time        string `json:"time"`
	Level       string `json:"level"`
	Message     string `json:"message"`
	App         string `json:"app"`
	RequestID   string `json:"request_id"`
	GoroutineID uint64 `json:"goroutine_id"`
	Package     string `json:"package"`
	File        string `json:"file"`
	Function    string `json:"function"`
	Line        int    `json:"line"`
}

Record is an application log captured before asynchronous dispatch.

type RecordHandler added in v2.0.11 

type RecordHandler interface {
	Name() string
	Write(*Record) error
}

RecordHandler receives application logs, independently of PostLog handlers. Write must be concurrency safe and must not log back into the same sink.

type RecordHandlerFunc added in v2.0.11 

type RecordHandlerFunc struct {
	ID        string
	WriteFunc func(*Record) error
	// contains filtered or unexported fields
}

RecordHandlerFunc adapts a function to RecordHandler.

func NewRecordHandlerFunc added in v2.0.11 

func NewRecordHandlerFunc(name string, write func(*Record) error) *RecordHandlerFunc

NewRecordHandlerFunc adapts a write callback to an application log handler.

func (RecordHandlerFunc) Close added in v2.0.11 

func (h RecordHandlerFunc) Close() error

Close closes the destination after its queue drains.

func (RecordHandlerFunc) Name added in v2.0.11 

func (h RecordHandlerFunc) Name() string

Name returns the destination name.

func (*RecordHandlerFunc) WithClose added in v2.0.11 

func (h *RecordHandlerFunc) WithClose(closeFn func() error) *RecordHandlerFunc

WithClose attaches a shutdown callback.

func (RecordHandlerFunc) Write added in v2.0.11 

func (h RecordHandlerFunc) Write(r *Record) error

Write invokes the configured application log callback.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.