forked from beego/beego
-
Notifications
You must be signed in to change notification settings - Fork 0
/
alils.go
206 lines (170 loc) · 3.9 KB
/
alils.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
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
package alils
import (
"encoding/json"
"fmt"
"strings"
"sync"
"github.com/gogo/protobuf/proto"
"github.com/pkg/errors"
"github.com/tianmaotalk/beego/v2/core/logs"
)
const (
// CacheSize sets the flush size
CacheSize int = 64
// Delimiter defines the topic delimiter
Delimiter string = "##"
)
// Config is the Config for Ali Log
type Config struct {
Project string `json:"project"`
Endpoint string `json:"endpoint"`
KeyID string `json:"key_id"`
KeySecret string `json:"key_secret"`
LogStore string `json:"log_store"`
Topics []string `json:"topics"`
Source string `json:"source"`
Level int `json:"level"`
FlushWhen int `json:"flush_when"`
Formatter string `json:"formatter"`
}
// aliLSWriter implements LoggerInterface.
// Writes messages in keep-live tcp connection.
type aliLSWriter struct {
store *LogStore
group []*LogGroup
withMap bool
groupMap map[string]*LogGroup
lock *sync.Mutex
Config
formatter logs.LogFormatter
}
// NewAliLS creates a new Logger
func NewAliLS() logs.Logger {
alils := new(aliLSWriter)
alils.Level = logs.LevelTrace
alils.formatter = alils
return alils
}
// Init parses config and initializes struct
func (c *aliLSWriter) Init(config string) error {
err := json.Unmarshal([]byte(config), c)
if err != nil {
return err
}
if c.FlushWhen > CacheSize {
c.FlushWhen = CacheSize
}
prj := &LogProject{
Name: c.Project,
Endpoint: c.Endpoint,
AccessKeyID: c.KeyID,
AccessKeySecret: c.KeySecret,
}
store, err := prj.GetLogStore(c.LogStore)
if err != nil {
return err
}
c.store = store
// Create default Log Group
c.group = append(c.group, &LogGroup{
Topic: proto.String(""),
Source: proto.String(c.Source),
Logs: make([]*Log, 0, c.FlushWhen),
})
// Create other Log Group
c.groupMap = make(map[string]*LogGroup)
for _, topic := range c.Topics {
lg := &LogGroup{
Topic: proto.String(topic),
Source: proto.String(c.Source),
Logs: make([]*Log, 0, c.FlushWhen),
}
c.group = append(c.group, lg)
c.groupMap[topic] = lg
}
if len(c.group) == 1 {
c.withMap = false
} else {
c.withMap = true
}
c.lock = &sync.Mutex{}
if len(c.Formatter) > 0 {
fmtr, ok := logs.GetFormatter(c.Formatter)
if !ok {
return errors.New(fmt.Sprintf("the formatter with name: %s not found", c.Formatter))
}
c.formatter = fmtr
}
return nil
}
func (c *aliLSWriter) Format(lm *logs.LogMsg) string {
return lm.OldStyleFormat()
}
func (c *aliLSWriter) SetFormatter(f logs.LogFormatter) {
c.formatter = f
}
// WriteMsg writes a message in connection.
// If connection is down, try to re-connect.
func (c *aliLSWriter) WriteMsg(lm *logs.LogMsg) error {
if lm.Level > c.Level {
return nil
}
var topic string
var content string
var lg *LogGroup
if c.withMap {
// Topic,LogGroup
strs := strings.SplitN(lm.Msg, Delimiter, 2)
if len(strs) == 2 {
pos := strings.LastIndex(strs[0], " ")
topic = strs[0][pos+1 : len(strs[0])]
lg = c.groupMap[topic]
}
// send to empty Topic
if lg == nil {
lg = c.group[0]
}
} else {
lg = c.group[0]
}
content = c.formatter.Format(lm)
c1 := &LogContent{
Key: proto.String("msg"),
Value: proto.String(content),
}
l := &Log{
Time: proto.Uint32(uint32(lm.When.Unix())),
Contents: []*LogContent{
c1,
},
}
c.lock.Lock()
lg.Logs = append(lg.Logs, l)
c.lock.Unlock()
if len(lg.Logs) >= c.FlushWhen {
c.flush(lg)
}
return nil
}
// Flush implementing method. empty.
func (c *aliLSWriter) Flush() {
// flush all group
for _, lg := range c.group {
c.flush(lg)
}
}
// Destroy destroy connection writer and close tcp listener.
func (c *aliLSWriter) Destroy() {
}
func (c *aliLSWriter) flush(lg *LogGroup) {
c.lock.Lock()
defer c.lock.Unlock()
err := c.store.PutLogs(lg)
if err != nil {
return
}
lg.Logs = make([]*Log, 0, c.FlushWhen)
}
func init() {
logs.Register(logs.AdapterAliLS, NewAliLS)
}