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
|
// Copyright (c) Tailscale Inc & contributors
// SPDX-License-Identifier: BSD-3-Clause
//go:build !ts_omit_logtail
package logtail
import (
"bytes"
"errors"
"expvar"
"fmt"
"tailscale.com/metrics"
"tailscale.com/syncs"
)
type Buffer interface {
// TryReadLine tries to read a log line from the ring buffer.
// If no line is available it returns a nil slice.
// If the ring buffer is closed it returns io.EOF.
//
// The returned slice may point to data that will be overwritten
// by a subsequent call to TryReadLine.
TryReadLine() ([]byte, error)
// Write writes a log line into the ring buffer.
// Implementations must not retain the provided buffer.
Write([]byte) (int, error)
}
func NewMemoryBuffer(numEntries int) Buffer {
return &memBuffer{
pending: make(chan qentry, numEntries),
}
}
type memBuffer struct {
next []byte
pending chan qentry
dropMu syncs.Mutex
dropCount int
// Metrics (see [memBuffer.ExpVar] for details).
writeCalls expvar.Int
readCalls expvar.Int
writeBytes expvar.Int
readBytes expvar.Int
droppedBytes expvar.Int
storedBytes expvar.Int
}
// ExpVar returns a [metrics.Set] with metrics about the buffer.
//
// - counter_write_calls: Total number of write calls.
// - counter_read_calls: Total number of read calls.
// - counter_write_bytes: Total number of bytes written.
// - counter_read_bytes: Total number of bytes read.
// - counter_dropped_bytes: Total number of bytes dropped.
// - gauge_stored_bytes: Current number of bytes stored in memory.
func (b *memBuffer) ExpVar() expvar.Var {
m := new(metrics.Set)
m.Set("counter_write_calls", &b.writeCalls)
m.Set("counter_read_calls", &b.readCalls)
m.Set("counter_write_bytes", &b.writeBytes)
m.Set("counter_read_bytes", &b.readBytes)
m.Set("counter_dropped_bytes", &b.droppedBytes)
m.Set("gauge_stored_bytes", &b.storedBytes)
return m
}
func (m *memBuffer) TryReadLine() ([]byte, error) {
m.readCalls.Add(1)
if m.next != nil {
msg := m.next
m.next = nil
m.readBytes.Add(int64(len(msg)))
m.storedBytes.Add(-int64(len(msg)))
return msg, nil
}
select {
case ent := <-m.pending:
if ent.dropCount > 0 {
m.next = ent.msg
b := fmt.Appendf(nil, "----------- %d logs dropped ----------", ent.dropCount)
m.writeBytes.Add(int64(len(b))) // indicate pseudo-injected log message
m.readBytes.Add(int64(len(b)))
return b, nil
}
m.readBytes.Add(int64(len(ent.msg)))
m.storedBytes.Add(-int64(len(ent.msg)))
return ent.msg, nil
default:
return nil, nil
}
}
func (m *memBuffer) Write(b []byte) (int, error) {
m.writeCalls.Add(1)
m.dropMu.Lock()
defer m.dropMu.Unlock()
ent := qentry{
msg: bytes.Clone(b),
dropCount: m.dropCount,
}
select {
case m.pending <- ent:
m.writeBytes.Add(int64(len(b)))
m.storedBytes.Add(+int64(len(b)))
m.dropCount = 0
return len(b), nil
default:
m.dropCount++
m.droppedBytes.Add(int64(len(b)))
return 0, errBufferFull
}
}
type qentry struct {
msg []byte
dropCount int
}
var errBufferFull = errors.New("logtail: buffer full")
|
