Something went wrong. Try again.
forked from tailscale/tailscale
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109// Copyright (c) Tailscale Inc & contributors// SPDX-License-Identifier: BSD-3-Clause
//go:build !ts_omit_logtail
// Package logtail sends logs to log.tailscale.com.package logtail
import ( "bytes" "cmp" "context" "crypto/rand" "encoding/binary" "expvar" "fmt" "io" "iter" "log" mrand "math/rand/v2" "net/http" "os" "runtime" "slices" "strconv" "sync" "sync/atomic" "time"
"github.com/creachadair/msync/trigger" jsonv2 "github.com/go-json-experiment/json" "github.com/go-json-experiment/json/jsontext" "tailscale.com/envknob" "tailscale.com/metrics" "tailscale.com/net/netmon" "tailscale.com/net/sockstats" "tailscale.com/tstime" tslogger "tailscale.com/types/logger" "tailscale.com/types/logid" "tailscale.com/util/eventbus" "tailscale.com/util/set" "tailscale.com/util/truncate" "tailscale.com/util/zstdframe")
// maxSize is the maximum size that a single log entry can be.// It is also the maximum body size that may be uploaded at a time.const maxSize = 256 << 10
// maxTextSize is the maximum size for a text log message.// Note that JSON log messages can be as large as maxSize.const maxTextSize = 16 << 10
// bufferSize is the typical buffer size to retain.// It is large enough to handle most log messages,// but not too large to be a notable waste of memory if retained forever.const bufferSize = 4 << 10
// newLogger constructs a *Logger from cfg, applying defaults, but does not start// the background uploading goroutine. It is shared by [NewLogger] and the// stateless [UploadLogs].func newLogger(cfg Config) *Logger { if cfg.BaseURL == "" { cfg.BaseURL = "https://" + DefaultHost } if cfg.HTTPC == nil { cfg.HTTPC = http.DefaultClient } if cfg.Clock == nil { cfg.Clock = tstime.StdClock{} } if cfg.Stderr == nil { cfg.Stderr = os.Stderr } if cfg.Buffer == nil { cfg.Buffer = NewMemoryBuffer(256) } var procID uint32 if cfg.IncludeProcID { keyBytes := make([]byte, 4) rand.Read(keyBytes) procID = binary.LittleEndian.Uint32(keyBytes) if procID == 0 { // 0 is the empty/off value, assign a different (non-zero) value to // make sure we still include an ID (actual value does not matter). procID = 7 } } if s := envknob.String("TS_DEBUG_LOGTAIL_FLUSHDELAY"); s != "" { if delay, err := time.ParseDuration(s); err == nil { cfg.FlushDelayFn = func() time.Duration { return delay } } else { log.Fatalf("invalid TS_DEBUG_LOGTAIL_FLUSHDELAY: %v", err) } } else if cfg.FlushDelayFn == nil && envknob.Bool("IN_TS_TEST") { cfg.FlushDelayFn = func() time.Duration { return 0 } }
var urlSuffix string if !cfg.CopyPrivateID.IsZero() { urlSuffix = "?copyId=" + cfg.CopyPrivateID.String() } logger := &Logger{ privateID: cfg.PrivateID, stderr: cfg.Stderr, stderrLevel: int64(cfg.StderrLevel), httpc: cfg.HTTPC, url: cfg.BaseURL + "/c/" + cfg.Collection + "/" + cfg.PrivateID.String() + urlSuffix, buffer: cfg.Buffer, maxUploadSize: cfg.MaxUploadSize, skipClientTime: cfg.SkipClientTime, drainWake: make(chan struct{}, 1), sentinel: make(chan int32, 16), flushDelayFn: cfg.FlushDelayFn, clock: cfg.Clock, metricsDelta: cfg.MetricsDelta,
procID: procID, includeProcSequence: cfg.IncludeProcSequence,
shutdownStart: make(chan struct{}), shutdownDone: make(chan struct{}), }
if cfg.Bus != nil { logger.eventClient = cfg.Bus.Client("logtail.Logger") // Subscribe to change deltas from NetMon to detect when the network comes up. eventbus.SubscribeFunc(logger.eventClient, logger.onChangeDelta) } logger.SetSockstatsLabel(sockstats.LabelLogtailLogger) logger.compressLogs = cfg.CompressLogs logger.disabled.Store(cfg.Disabled)
return logger}
// NewLogger returns a new Logger that splits logs as configured between local// logging facilities and uploading to a log server in the background.func NewLogger(cfg Config, logf tslogger.Logf) *Logger { logger := newLogger(cfg)
ctx, cancel := context.WithCancel(context.Background()) logger.uploadCancel = cancel
go logger.uploading(ctx) if envknob.Bool("TS_DEBUG_LOGTAIL") { logger.Write([]byte("logtail started")) } return logger}
// Logtail is the reserved "logtail" metadata member of a [LogEntry].//// Zero-valued fields are omitted when an entry is uploaded. [UploadLogs] fills// in any fields the caller leaves zero from its [Config] (e.g. ClientTime and// ProcID), so most callers can leave this empty.type Logtail struct { // ClientTime is the time the entry was generated. If zero, [UploadLogs] // fills it with the current time unless Config.SkipClientTime is set. ClientTime time.Time `json:"client_time,omitzero"` // ProcID is an ephemeral process identifier; see Config.IncludeProcID. ProcID uint32 `json:"proc_id,omitzero"` // ProcSeq is an ephemeral per-process sequence number; see // Config.IncludeProcSequence. ProcSeq uint64 `json:"proc_seq,omitzero"`}
// LogEntry is a single log entry to be uploaded via [UploadLogs].//// It marshals to a JSON object whose reserved "logtail" member carries the// metadata in Logtail and whose remaining members are taken from Value, which// is inlined at the top level alongside "logtail".//// Value must marshal to a JSON object: T must be a Go struct (or pointer to// one), a Go map with a string key, or a [jsontext.Value] holding an object// (for example jsontext.Value(`{"text":"hello"}`)). This is enforced when the// entry is uploaded, not at compile time. Use T = [jsontext.Value] to mix// differently-shaped payloads in a single upload.type LogEntry[T any] struct { Logtail Logtail `json:"logtail,omitzero"` // The `inline` tag option was renamed to `embed` in Go 1.27's // encoding/json/v2, but the pinned go-json-experiment module // (used on older Go versions) only knows `inline`. Each // implementation ignores the option it doesn't know, so specify // both until we require Go 1.27 and drop `inline`. Value T `json:",inline,embed"`}
// UploadLogs uploads entries to the log server described by conf and returns// once they have all been sent (or an upload fails). It returns the number of// entries that were successfully uploaded.//// It is a stateless alternative to [NewLogger] for callers that just want to// push a batch of log entries without the background uploader, ring buffer,// stderr echoing, or network-up gating that a [Logger] provides. Each entry is// marshaled to JSON, its [Logtail] metadata is filled in from conf where the// caller left it zero (honoring conf.SkipClientTime, conf.IncludeProcID, and// conf.IncludeProcSequence), and entries are batched up to the server's maximum// upload size and POSTed synchronously (compressed when conf.CompressLogs is// set).//// Unlike [Logger], UploadLogs does not retry: if a batch fails to upload it// returns the count of entries sent in prior batches and the error immediately,// and any remaining entries are not sent. The conf.Stderr and conf.Bus fields// are ignored.func UploadLogs[T any](ctx context.Context, conf Config, entries iter.Seq[LogEntry[T]]) (int, error) { conf.Stderr = io.Discard // pure uploader: never echo to stderr conf.Bus = nil // no netmon/eventbus subscription for a one-shot lg := newLogger(conf)
maxLen := cmp.Or(lg.maxUploadSize, maxSize)
// body accumulates a JSON array of encoded entries: "[e1,e2,...]". // The framing mirrors Logger.drainPending. body := make([]byte, 0, bufferSize) // reused across batches body = append(body, '[')
// sent counts entries confirmed uploaded; pending counts entries // accumulated in body but not yet uploaded. A successful sendBatch moves // pending into sent. var sent, pending int sendBatch := func() error { if len(body) <= len("[") { return nil } out := bytes.TrimRight(body, ",") out = append(out, ']') origlen := -1 // sentinel value: uncompressed // Don't attempt to compress tiny bodies; not worth the CPU cycles. if lg.compressLogs && len(out) > 256 { zbody := zstdframe.AppendEncode(nil, out, zstdframe.FastestCompression, zstdframe.LowMemory(true)) // Only send it compressed if the bandwidth savings are sufficient. if len(out)-len(zbody) > 64 { origlen = len(out) out = zbody } } // upload is synchronous, so it is safe to reuse body's backing array // for the next batch once upload returns. if _, err := lg.upload(ctx, out, origlen); err != nil { return err } body = body[:len("[")] sent += pending pending = 0 return nil }
var procSeq uint64 for e := range entries { if err := ctx.Err(); err != nil { return sent, err }
// Fill in metadata from conf, preserving any caller-set values. if e.Logtail.ClientTime.IsZero() && !lg.skipClientTime { e.Logtail.ClientTime = lg.clock.Now().UTC() } if e.Logtail.ProcID == 0 { e.Logtail.ProcID = lg.procID } procSeq++ if lg.includeProcSequence { e.Logtail.ProcSeq = procSeq }
enc, err := jsonv2.Marshal(e) if err != nil { return sent, fmt.Errorf("logtail: encoding entry %d: %w", procSeq, err) }
// Flush the current batch before adding an entry that would overflow it. if len(body) > len("[") && len(body)+len(enc) > maxLen { if err := sendBatch(); err != nil { return sent, err } } body = append(body, enc...) body = append(body, ',') pending++ } err := sendBatch() return sent, err}
// Logger writes logs, splitting them as configured between local// logging facilities and uploading to a log server.type Logger struct { stderr io.Writer stderrLevel int64 // accessed atomically httpc *http.Client url string skipClientTime bool netMonitor *netmon.Monitor buffer Buffer maxUploadSize int drainWake chan struct{} // signal to speed up drain drainBuf []byte // owned by drainPending for reuse flushDelayFn func() time.Duration // negative or zero return value to upload aggressively, or >0 to batch at this delay flushPending atomic.Bool sentinel chan int32 clock tstime.Clock compressLogs bool uploadCancel func() explainedRaw bool metricsDelta func() string // or nil privateID logid.PrivateID httpDoCalls atomic.Int32 sockstatsLabel atomicSocktatsLabel eventClient *eventbus.Client networkIsUp trigger.Cond // set/reset by netmon.ChangeDelta events
procID uint32 includeProcSequence bool
// disabled, when true, causes this logger to drop incoming log entries // without buffering or uploading. It is independent of the process-wide // Disable kill switch, which takes precedence. Toggled by SetEnabled. disabled atomic.Bool
writeLock sync.Mutex // guards procSequence, flushTimer, buffer.Write calls procSequence uint64 flushTimer tstime.TimerController // used when flushDelay is >0 writeBuf [bufferSize]byte // owned by Write for reuse bytesBuf bytes.Buffer // owned by appendTextOrJSONLocked for reuse jsonDec jsontext.Decoder // owned by appendTextOrJSONLocked for reuse
shutdownStartMu sync.Mutex // guards the closing of shutdownStart shutdownStart chan struct{} // closed when shutdown begins shutdownDone chan struct{} // closed when shutdown complete
// Metrics (see [Logger.ExpVar] for details). uploadCalls expvar.Int failedCalls expvar.Int uploadedBytes expvar.Int uploadingTime expvar.Int}
type atomicSocktatsLabel struct{ p atomic.Uint32 }
func (p *atomicSocktatsLabel) Load() sockstats.Label { return sockstats.Label(p.p.Load()) }func (p *atomicSocktatsLabel) Store(label sockstats.Label) { p.p.Store(uint32(label)) }
// SetVerbosityLevel controls the verbosity level that should be// written to stderr. 0 is the default (not verbose). Levels 1 or higher// are increasingly verbose.func (lg *Logger) SetVerbosityLevel(level int) { atomic.StoreInt64(&lg.stderrLevel, int64(level))}
// SetNetMon sets the network monitor.//// It should not be changed concurrently with log writes and should// only be set once.func (lg *Logger) SetNetMon(lm *netmon.Monitor) { lg.netMonitor = lm}
// SetSockstatsLabel sets the label used in sockstat logs to identify network traffic from this logger.func (lg *Logger) SetSockstatsLabel(label sockstats.Label) { lg.sockstatsLabel.Store(label)}
// PrivateID returns the logger's private log ID.//// It exists for internal use only.func (lg *Logger) PrivateID() logid.PrivateID { return lg.privateID }
// Shutdown gracefully shuts down the logger while completing any// remaining uploads.//// It will block, continuing to try and upload unless the passed// context object interrupts it by being done.// If the shutdown is interrupted, an error is returned.func (lg *Logger) Shutdown(ctx context.Context) error { done := make(chan struct{}) go func() { select { case <-ctx.Done(): lg.uploadCancel() <-lg.shutdownDone case <-lg.shutdownDone: } close(done) lg.httpc.CloseIdleConnections() }()
if lg.eventClient != nil { lg.eventClient.Close() } lg.shutdownStartMu.Lock() select { case <-lg.shutdownStart: lg.shutdownStartMu.Unlock() return nil default: } close(lg.shutdownStart) lg.shutdownStartMu.Unlock()
io.WriteString(lg, "logger closing down\n") <-done
return nil}
// Close shuts down this logger object, the background log uploader// process, and any associated goroutines.//// Deprecated: use Shutdownfunc (lg *Logger) Close() { lg.Shutdown(context.Background())}
// drainBlock is called by drainPending when there are no logs to drain.//// In typical operation, every call to the Write method unblocks and triggers a// buffer.TryReadline, so logs are written with very low latency.//// If the caller specified FlushInterface, drainWake is only sent to// periodically.func (lg *Logger) drainBlock() (shuttingDown bool) { select { case <-lg.shutdownStart: return true case <-lg.drainWake: } return false}
// drainPending drains and encodes a batch of logs from the buffer for upload.// If no logs are available, drainPending blocks until logs are available.// The returned buffer is only valid until the next call to drainPending.func (lg *Logger) drainPending() (b []byte) { b = lg.drainBuf[:0] b = append(b, '[') defer func() { b = bytes.TrimRight(b, ",") b = append(b, ']') lg.drainBuf = b if len(b) <= len("[]") { b = nil } }()
maxLen := cmp.Or(lg.maxUploadSize, maxSize) for len(b) < maxLen { line, err := lg.buffer.TryReadLine() switch { case err == io.EOF: return b case err != nil: b = append(b, '{') b = lg.appendMetadata(b, false, true, 0, 0, "reading ringbuffer: "+err.Error(), nil, 0) b = bytes.TrimRight(b, ",") b = append(b, '}') return b case line == nil: // If we read at least some log entries, return immediately. if len(b) > len("[") { return b }
// We're about to block. If we're holding on to too much memory // in our buffer from a previous large write, let it go. if cap(b) > bufferSize { b = bytes.Clone(b) lg.drainBuf = b }
if shuttingDown := lg.drainBlock(); shuttingDown { return b } continue }
switch { case len(line) == 0: continue case line[0] == '{' && jsontext.Value(line).IsValid(): // This is already a valid JSON object, so just append it. // This may exceed maxLen, but should be no larger than maxSize // so long as logic writing into the buffer enforces the limit. b = append(b, line...) default: // This is probably a log added to stderr by filch // outside of the logtail logger. Encode it. if !lg.explainedRaw { fmt.Fprintf(lg.stderr, "RAW-STDERR: ***\n") fmt.Fprintf(lg.stderr, "RAW-STDERR: *** Lines prefixed with RAW-STDERR below bypassed logtail and probably come from a previous run of the program\n") fmt.Fprintf(lg.stderr, "RAW-STDERR: ***\n") fmt.Fprintf(lg.stderr, "RAW-STDERR:\n") lg.explainedRaw = true } fmt.Fprintf(lg.stderr, "RAW-STDERR: %s", b) // Do not add a client time, as it could be really old. // Do not include instance key or ID either, // since this came from a different instance. b = lg.appendText(b, line, true, 0, 0, 0) } b = append(b, ',') } return b}
// This is the goroutine that repeatedly uploads logs in the background.func (lg *Logger) uploading(ctx context.Context) { defer close(lg.shutdownDone)
for { body := lg.drainPending() origlen := -1 // sentinel value: uncompressed // Don't attempt to compress tiny bodies; not worth the CPU cycles. if lg.compressLogs && len(body) > 256 { zbody := zstdframe.AppendEncode(nil, body, zstdframe.FastestCompression, zstdframe.LowMemory(true))
// Only send it compressed if the bandwidth savings are sufficient. // Just the extra headers associated with enabling compression // are 50 bytes by themselves. if len(body)-len(zbody) > 64 { origlen = len(body) body = zbody } }
var lastError string var numFailures int var firstFailure time.Time for len(body) > 0 && ctx.Err() == nil { if logtailDisabled.Load() || lg.disabled.Load() { break } retryAfter, err := lg.upload(ctx, body, origlen) if err != nil { numFailures++ firstFailure = lg.clock.Now()
if !lg.internetUp() { fmt.Fprintf(lg.stderr, "logtail: internet down; waiting\n") lg.awaitInternetUp(ctx) continue }
// Only print the same message once. if currError := err.Error(); lastError != currError { fmt.Fprintf(lg.stderr, "logtail: upload: %v\n", err) lastError = currError }
// Sleep for the specified retryAfter period, // otherwise default to some random value. if retryAfter <= 0 { retryAfter = mrand.N(30*time.Second) + 30*time.Second } tstime.Sleep(ctx, min(retryAfter, 5*time.Minute)) // ignore absurdly large retryAfter values } else { // Only print a success message after recovery. if numFailures > 0 { fmt.Fprintf(lg.stderr, "logtail: upload succeeded after %d failures and %s\n", numFailures, lg.clock.Since(firstFailure).Round(time.Second)) } break } }
select { case <-lg.shutdownStart: return default: } }}
func (lg *Logger) internetUp() bool { select { case <-lg.networkIsUp.Ready(): return true default: if lg.netMonitor == nil { return true // No way to tell, so assume it is. } return lg.netMonitor.InterfaceState().AnyInterfaceUp() }}
// onChangeDelta is an eventbus subscriber function that handles// [netmon.ChangeDelta] events to detect whether the Internet is expected to be// reachable.func (lg *Logger) onChangeDelta(delta *netmon.ChangeDelta) { if delta.AnyInterfaceUp() { fmt.Fprintf(lg.stderr, "logtail: internet back up\n") lg.networkIsUp.Set() } else { fmt.Fprintf(lg.stderr, "logtail: network changed, but is not up\n") lg.networkIsUp.Reset() }}
func (lg *Logger) awaitInternetUp(ctx context.Context) { if lg.eventClient != nil { select { case <-lg.networkIsUp.Ready(): case <-ctx.Done(): } return } upc := make(chan bool, 1) defer lg.netMonitor.RegisterChangeCallback(func(delta *netmon.ChangeDelta) { if delta.AnyInterfaceUp() { select { case upc <- true: default: } } })() if lg.internetUp() { return } select { case <-upc: fmt.Fprintf(lg.stderr, "logtail: internet back up\n") case <-ctx.Done(): }}
// upload uploads body to the log server.// origlen indicates the pre-compression body length.// origlen of -1 indicates that the body is not compressed.func (lg *Logger) upload(ctx context.Context, body []byte, origlen int) (retryAfter time.Duration, err error) { lg.uploadCalls.Add(1) startUpload := time.Now()
const maxUploadTime = 45 * time.Second ctx = sockstats.WithSockStats(ctx, lg.sockstatsLabel.Load(), lg.Logf) ctx, cancel := context.WithTimeout(ctx, maxUploadTime) defer cancel()
req, err := http.NewRequestWithContext(ctx, "POST", lg.url, bytes.NewReader(body)) if err != nil { // I know of no conditions under which this could fail. // Report it very loudly. // TODO record logs to disk panic("logtail: cannot build http request: " + err.Error()) } if origlen != -1 { req.Header.Add("Content-Encoding", "zstd") req.Header.Add("Orig-Content-Length", strconv.Itoa(origlen)) } if runtime.GOOS == "js" { // We once advertised we'd accept optional client certs (for internal use) // on log.tailscale.com but then Tailscale SSH js/wasm clients prompted // users (on some browsers?) to pick a client cert. We'll fix the server's // TLS ServerHello, but we can also fix it client side for good measure. // // Corp details: https://github.com/tailscale/corp/issues/18177#issuecomment-2026598715 // and https://github.com/tailscale/corp/pull/18775#issuecomment-2027505036 // // See https://github.com/golang/go/wiki/WebAssembly#configuring-fetch-options-while-using-nethttp // and https://developer.mozilla.org/en-US/docs/Web/API/fetch#credentials req.Header.Set("js.fetch:credentials", "omit") } req.Header["User-Agent"] = nil // not worth writing one; save some bytes
compressedNote := "not-compressed" if origlen != -1 { compressedNote = "compressed" }
lg.httpDoCalls.Add(1) resp, err := lg.httpc.Do(req) if err != nil { lg.failedCalls.Add(1) return 0, fmt.Errorf("log upload of %d bytes %s failed: %v", len(body), compressedNote, err) } defer resp.Body.Close()
if resp.StatusCode != http.StatusOK { lg.failedCalls.Add(1) n, _ := strconv.Atoi(resp.Header.Get("Retry-After")) b, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<10)) return time.Duration(n) * time.Second, fmt.Errorf("log upload of %d bytes %s failed %d: %s", len(body), compressedNote, resp.StatusCode, bytes.TrimSpace(b)) }
lg.uploadedBytes.Add(int64(len(body))) lg.uploadingTime.Add(int64(time.Since(startUpload))) return 0, nil}
// Flush uploads all logs to the server. It blocks until complete or there is an// unrecoverable error.//// TODO(bradfitz): this apparently just returns nil, as of tailscale/corp@9c2ec35.// Finish cleaning this up.func (lg *Logger) Flush() error { return nil}
// StartFlush starts a log upload, if anything is pending.//// If l is nil, StartFlush is a no-op.func (lg *Logger) StartFlush() { if lg != nil { lg.tryDrainWake() }}
// ExpVar report metrics about the logger.//// - counter_upload_calls: Total number of upload attempts.//// - counter_upload_errors: Total number of upload attempts that failed.//// - counter_uploaded_bytes: Total number of bytes successfully uploaded// (which is calculated after compression is applied).//// - counter_uploading_nsecs: Total number of nanoseconds spent uploading.//// - buffer: An optional [metrics.Set] with metrics for the [Buffer].func (lg *Logger) ExpVar() expvar.Var { m := new(metrics.Set) m.Set("counter_upload_calls", &lg.uploadCalls) m.Set("counter_upload_errors", &lg.failedCalls) m.Set("counter_uploaded_bytes", &lg.uploadedBytes) m.Set("counter_uploading_nsecs", &lg.uploadingTime) if v, ok := lg.buffer.(interface{ ExpVar() expvar.Var }); ok { m.Set("buffer", v.ExpVar()) } return m}
// logtailDisabled is whether logtail uploads to logcatcher are disabled.var logtailDisabled atomic.Bool
// Disable disables logtail uploads for the lifetime of the process.func Disable() { logtailDisabled.Store(true)}
// SetEnabled enables or disables log uploading by lg. When disabled, log// entries passed to lg are dropped rather than buffered or uploaded; already// buffered entries may still drain. The process-wide [Disable] kill switch// takes precedence: if Disable has been called, SetEnabled(true) does not// re-enable uploads.func (lg *Logger) SetEnabled(enabled bool) { lg.disabled.Store(!enabled)}
var debugWakesAndUploads = envknob.RegisterBool("TS_DEBUG_LOGTAIL_WAKES")
// tryDrainWake tries to send to lg.drainWake, to cause an uploading wakeup.// It does not block.func (lg *Logger) tryDrainWake() { lg.flushPending.Store(false) if debugWakesAndUploads() { // Using println instead of log.Printf here to avoid recursing back into // ourselves. println("logtail: try drain wake, numHTTP:", lg.httpDoCalls.Load()) } select { case lg.drainWake <- struct{}{}: default: }}
func (lg *Logger) sendLocked(jsonBlob []byte) (int, error) { tapSend(jsonBlob) if logtailDisabled.Load() || lg.disabled.Load() { return len(jsonBlob), nil }
n, err := lg.buffer.Write(jsonBlob)
flushDelay := defaultFlushDelay if lg.flushDelayFn != nil { flushDelay = lg.flushDelayFn() } if flushDelay > 0 { if lg.flushPending.CompareAndSwap(false, true) { if lg.flushTimer == nil { lg.flushTimer = lg.clock.AfterFunc(flushDelay, lg.tryDrainWake) } else { lg.flushTimer.Reset(flushDelay) } } } else { lg.tryDrainWake() } return n, err}
// appendMetadata appends optional "logtail", "metrics", and "v" JSON members.// This assumes dst is already within a JSON object.// Each member is comma-terminated.func (lg *Logger) appendMetadata(dst []byte, skipClientTime, skipMetrics bool, procID uint32, procSequence uint64, errDetail string, errData jsontext.Value, level int) []byte { // Append optional logtail metadata. if !skipClientTime || procID != 0 || procSequence != 0 || errDetail != "" || errData != nil { dst = append(dst, `"logtail":{`...) if !skipClientTime { dst = append(dst, `"client_time":"`...) dst = lg.clock.Now().UTC().AppendFormat(dst, time.RFC3339Nano) dst = append(dst, '"', ',') } if procID != 0 { dst = append(dst, `"proc_id":`...) dst = strconv.AppendUint(dst, uint64(procID), 10) dst = append(dst, ',') } if procSequence != 0 { dst = append(dst, `"proc_seq":`...) dst = strconv.AppendUint(dst, procSequence, 10) dst = append(dst, ',') } if errDetail != "" || errData != nil { dst = append(dst, `"error":{`...) if errDetail != "" { dst = append(dst, `"detail":`...) dst, _ = jsontext.AppendQuote(dst, errDetail) dst = append(dst, ',') } if errData != nil { dst = append(dst, `"bad_data":`...) dst = append(dst, errData...) dst = append(dst, ',') } dst = bytes.TrimRight(dst, ",") dst = append(dst, '}', ',') } dst = bytes.TrimRight(dst, ",") dst = append(dst, '}', ',') }
// Append optional metrics metadata. if !skipMetrics && lg.metricsDelta != nil { if d := lg.metricsDelta(); d != "" { dst = append(dst, `"metrics":"`...) dst = append(dst, d...) dst = append(dst, '"', ',') } }
// Add the optional log level, if non-zero. // Note that we only use log levels 1 and 2 currently. // It's unlikely we'll ever make it past 9. if level > 0 && level < 10 { dst = append(dst, `"v":`...) dst = append(dst, '0'+byte(level)) dst = append(dst, ',') }
return dst}
// appendText appends a raw text message in the Tailscale JSON log entry format.func (lg *Logger) appendText(dst, src []byte, skipClientTime bool, procID uint32, procSequence uint64, level int) []byte { dst = slices.Grow(dst, len(src)) dst = append(dst, '{') dst = lg.appendMetadata(dst, skipClientTime, false, procID, procSequence, "", nil, level) if len(src) == 0 { dst = bytes.TrimRight(dst, ",") return append(dst, "}\n"...) }
// Append the text string, which may be truncated. // Invalid UTF-8 will be mangled with the Unicode replacement character. dst = append(dst, `"text":`...) dst = appendTruncatedString(dst, src, maxTextSize) return append(dst, "}\n"...)}
// appendTruncatedString appends a JSON string for src,// truncating the src to be no larger than n.func appendTruncatedString(dst, src []byte, n int) []byte { srcLen := len(src) src = truncate.String(src, n) dst, _ = jsontext.AppendQuote(dst, src) // ignore error; only occurs for invalid UTF-8 if srcLen > len(src) { dst = dst[:len(dst)-len(`"`)] // trim off preceding double-quote dst = append(dst, "…+"...) dst = strconv.AppendInt(dst, int64(srcLen-len(src)), 10) dst = append(dst, '"') // re-append succeeding double-quote } return dst}
// appendTextOrJSONLocked appends a raw text message or a raw JSON object// in the Tailscale JSON log format.func (lg *Logger) appendTextOrJSONLocked(dst, src []byte, level int) []byte { if lg.includeProcSequence { lg.procSequence++ } if len(src) == 0 || src[0] != '{' { return lg.appendText(dst, src, lg.skipClientTime, lg.procID, lg.procSequence, level) }
// Check whether the input is a valid JSON object and // whether it contains the reserved "logtail" name at the top-level. var logtailKeyOffset, logtailValOffset, logtailValLength int validJSON := func() bool { // The jsontext.NewDecoder API operates on an io.Reader, for which // bytes.Buffer provides a means to convert a []byte into an io.Reader. // However, bytes.NewBuffer normally allocates unless // we immediately shallow copy it into a pre-allocated Buffer struct. // See https://go.dev/issue/67004. lg.bytesBuf = *bytes.NewBuffer(src) defer func() { lg.bytesBuf = bytes.Buffer{} }() // avoid pinning src
dec := &lg.jsonDec dec.Reset(&lg.bytesBuf) if tok, err := dec.ReadToken(); tok.Kind() != '{' || err != nil { return false } for dec.PeekKind() != '}' { keyOffset := dec.InputOffset() tok, err := dec.ReadToken() if err != nil { return false } isLogtail := tok.String() == "logtail" valOffset := dec.InputOffset() if dec.SkipValue() != nil { return false } if isLogtail { logtailKeyOffset = int(keyOffset) logtailValOffset = int(valOffset) logtailValLength = int(dec.InputOffset()) - logtailValOffset } } if tok, err := dec.ReadToken(); tok.Kind() != '}' || err != nil { return false } if _, err := dec.ReadToken(); err != io.EOF { return false // trailing junk after JSON object } return true }()
// Treat invalid JSON as a raw text message. if !validJSON { return lg.appendText(dst, src, lg.skipClientTime, lg.procID, lg.procSequence, level) }
// Check whether the JSON payload is too large. // Due to logtail metadata, the formatted log entry could exceed maxSize. // That's okay as the Tailscale log service limit is actually 2*maxSize. // However, so long as logging applications aim to target the maxSize limit, // there should be no trouble eventually uploading logs. maxLen := cmp.Or(lg.maxUploadSize, maxSize) if len(src) > maxLen { errDetail := fmt.Sprintf("entry too large: %d bytes", len(src)) errData := appendTruncatedString(nil, src, maxLen/len(`\uffff`)) // escaping could increase size
dst = append(dst, '{') dst = lg.appendMetadata(dst, lg.skipClientTime, true, lg.procID, lg.procSequence, errDetail, errData, level) dst = bytes.TrimRight(dst, ",") return append(dst, "}\n"...) }
// Check whether the reserved logtail member occurs in the log data. // If so, it is moved to the logtail/error member. const jsonSeperators = ",:" // per RFC 8259, section 2 const jsonWhitespace = " \n\r\t" // per RFC 8259, section 2 var errDetail string var errData jsontext.Value if logtailValLength > 0 { errDetail = "duplicate logtail member" errData = bytes.Trim(src[logtailValOffset:][:logtailValLength], jsonSeperators+jsonWhitespace) } dst = slices.Grow(dst, len(src)) dst = append(dst, '{') dst = lg.appendMetadata(dst, lg.skipClientTime, true, lg.procID, lg.procSequence, errDetail, errData, level) if logtailValLength > 0 { // Exclude original logtail member from the message. dst = appendWithoutNewline(dst, src[len("{"):logtailKeyOffset]) dst = bytes.TrimRight(dst, jsonSeperators+jsonWhitespace) dst = appendWithoutNewline(dst, src[logtailValOffset+logtailValLength:]) } else { dst = appendWithoutNewline(dst, src[len("{"):]) } dst = bytes.TrimRight(dst, jsonWhitespace) dst = dst[:len(dst)-len("}")] dst = bytes.TrimRight(dst, jsonSeperators+jsonWhitespace) return append(dst, "}\n"...)}
// appendWithoutNewline appends src to dst except that it ignores newlines// since newlines are used to frame individual log entries.func appendWithoutNewline(dst, src []byte) []byte { for _, c := range src { if c != '\n' { dst = append(dst, c) } } return dst}
// Logf logs to l using the provided fmt-style format and optional arguments.func (lg *Logger) Logf(format string, args ...any) { fmt.Fprintf(lg, format, args...)}
// Write logs an encoded JSON blob.//// If the []byte passed to Write is not an encoded JSON blob,// then contents is fit into a JSON blob and written.//// This is intended as an interface for the stdlib "log" package.func (lg *Logger) Write(buf []byte) (int, error) { if len(buf) == 0 { return 0, nil } inLen := len(buf) // length as provided to us, before modifications to downstream writers
level, buf := parseAndRemoveLogLevel(buf) if lg.stderr != nil && lg.stderr != io.Discard && int64(level) <= atomic.LoadInt64(&lg.stderrLevel) { if buf[len(buf)-1] == '\n' { lg.stderr.Write(buf) } else { // The log package always line-terminates logs, // so this is an uncommon path. withNL := append(buf[:len(buf):len(buf)], '\n') lg.stderr.Write(withNL) } }
lg.writeLock.Lock() defer lg.writeLock.Unlock()
b := lg.appendTextOrJSONLocked(lg.writeBuf[:0], buf, level) _, err := lg.sendLocked(b) return inLen, err}
var ( openBracketV = []byte("[v") v1 = []byte("[v1] ") v2 = []byte("[v2] ") vJSON = []byte("[v\x00JSON]") // precedes log level '0'-'9' byte, then JSON value)
// level 0 is normal (or unknown) level; 1+ are increasingly verbosefunc parseAndRemoveLogLevel(buf []byte) (level int, cleanBuf []byte) { if len(buf) == 0 || buf[0] == '{' || !bytes.Contains(buf, openBracketV) { return 0, buf } if bytes.Contains(buf, v1) { return 1, bytes.ReplaceAll(buf, v1, nil) } if bytes.Contains(buf, v2) { return 2, bytes.ReplaceAll(buf, v2, nil) } if _, after, ok := bytes.Cut(buf, vJSON); ok { rest := after if len(rest) >= 2 { v := rest[0] if v >= '0' && v <= '9' { return int(v - '0'), rest[1:] } } } return 0, buf}
var ( tapSetSize atomic.Int32 tapMu sync.Mutex tapSet set.HandleSet[chan<- string])
// RegisterLogTap registers dst to get a copy of every log write. The caller// must call unregister when done watching.//// This would ideally be a method on Logger, but Logger isn't really available// in most places; many writes go via stderr which filch redirects to the// singleton Logger set up early. For better or worse, there's basically only// one Logger within the program. This mechanism at least works well for// tailscaled. It works less well for a binary with multiple tsnet.Servers. Oh// well. This then subscribes to all of them.func RegisterLogTap(dst chan<- string) (unregister func()) { tapMu.Lock() defer tapMu.Unlock() h := tapSet.Add(dst) tapSetSize.Store(int32(len(tapSet))) return func() { tapMu.Lock() defer tapMu.Unlock() delete(tapSet, h) tapSetSize.Store(int32(len(tapSet))) }}
// tapSend relays the JSON blob to any/all registered local debug log watchers// (somebody running "tailscale debug daemon-logs").func tapSend(jsonBlob []byte) { if tapSetSize.Load() == 0 { return } s := string(jsonBlob) tapMu.Lock() defer tapMu.Unlock() for _, dst := range tapSet { select { case dst <- s: default: } }}