|
|
@@ -25,6 +25,7 @@ import (
|
|
|
"sync/atomic"
|
|
|
"time"
|
|
|
|
|
|
+ "github.com/creachadair/msync/trigger"
|
|
|
"github.com/go-json-experiment/json/jsontext"
|
|
|
"tailscale.com/envknob"
|
|
|
"tailscale.com/net/netmon"
|
|
|
@@ -124,6 +125,8 @@ func NewLogger(cfg Config, logf tslogger.Logf) *Logger {
|
|
|
|
|
|
if cfg.Bus != nil {
|
|
|
l.eventClient = cfg.Bus.Client("logtail.Logger")
|
|
|
+ // Subscribe to change deltas from NetMon to detect when the network comes up.
|
|
|
+ eventbus.SubscribeFunc(l.eventClient, l.onChangeDelta)
|
|
|
}
|
|
|
l.SetSockstatsLabel(sockstats.LabelLogtailLogger)
|
|
|
l.compressLogs = cfg.CompressLogs
|
|
|
@@ -162,6 +165,7 @@ type Logger struct {
|
|
|
httpDoCalls atomic.Int32
|
|
|
sockstatsLabel atomicSocktatsLabel
|
|
|
eventClient *eventbus.Client
|
|
|
+ networkIsUp trigger.Cond // set/reset by netmon.ChangeDelta events
|
|
|
|
|
|
procID uint32
|
|
|
includeProcSequence bool
|
|
|
@@ -418,16 +422,36 @@ func (l *Logger) uploading(ctx context.Context) {
|
|
|
}
|
|
|
|
|
|
func (l *Logger) internetUp() bool {
|
|
|
- if l.netMonitor == nil {
|
|
|
- // No way to tell, so assume it is.
|
|
|
+ select {
|
|
|
+ case <-l.networkIsUp.Ready():
|
|
|
return true
|
|
|
+ default:
|
|
|
+ if l.netMonitor == nil {
|
|
|
+ return true // No way to tell, so assume it is.
|
|
|
+ }
|
|
|
+ return l.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 (l *Logger) onChangeDelta(delta *netmon.ChangeDelta) {
|
|
|
+ if delta.New.AnyInterfaceUp() {
|
|
|
+ fmt.Fprintf(l.stderr, "logtail: internet back up\n")
|
|
|
+ l.networkIsUp.Set()
|
|
|
+ } else {
|
|
|
+ fmt.Fprintf(l.stderr, "logtail: network changed, but is not up\n")
|
|
|
+ l.networkIsUp.Reset()
|
|
|
}
|
|
|
- return l.netMonitor.InterfaceState().AnyInterfaceUp()
|
|
|
}
|
|
|
|
|
|
func (l *Logger) awaitInternetUp(ctx context.Context) {
|
|
|
if l.eventClient != nil {
|
|
|
- l.awaitInternetUpBus(ctx)
|
|
|
+ select {
|
|
|
+ case <-l.networkIsUp.Ready():
|
|
|
+ case <-ctx.Done():
|
|
|
+ }
|
|
|
return
|
|
|
}
|
|
|
upc := make(chan bool, 1)
|
|
|
@@ -449,24 +473,6 @@ 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.
|
|
|
// origlen indicates the pre-compression body length.
|
|
|
// origlen of -1 indicates that the body is not compressed.
|
M. J. Fromberger