Something went wrong. Try again.
forked from tailscale/tailscale
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128// 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")