mirrors/tailscale
1
0
Fork 0
mirror of https://github.com/tailscale/tailscale.git synced 2025-12-03 08:22:00 +01:00
Code Issues Packages Projects Releases Wiki Activity

Revert "logtail: avoid racing eventbus subscriptions with Shutdown (#17639)" (#17684)

Browse Source 
This reverts commit 4346615d77a6de16854c6e78f9d49375d6424e6e.
We averted the shutdown race, but will need to service the subscriber even when
we are not waiting for a change so that we do not delay the bus as a whole.

Updates #17638

Change-Id: I5488466ed83f5ad1141c95267f5ae54878a24657
Signed-off-by: M. J. Fromberger <fromberger@tailscale.com>
This commit is contained in:
M. J. Fromberger 2025-10-28 08:45:22 -07:00 committed by GitHub
parent 02681732d1
commit db5815fb97
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 20 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
logtail/logtail.go
View File

@ -124,7 +124,6 @@ func NewLogger(cfg Config, logf tslogger.Logf) *Logger {
if cfg.Bus != nil { if cfg.Bus != nil {
l.eventClient = cfg.Bus.Client("logtail.Logger") l.eventClient = cfg.Bus.Client("logtail.Logger")
l.changeDeltaSub = eventbus.Subscribe[netmon.ChangeDelta](l.eventClient)
} }
l.SetSockstatsLabel(sockstats.LabelLogtailLogger) l.SetSockstatsLabel(sockstats.LabelLogtailLogger)
l.compressLogs = cfg.CompressLogs l.compressLogs = cfg.CompressLogs
@ -163,7 +162,6 @@ type Logger struct {
httpDoCalls atomic.Int32 httpDoCalls atomic.Int32
sockstatsLabel atomicSocktatsLabel sockstatsLabel atomicSocktatsLabel
eventClient *eventbus.Client eventClient *eventbus.Client
changeDeltaSub *eventbus.Subscriber[netmon.ChangeDelta]
procID uint32 procID uint32
includeProcSequence bool includeProcSequence bool
@ -429,23 +427,8 @@ func (l *Logger) internetUp() bool {
func (l *Logger) awaitInternetUp(ctx context.Context) { func (l *Logger) awaitInternetUp(ctx context.Context) {
if l.eventClient != nil { if l.eventClient != nil {
for { l.awaitInternetUpBus(ctx)
if l.internetUp() { return
return
}
select {
case <-ctx.Done():
return // give up
case <-l.changeDeltaSub.Done():
return // give up (closing down)
case delta := <-l.changeDeltaSub.Events():
if delta.New.AnyInterfaceUp() || l.internetUp() {
fmt.Fprintf(l.stderr, "logtail: internet back up\n")
return
}
fmt.Fprintf(l.stderr, "logtail: network changed, but is not up")
}
}
} }
upc := make(chan bool, 1) upc := make(chan bool, 1)
defer l.netMonitor.RegisterChangeCallback(func(delta *netmon.ChangeDelta) { defer l.netMonitor.RegisterChangeCallback(func(delta *netmon.ChangeDelta) {
@ -466,6 +449,24 @@ func (l *Logger) awaitInternetUp(ctx context.Context) {
} }
} }
func (l *Logger) awaitInternetUpBus(ctx context.Context) {
if l.internetUp() {
return
}
sub := eventbus.Subscribe[netmon.ChangeDelta](l.eventClient)
defer sub.Close()
select {
case delta := <-sub.Events():
if delta.New.AnyInterfaceUp() {
fmt.Fprintf(l.stderr, "logtail: internet back up\n")
return
}
fmt.Fprintf(l.stderr, "logtail: network changed, but is not up")
case <-ctx.Done():
return
}
}
// upload uploads body to the log server. // upload uploads body to the log server.
// origlen indicates the pre-compression body length. // origlen indicates the pre-compression body length.
// origlen of -1 indicates that the body is not compressed. // origlen of -1 indicates that the body is not compressed.