-
Notifications
You must be signed in to change notification settings - Fork 0
/
handler.go
163 lines (141 loc) · 3.74 KB
/
handler.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
package pslog
import (
"bytes"
"errors"
"fmt"
"os"
"time"
"gitee.com/xuesongtao/ps-log/line"
)
type PsLogWriter interface {
WriteTo(bus *LogHandlerBus)
}
type Stdout struct{}
func (p *Stdout) WriteTo(bus *LogHandlerBus) {
os.Stdout.WriteString(bus.Msg)
}
// Target 目标内容
type Target struct {
no int // 自增编号
Content string // 目标内容
excludes Matcher
Excludes []string // 排除 msg
To []PsLogWriter // 一个目标内容, 多种处理方式
}
// Handler 处理的部分
type Handler struct {
LoopParse bool // 循环解析, 用于监听单文件日志, 说明: 这个采集的有可能不准确(在这种是基于文件大小和内存记录的偏移量做比较, 模式建议用 tail, cron 的话如果间隔时间太长就可能漏)
CleanOffset bool // 是否需要清理保存的 offset, 只限于开机后一次
Tail bool // 是否实时处理, 说明: true 为实时; false 需要外部定时调用
Change int32 // 文件 offset 变化次数, 为持久化文件偏移量数阈值, 当, 说明: -1 为实时保存; 0 达到默认值 defaultHandleChange 时保存; 其他 大于后会保存
ExpireDur time.Duration // 文件句柄过期间隔, 常用于全局配置
ExpireAt time.Time // 文件句柄过期时间, 优先 ExpireDur 如: 2022-12-03 11:11:10
MergeRule line.Merger // 日志文件行合并规则, 默认 单行处理
targets Matcher
Targets []*Target // 目标 msg
Ext string // 外部存入, 回调返回
}
// initMatcher 初始化匹配
// arrLen 为匹配的数组长度
func (h *Handler) initMatcher(arrLen int) Matcher {
if arrLen <= 1 {
return &Simple{}
}
return newTire()
}
func (h *Handler) Valid() error {
if h.ExpireAt.IsZero() && h.ExpireDur == 0 {
return errors.New("ExpireAt, ExpireDur can not both null")
}
if len(h.Targets) == 0 {
return errors.New("Targets is required")
}
for i, target := range h.Targets {
if target.Content == "" {
return fmt.Errorf("Targets.Content[%d] is null", i)
}
if target.To == nil {
return fmt.Errorf("%q[%d] To is null", target.Content, i)
}
}
return nil
}
func (h *Handler) init() {
if h.Change == 0 {
h.Change = defaultHandleChange
}
if h.ExpireAt.IsZero() {
h.ExpireAt = time.Now().Add(h.ExpireDur)
}
if h.MergeRule == nil {
h.MergeRule = line.NewSing()
}
// 预处理 targets, exclude
h.targets = h.initMatcher(len(h.Targets))
no := 1
for _, target := range h.Targets {
if target.Content == "" {
continue
}
target.no = no
h.targets.Insert([]byte(target.Content), target)
target.excludes = h.initMatcher(len(target.Excludes))
for _, exclude := range target.Excludes {
if exclude == "" {
continue
}
target.excludes.Insert([]byte(exclude), nil)
}
no++
}
}
func (h *Handler) getTargetDump() string {
data := ""
for _, v := range h.Targets {
data += "【" + v.Content + "】"
}
return data
}
func (h *Handler) getExcludesDump() string {
data := ""
for _, t := range h.Targets {
es := ""
for _, e := range t.Excludes {
if es == "" {
es += e
} else {
es += ";" + e
}
}
if es == "" {
continue
}
data += "【" + t.Content + "排除" + es + "】"
}
return data
}
// logHandler 解析到的内容
type LogHandlerBus struct {
LogPath string // log 的路径
Msg string // buf 中的 string
Ext string // Handler 中的 Ext 值
buf *bytes.Buffer
tos []PsLogWriter
}
func (l *LogHandlerBus) skip() bool {
if l.buf.Len() == 0 {
return true
}
l.Msg = l.buf.String()
l.buf.Reset()
return l.Msg == ""
}
func (l *LogHandlerBus) Write(b []byte) {
l.buf.WriteString(string(b) + "\n")
}
func (l *LogHandlerBus) Reset() {
l.LogPath = ""
l.Msg = ""
l.buf.Reset()
l.tos = nil
}