mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-08-20 10:52:25 +00:00
Buffer logs to send them in batches to the splunk server; use nanotime to tag logs more closely; use Stringer to format context values
This commit is contained in:
parent
4d6620f22e
commit
535cf48805
2 changed files with 58 additions and 13 deletions
|
|
@ -159,7 +159,7 @@ func Setup(ctx *cli.Context) error {
|
||||||
SkipTLSVerify: ctx.GlobalBool(splunkskiptlsverifyFlag.Name),
|
SkipTLSVerify: ctx.GlobalBool(splunkskiptlsverifyFlag.Name),
|
||||||
OriginalHandler: originalHandler,
|
OriginalHandler: originalHandler,
|
||||||
}
|
}
|
||||||
glogger.SetHandler(log.BufferedHandler(100, splunkHandler))
|
glogger.SetHandler(splunkHandler)
|
||||||
}
|
}
|
||||||
log.PrintOrigins(ctx.GlobalBool(debugFlag.Name))
|
log.PrintOrigins(ctx.GlobalBool(debugFlag.Name))
|
||||||
glogger.Verbosity(log.Lvl(ctx.GlobalInt(verbosityFlag.Name)))
|
glogger.Verbosity(log.Lvl(ctx.GlobalInt(verbosityFlag.Name)))
|
||||||
|
|
|
||||||
|
|
@ -30,6 +30,8 @@ type Handler struct {
|
||||||
OriginalHandler log.Handler
|
OriginalHandler log.Handler
|
||||||
errors chan error
|
errors chan error
|
||||||
once sync.Once
|
once sync.Once
|
||||||
|
events chan event
|
||||||
|
sendQueue chan []event
|
||||||
}
|
}
|
||||||
|
|
||||||
func (h *Handler) init() {
|
func (h *Handler) init() {
|
||||||
|
|
@ -57,6 +59,36 @@ func (h *Handler) init() {
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
}
|
}
|
||||||
|
h.sendQueue = make(chan []event)
|
||||||
|
h.events = make(chan event)
|
||||||
|
go func() {
|
||||||
|
eventsToSend := make([]event, 0)
|
||||||
|
ticker := time.NewTicker(1 * time.Second)
|
||||||
|
for {
|
||||||
|
select {
|
||||||
|
case e := <-h.events:
|
||||||
|
eventsToSend = append(eventsToSend, e)
|
||||||
|
if len(eventsToSend) > 50 {
|
||||||
|
oldEvents := eventsToSend
|
||||||
|
eventsToSend = make([]event, 0)
|
||||||
|
h.sendQueue <- oldEvents
|
||||||
|
}
|
||||||
|
break
|
||||||
|
case <-ticker.C:
|
||||||
|
if len(eventsToSend) > 0 {
|
||||||
|
oldEvents := eventsToSend
|
||||||
|
eventsToSend = make([]event, 0)
|
||||||
|
h.sendQueue <- oldEvents
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
go func() {
|
||||||
|
for {
|
||||||
|
events := <-h.sendQueue
|
||||||
|
h.logEvents(events)
|
||||||
|
}
|
||||||
|
}()
|
||||||
}
|
}
|
||||||
|
|
||||||
type eventdata struct {
|
type eventdata struct {
|
||||||
|
|
@ -66,7 +98,7 @@ type eventdata struct {
|
||||||
}
|
}
|
||||||
|
|
||||||
type event struct {
|
type event struct {
|
||||||
Time int64 `json:"time"` // epoch time in seconds
|
Time int64 `json:"time"` // epoch time in nano-seconds
|
||||||
Host string `json:"host"` // hostname
|
Host string `json:"host"` // hostname
|
||||||
Source string `json:"source,omitempty"` // optional description of the source of the event; typically the app's name
|
Source string `json:"source,omitempty"` // optional description of the source of the event; typically the app's name
|
||||||
SourceType string `json:"sourcetype,omitempty"` // optional name of a Splunk parsing configuration; this is usually inferred by Splunk
|
SourceType string `json:"sourcetype,omitempty"` // optional name of a Splunk parsing configuration; this is usually inferred by Splunk
|
||||||
|
|
@ -76,34 +108,47 @@ type event struct {
|
||||||
|
|
||||||
func (h *Handler) Log(r *log.Record) error {
|
func (h *Handler) Log(r *log.Record) error {
|
||||||
h.once.Do(h.init)
|
h.once.Do(h.init)
|
||||||
|
|
||||||
context := make(map[string]string)
|
context := make(map[string]string)
|
||||||
for index := 0; index < len(r.Ctx); index++ {
|
for index := 0; index < len(r.Ctx); index++ {
|
||||||
key := fmt.Sprintf("%v", r.Ctx[index])
|
key := fmt.Sprintf("%v", r.Ctx[index])
|
||||||
index++
|
index++
|
||||||
if r.Ctx[index] != nil {
|
if r.Ctx[index] != nil {
|
||||||
value := fmt.Sprintf("%v", r.Ctx[index])
|
if stringer, ok := r.Ctx[index].(fmt.Stringer); ok {
|
||||||
context[key] = value
|
context[key] = stringer.String()
|
||||||
|
} else {
|
||||||
|
context[key] = fmt.Sprintf("%v", r.Ctx[index])
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
e := &event{
|
}
|
||||||
Time: r.Time.Unix(),
|
e := event{
|
||||||
|
Time: r.Time.UnixNano(),
|
||||||
Host: h.Hostname,
|
Host: h.Hostname,
|
||||||
Source: h.Source,
|
Source: h.Source,
|
||||||
SourceType: h.SourceType,
|
SourceType: h.SourceType,
|
||||||
Index: h.Index,
|
Index: h.Index,
|
||||||
Event: eventdata{Msg: r.Msg, Level: r.Lvl.String(), Ctxt: context},
|
Event: eventdata{Msg: r.Msg, Level: r.Lvl.String(), Ctxt: context},
|
||||||
}
|
}
|
||||||
|
h.events <- e
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (h *Handler) logEvents(events []event) {
|
||||||
|
buf := new(bytes.Buffer)
|
||||||
|
for _, e := range events {
|
||||||
|
|
||||||
b, err := json.Marshal(e)
|
b, err := json.Marshal(e)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
h.errors <- err
|
h.errors <- err
|
||||||
return err
|
} else {
|
||||||
|
buf.Write(b)
|
||||||
|
buf.WriteString("\r\n\r\n")
|
||||||
}
|
}
|
||||||
err = h.doRequest(bytes.NewBuffer(b))
|
}
|
||||||
|
|
||||||
|
err := h.doRequest(buf)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
h.errors <- err
|
h.errors <- err
|
||||||
}
|
}
|
||||||
return err
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (h *Handler) doRequest(b *bytes.Buffer) error {
|
func (h *Handler) doRequest(b *bytes.Buffer) error {
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue