felixfoertsch/syncthing
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 10 Wiki Activity

feat: switch logging framework (#10220)

Browse Source 
This updates our logging framework from legacy freetext strings using
the `log` package to structured log entries using `log/slog`. I have
updated all INFO or higher level entries, but not yet DEBUG (😓)... So,
at a high level:

There is a slight change in log levels, effectively adding a new warning
level:

- DEBUG is still debug (ideally not for users but developers, though
this is something we need to work on)
- INFO is still info, though I've added more data here, effectively
making Syncthing more verbose by default (more on this below)
- WARNING is a new log level that is different from the _old_ WARNING
(more below)
- ERROR is what was WARNING before -- problems that must be dealt with,
and also bubbled as a popup in the GUI.

A new feature is that the logging level can be set per package to
something other than just debug or info, and hence I feel that we can
add a bit more things into INFO while moving some (in fact, most)
current INFO level warnings into WARNING. For example, I think it's
justified to get a log of synced files in INFO and sync failures in
WARNING. These are things that have historically been tricky to debug
properly, and having more information by default will be useful to many,
while still making it possible get close to told level of inscrutability
by setting the log level to WARNING. I'd like to get to a stage where
DEBUG is never necessary to just figure out what's going on, as opposed
to trying to narrow down a likely bug.

Code wise:

- Our logging object, generally known as `l` in each package, is now a
new adapter object that provides the old API on top of the newer one.
(This should go away once all old log entries are migrated.) This is
only for `l.Debugln` and `l.Debugf`.
- There is a new level tracker that keeps the log level for each
package.
- There is a nested setup of handlers, since the structure mandated by
`log/slog` is slightly convoluted (imho). We do this because we need to
do formatting at a "medium" level internally so we can buffer log lines
in text format but with separate timestamp and log level for the API/GUI
to consume.
- The `debug` API call becomes a `loglevels` API call, which can set the
log level to `DEBUG`, `INFO`, `WARNING` or `ERROR` per package. The GUI
is updated to handle this.
- Our custom `sync` package provided some debugging of mutexes quite
strongly integrated into the old logging framework, only turned on when
`STTRACE` was set to certain values at startup, etc. It's been a long
time since this has been useful; I removed it.
- The `STTRACE` env var remains and can be used the same way as before,
while additionally permitting specific log levels to be specified,
`STTRACE=model:WARN,scanner:DEBUG`.
- There is a new command line option `--log-level=INFO` to set the
default log level.
- The command line options `--log-flags` and `--verbose` go away, but
are currently retained as hidden & ignored options since we set them by
default in some of our startup examples and Syncthing would otherwise
fail to start.

Sample format messages:

```
2009-02-13 23:31:30 INF A basic info line (attr1="val with spaces" attr2=2 attr3="val\"quote" a=a log.pkg=slogutil)
2009-02-13 23:31:30 INF An info line with grouped values (attr1=val1 foo.attr2=2 foo.bar.attr3=3 a=a log.pkg=slogutil)
2009-02-13 23:31:30 INF An info line with grouped values via logger (foo.attr1=val1 foo.attr2=2 a=a log.pkg=slogutil)
2009-02-13 23:31:30 INF An info line with nested grouped values via logger (bar.foo.attr1=val1 bar.foo.attr2=2 a=a log.pkg=slogutil)
2009-02-13 23:31:30 WRN A warning entry (a=a log.pkg=slogutil)
2009-02-13 23:31:30 ERR An error (a=a log.pkg=slogutil)
```

---------

Co-authored-by: Ross Smith II <ross@smithii.com>
This commit is contained in:
Jakob Borg
2025-08-07 11:19:36 +02:00
committed by GitHub
co-authored by Ross Smith II
parent 49462448d0
commit 836045ee87
149 changed files with 1797 additions and 2641 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/connections/connections_test.go
+2 -2
View File
@@ -17,6 +17,7 @@ import (
"net"
"net/url"
"strings"
"sync"
"testing"
"time"
@@ -27,7 +28,6 @@ import (
"github.com/syncthing/syncthing/lib/events"
"github.com/syncthing/syncthing/lib/nat"
"github.com/syncthing/syncthing/lib/protocol"
"github.com/syncthing/syncthing/lib/sync"
"github.com/syncthing/syncthing/lib/tlsutil"
)
@@ -336,7 +336,7 @@ func BenchmarkConnections(b *testing.B) {
total := 0
b.ResetTimer()
for i := 0; i < b.N; i++ {
wg := sync.NewWaitGroup()
var wg sync.WaitGroup
wg.Add(2)
errC := make(chan error, 2)
go func() {
lib/connections/debug.go
+2 -4
View File
@@ -6,8 +6,6 @@
package connections
import (
"github.com/syncthing/syncthing/lib/logger"
)
import "github.com/syncthing/syncthing/internal/slogutil"
var l = logger.DefaultLogger.NewFacility("connections", "Connection handling")
var l = slogutil.NewAdapter("Connection handling")
lib/connections/limiter.go
+6 -8
View File
@@ -10,13 +10,14 @@ import (
"context"
"fmt"
"io"
"log/slog"
"sync"
"sync/atomic"
"golang.org/x/time/rate"
"github.com/syncthing/syncthing/lib/config"
"github.com/syncthing/syncthing/lib/protocol"
"github.com/syncthing/syncthing/lib/sync"
)
// limiter manages a read and write rate limit, reacting to config changes
@@ -46,7 +47,6 @@ func newLimiter(myId protocol.DeviceID, cfg config.Wrapper) *limiter {
myID: myId,
write: rate.NewLimiter(rate.Inf, limiterBurstSize),
read: rate.NewLimiter(rate.Inf, limiterBurstSize),
mu: sync.NewMutex(),
deviceReadLimiters: make(map[protocol.DeviceID]*rate.Limiter),
deviceWriteLimiters: make(map[protocol.DeviceID]*rate.Limiter),
}
@@ -107,15 +107,13 @@ func (lim *limiter) processDevicesConfigurationLocked(from, to config.Configurat
writeLimitStr = fmt.Sprintf("limit is %d KiB/s", dev.MaxSendKbps)
}
l.Infof("Device %s send rate %s, receive rate %s", dev.DeviceID, writeLimitStr, readLimitStr)
slog.Info("Device is rate limited", dev.DeviceID.LogAttr(), slog.String("send", writeLimitStr), slog.String("recv", readLimitStr))
}
}
// Delete remote devices which were removed in new configuration
for _, dev := range from.Devices {
if _, ok := seen[dev.DeviceID]; !ok {
l.Debugf("deviceID: %s should be removed", dev.DeviceID)
delete(lim.deviceWriteLimiters, dev.DeviceID)
delete(lim.deviceReadLimiters, dev.DeviceID)
}
@@ -160,13 +158,13 @@ func (lim *limiter) CommitConfiguration(from, to config.Configuration) bool {
lim.limitsLAN.Store(to.Options.LimitBandwidthInLan)
l.Infof("Overall send rate %s, receive rate %s", sendLimitStr, recvLimitStr)
slog.Info("Overall rate limit in use", "send", sendLimitStr, "recv", recvLimitStr)
if limited {
if to.Options.LimitBandwidthInLan {
l.Infoln("Rate limits apply to LAN connections")
slog.Info("Rate limits apply to LAN connections")
} else {
l.Infoln("Rate limits do not apply to LAN connections")
slog.Info("Rate limits do not apply to LAN connections")
}
}
lib/connections/quic_listen.go
+12 -10
View File
@@ -13,6 +13,7 @@ import (
"context"
"crypto/tls"
"errors"
"log/slog"
"net"
"net/url"
"sync"
@@ -21,6 +22,7 @@ import (
"github.com/quic-go/quic-go"
"github.com/syncthing/syncthing/internal/slogutil"
"github.com/syncthing/syncthing/lib/config"
"github.com/syncthing/syncthing/lib/connections/registry"
"github.com/syncthing/syncthing/lib/nat"
@@ -58,7 +60,7 @@ type quicListener struct {
func (t *quicListener) OnNATTypeChanged(natType stun.NATType) {
if natType != stun.NATUnknown {
l.Infof("%s detected NAT type: %s", t.uri, natType)
slog.Info("Detected NAT type", slogutil.URI(t.uri), slog.Any("type", natType))
}
t.nat.Store(uint64(natType))
}
@@ -77,7 +79,7 @@ func (t *quicListener) OnExternalAddressChanged(address *stun.Host, via string)
t.mut.Unlock()
if uri != nil && (existingAddress == nil || existingAddress.String() != uri.String()) {
l.Infof("%s resolved external address %s (via %s)", t.uri, uri.String(), via)
slog.Info("Resolved external address", slogutil.URI(t.uri), slogutil.Address(uri.String()), slog.String("via", via))
t.notifyAddressesChanged(t)
} else if uri == nil && existingAddress != nil {
t.notifyAddressesChanged(t)
@@ -89,13 +91,13 @@ func (t *quicListener) serve(ctx context.Context) error {
udpAddr, err := net.ResolveUDPAddr(network, t.uri.Host)
if err != nil {
l.Infoln("Listen (BEP/quic):", err)
slog.WarnContext(ctx, "Failed to listen (QUIC)", slogutil.Error(err))
return err
}
udpConn, err := net.ListenUDP(network, udpAddr)
if err != nil {
l.Infoln("Listen (BEP/quic):", err)
slog.WarnContext(ctx, "Failed to listen (QUIC)", slogutil.Error(err))
return err
}
defer udpConn.Close()
@@ -117,7 +119,7 @@ func (t *quicListener) serve(ctx context.Context) error {
listener, err := quicTransport.Listen(t.tlsCfg, quicConfig)
if err != nil {
l.Infoln("Listen (BEP/quic):", err)
slog.WarnContext(ctx, "Failed to listen (QUIC)", slogutil.Error(err))
return err
}
defer listener.Close()
@@ -125,8 +127,8 @@ func (t *quicListener) serve(ctx context.Context) error {
t.notifyAddressesChanged(t)
defer t.clearAddresses(t)
l.Infof("QUIC listener (%v) starting", udpConn.LocalAddr())
defer l.Infof("QUIC listener (%v) shutting down", udpConn.LocalAddr())
slog.InfoContext(ctx, "QUIC listener starting", slogutil.Address(udpConn.LocalAddr()))
defer slog.InfoContext(ctx, "QUIC listener shutting down", slogutil.Address(udpConn.LocalAddr()))
var ipVersion nat.IPVersion
switch t.uri.Scheme {
@@ -168,7 +170,7 @@ func (t *quicListener) serve(ctx context.Context) error {
if errors.Is(err, context.Canceled) {
return nil
} else if err != nil {
l.Infoln("Listen (BEP/quic): Accepting connection:", err)
slog.WarnContext(ctx, "Failed to accept QUIC connection", slogutil.Error(err))
acceptFailures++
if acceptFailures > maxAcceptFailures {
@@ -185,13 +187,13 @@ func (t *quicListener) serve(ctx context.Context) error {
acceptFailures = 0
l.Debugln("connect from", session.RemoteAddr())
slog.DebugContext(ctx, "Incoming connection", "from", session.RemoteAddr())
streamCtx, cancel := context.WithTimeout(ctx, quicOperationTimeout)
stream, err := session.AcceptStream(streamCtx)
cancel()
if err != nil {
l.Debugf("failed to accept stream from %s: %v", session.RemoteAddr(), err)
slog.DebugContext(ctx, "Failed to accept stream", slogutil.Address(session.RemoteAddr()), slogutil.Error(err))
_ = session.CloseWithError(1, err.Error())
continue
}
lib/connections/registry/registry.go
+1 -2
View File
@@ -11,9 +11,9 @@ package registry
import (
"strings"
"sync"
"github.com/syncthing/syncthing/lib/sliceutil"
"github.com/syncthing/syncthing/lib/sync"
)
type Registry struct {
@@ -23,7 +23,6 @@ type Registry struct {
func New() *Registry {
return &Registry{
mut: sync.NewMutex(),
available: make(map[string][]interface{}),
}
}
lib/connections/relay_listen.go
+9 -7
View File
@@ -10,10 +10,12 @@ import (
"context"
"crypto/tls"
"errors"
"log/slog"
"net/url"
"sync"
"time"
"github.com/syncthing/syncthing/internal/slogutil"
"github.com/syncthing/syncthing/lib/config"
"github.com/syncthing/syncthing/lib/connections/registry"
"github.com/syncthing/syncthing/lib/dialer"
@@ -46,7 +48,7 @@ type relayListener struct {
func (t *relayListener) serve(ctx context.Context) error {
clnt, err := client.NewClient(t.uri, t.tlsCfg.Certificates, 10*time.Second)
if err != nil {
l.Infoln("Listen (BEP/relay):", err)
slog.WarnContext(ctx, "Failed to listen (relay)", slogutil.Error(err))
return err
}
@@ -54,8 +56,8 @@ func (t *relayListener) serve(ctx context.Context) error {
t.client = clnt
t.mut.Unlock()
l.Infof("Relay listener (%v) starting", t)
defer l.Infof("Relay listener (%v) shutting down", t)
slog.InfoContext(ctx, "Relay listener starting", "id", t.String())
defer slog.InfoContext(ctx, "Relay listener shutting down", "id", t.String())
defer t.clearAddresses(t)
invitationCtx, cancel := context.WithCancel(ctx)
@@ -77,19 +79,19 @@ func (t *relayListener) handleInvitations(ctx context.Context, clnt client.Relay
conn, err := client.JoinSession(ctx, inv)
if err != nil {
if !errors.Is(err, context.Canceled) {
l.Infoln("Listen (BEP/relay): joining session:", err)
slog.InfoContext(ctx, "Failed to join session", slogutil.Error(err))
}
continue
}
err = dialer.SetTCPOptions(conn)
if err != nil {
l.Debugln("Listen (BEP/relay): setting tcp options:", err)
slog.DebugContext(ctx, "Failed to set TCP options", slogutil.Error(err))
}
err = dialer.SetTrafficClass(conn, t.cfg.Options().TrafficClass)
if err != nil {
l.Debugln("Listen (BEP/relay): setting traffic class:", err)
slog.DebugContext(ctx, "Failed to set traffic class", slogutil.Error(err))
}
var tc *tls.Conn
@@ -102,7 +104,7 @@ func (t *relayListener) handleInvitations(ctx context.Context, clnt client.Relay
err = tlsTimedHandshake(tc)
if err != nil {
tc.Close()
l.Infoln("Listen (BEP/relay): TLS handshake:", err)
slog.WarnContext(ctx, "Failed TLS handshake", slogutil.Error(err))
continue
}
lib/connections/service.go
+32 -72
View File
@@ -19,17 +19,18 @@ import (
"errors"
"fmt"
"io"
"log/slog"
"math"
"net"
"net/url"
"slices"
"strings"
stdsync "sync"
"sync"
"time"
"github.com/thejerf/suture/v4"
"golang.org/x/time/rate"
"github.com/syncthing/syncthing/internal/slogutil"
"github.com/syncthing/syncthing/lib/build"
"github.com/syncthing/syncthing/lib/config"
"github.com/syncthing/syncthing/lib/connections/registry"
@@ -42,7 +43,6 @@ import (
"github.com/syncthing/syncthing/lib/sliceutil"
"github.com/syncthing/syncthing/lib/stringutil"
"github.com/syncthing/syncthing/lib/svcutil"
"github.com/syncthing/syncthing/lib/sync"
// Registers NAT service providers
_ "github.com/syncthing/syncthing/lib/pmp"
@@ -185,7 +185,7 @@ type service struct {
}
func NewService(cfg config.Wrapper, myID protocol.DeviceID, mdl Model, tlsCfg *tls.Config, discoverer discover.Finder, bepProtocolName string, tlsDefaultCommonName string, evLogger events.Logger, registry *registry.Registry, keyGen *protocol.KeyGenerator) Service {
spec := svcutil.SpecWithInfoLogger(l)
spec := svcutil.SpecWithInfoLogger()
service := &service{
Supervisor: suture.New("connections.Service", spec),
connectionStatusHandler: newConnectionStatusHandler(),
@@ -206,11 +206,9 @@ func NewService(cfg config.Wrapper, myID protocol.DeviceID, mdl Model, tlsCfg *t
keyGen: keyGen,
lanChecker: &lanChecker{cfg},
dialNowDevicesMut: sync.NewMutex(),
dialNow: make(chan struct{}, 1),
dialNowDevices: make(map[protocol.DeviceID]struct{}),
dialNow: make(chan struct{}, 1),
dialNowDevices: make(map[protocol.DeviceID]struct{}),
listenersMut: sync.NewRWMutex(),
listeners: make(map[string]genericListener),
listenerTokens: make(map[string]suture.ServiceToken),
}
@@ -257,7 +255,7 @@ func (s *service) handleConns(ctx context.Context) error {
// because there are implementations out there that don't support
// protocol negotiation (iOS for one...).
if cs.NegotiatedProtocol != s.bepProtocolName {
l.Infof("Peer at %s did not negotiate bep/1.0", c)
slog.WarnContext(ctx, "Peer at did not negotiate bep/1.0", slogutil.Address(c.RemoteAddr()))
}
// We should have received exactly one certificate from the other
@@ -265,7 +263,7 @@ func (s *service) handleConns(ctx context.Context) error {
// connection.
certs := cs.PeerCertificates
if cl := len(certs); cl != 1 {
l.Infof("Got peer certificate list of length %d != 1 from peer at %s; protocol error", cl, c)
slog.WarnContext(ctx, "Got peer certificate list of incorrect length", slog.Int("length", cl), slogutil.Address(c.RemoteAddr()))
c.Close()
continue
}
@@ -276,17 +274,13 @@ func (s *service) handleConns(ctx context.Context) error {
// though, especially in the presence of NAT hairpinning, multiple
// clients between the same NAT gateway, and global discovery.
if remoteID == s.myID {
l.Debugf("Connected to myself (%s) at %s", remoteID, c)
slog.DebugContext(ctx, "Connected to myself", "id", remoteID, "addr", c)
c.Close()
continue
}
if err := s.connectionCheckEarly(remoteID, c); err != nil {
if errors.Is(err, errDeviceAlreadyConnected) {
l.Debugf("Connection from %s at %s (%s) rejected: %v", remoteID, c.RemoteAddr(), c.Type(), err)
} else {
l.Infof("Connection from %s at %s (%s) rejected: %v", remoteID, c.RemoteAddr(), c.Type(), err)
}
slog.DebugContext(ctx, "Connection rejected", remoteID.LogAttr(), slogutil.Address(c.RemoteAddr()), slog.String("type", c.Type()), slogutil.Error(err))
c.Close()
continue
}
@@ -382,22 +376,10 @@ func (s *service) handleHellos(ctx context.Context) error {
if err != nil {
if protocol.IsVersionMismatch(err) {
// The error will be a relatively user friendly description
// of what's wrong with the version compatibility. By
// default identify the other side by device ID and IP.
remote := fmt.Sprintf("%v (%v)", remoteID, c.RemoteAddr())
if hello.DeviceName != "" {
// If the name was set in the hello return, use that to
// give the user more info about which device is the
// affected one. It probably says more than the remote
// IP.
remote = fmt.Sprintf("%q (%s %s, %v)", hello.DeviceName, hello.ClientName, hello.ClientVersion, remoteID)
}
msg := fmt.Sprintf("Connecting to %s: %s", remote, err)
warningFor(remoteID, msg)
slog.WarnContext(ctx, "Remote device is too old", remoteID.LogAttr(), slogutil.Address(c.RemoteAddr()), slogutil.Error(err))
} else {
// It's something else - connection reset or whatever
l.Infof("Failed to exchange Hello messages with %s at %s: %s", remoteID, c, err)
slog.WarnContext(ctx, "Failed to exchange Hello messages", remoteID.LogAttr(), slogutil.Address(c.RemoteAddr()), slogutil.Error(err))
}
c.Close()
continue
@@ -407,14 +389,14 @@ func (s *service) handleHellos(ctx context.Context) error {
// The Model will return an error for devices that we don't want to
// have a connection with for whatever reason, for example unknown devices.
if err := s.model.OnHello(remoteID, c.RemoteAddr(), hello); err != nil {
l.Infof("Connection from %s at %s (%s) rejected: %v", remoteID, c.RemoteAddr(), c.Type(), err)
slog.WarnContext(ctx, "Connection rejected", remoteID.LogAttr(), slogutil.Address(c.RemoteAddr()), slog.Any("type", c.Type()), slogutil.Error(err))
c.Close()
continue
}
deviceCfg, ok := s.cfg.Device(remoteID)
if !ok {
l.Infof("Device %s removed from config during connection attempt at %s", remoteID, c)
slog.WarnContext(ctx, "Device removed from config during connection attempt", remoteID.LogAttr(), slogutil.Address(c.RemoteAddr()))
c.Close()
continue
}
@@ -434,7 +416,7 @@ func (s *service) handleHellos(ctx context.Context) error {
// Incorrect certificate name is something the user most
// likely wants to know about, since it's an advanced
// config. Warn instead of Info.
l.Warnf("Bad certificate from %s at %s: %v", remoteID, c, err)
slog.ErrorContext(ctx, "Bad certificate from remote", remoteID.LogAttr(), slogutil.Address(c.RemoteAddr()), slogutil.Error(err))
c.Close()
continue
}
@@ -455,7 +437,7 @@ func (s *service) handleHellos(ctx context.Context) error {
s.dialNowDevicesMut.Unlock()
}()
l.Infof("Established secure connection to %s at %s", remoteID.Short(), c)
slog.InfoContext(ctx, "Established secure connection", remoteID.LogAttr(), slog.Any("connection", c))
s.model.AddConnection(protoConn, hello)
continue
@@ -477,9 +459,9 @@ func (s *service) connect(ctx context.Context) error {
bestDialerPriority := s.bestDialerPriority(cfg)
isInitialRampup := initialRampup < stdConnectionLoopSleep
l.Debugln("Connection loop")
slog.DebugContext(ctx, "Connection loop")
if isInitialRampup {
l.Debugln("Connection loop in initial rampup")
slog.DebugContext(ctx, "Connection loop in initial rampup")
}
// Used for consistency throughout this loop run, as time passes
@@ -616,9 +598,9 @@ func (s *service) dialDevices(ctx context.Context, now time.Time, cfg config.Con
// Perform dials according to the queue, stopping when we've reached the
// allowed additional number of connections (if limited).
numConns := 0
var numConnsMut stdsync.Mutex
var numConnsMut sync.Mutex
dialSemaphore := semaphore.New(dialMaxParallel)
dialWG := new(stdsync.WaitGroup)
dialWG := new(sync.WaitGroup)
dialCtx, dialCancel := context.WithCancel(ctx)
defer func() {
dialWG.Wait()
@@ -675,7 +657,7 @@ func (s *service) resolveDialTargets(ctx context.Context, now time.Time, cfg con
uri, err := url.Parse(addr)
if err != nil {
s.setConnectionStatus(addr, err)
l.Infof("Parsing dialer address %s: %v", addr, err)
slog.WarnContext(ctx, "Failed to parse dialer address", slogutil.Address(addr), slogutil.Error(err))
continue
}
@@ -695,7 +677,7 @@ func (s *service) resolveDialTargets(ctx context.Context, now time.Time, cfg con
l.Debugf("Dialer for %v: %v", uri, err)
continue
} else if err != nil {
l.Infof("Dialer for %v: %v", uri, err)
slog.WarnContext(ctx, "Failed to get dialer", slogutil.URI(uri), slogutil.Error(err))
continue
}
@@ -711,7 +693,7 @@ func (s *service) resolveDialTargets(ctx context.Context, now time.Time, cfg con
continue
}
if currentConns >= s.desiredConnectionsToDevice(deviceCfg.DeviceID) && priority == priorityCutoff {
l.Debugf("Not dialing %s at %s using %s as priority is equal and we already have %d/%d connections", deviceID.Short(), addr, dialerFactory, currentConns, deviceCfg.NumConnections)
l.Debugf("Not dialing %s at %s using %s as priority is equal and we already have %d/%d connections", deviceID.Short(), addr, dialerFactory, currentConns, deviceCfg.NumConnections())
continue
}
@@ -818,7 +800,7 @@ func (s *lanChecker) isLAN(addr net.Addr) bool {
func (s *service) createListener(factory listenerFactory, uri *url.URL) bool {
// must be called with listenerMut held
l.Debugln("Starting listener", uri)
slog.Debug("Starting listener", "uri", uri)
listener := factory.New(uri, s.cfg, s.tlsCfg, s.conns, s.natService, s.registry, s.lanChecker)
listener.OnAddressesChanged(s.logListenAddressesChangedEvent)
@@ -826,7 +808,7 @@ func (s *service) createListener(factory listenerFactory, uri *url.URL) bool {
// Retrying a listener many times in rapid succession is unlikely to help,
// thus back off quickly. A listener may soon be functional again, e.g. due
// to a network interface coming back online - retry every minute.
spec := svcutil.SpecWithInfoLogger(l)
spec := svcutil.SpecWithInfoLogger()
spec.FailureThreshold = 2
spec.FailureBackoff = time.Minute
sup := suture.New(fmt.Sprintf("listenerSupervisor@%v", listener), spec)
@@ -854,9 +836,6 @@ func (s *service) CommitConfiguration(from, to config.Configuration) bool {
for _, dev := range from.Devices {
if !newDevices[dev.DeviceID] {
warningLimitersMut.Lock()
delete(warningLimiters, dev.DeviceID)
warningLimitersMut.Unlock()
metricDeviceActiveConnections.DeleteLabelValues(dev.DeviceID.String())
}
}
@@ -875,7 +854,7 @@ func (s *service) CommitConfiguration(from, to config.Configuration) bool {
uri, err := url.Parse(addr)
if err != nil {
l.Warnf("Skipping malformed listener URL %q: %v", addr, err)
slog.Error("Skipping malformed listener URL", slogutil.URI(addr), slogutil.Error(err))
continue
}
@@ -886,7 +865,7 @@ func (s *service) CommitConfiguration(from, to config.Configuration) bool {
// mean something entirely different to the computer (e.g.,
// tcp:/127.0.0.1:22000 in fact being equivalent to tcp://:22000).
if canonical := uri.String(); canonical != addr {
l.Warnf("Skipping malformed listener URL %q (not canonical)", addr)
slog.Error("Skipping malformed listener URL (not canonical)", slogutil.URI(addr))
continue
}
@@ -900,7 +879,7 @@ func (s *service) CommitConfiguration(from, to config.Configuration) bool {
l.Debugf("Listener for %v: %v", uri, err)
continue
} else if err != nil {
l.Infof("Listener for %v: %v", uri, err)
slog.Warn("Failed to get listener", slogutil.URI(uri), slogutil.Error(err))
continue
}
@@ -1007,8 +986,7 @@ type connectionStatusHandler struct {
func newConnectionStatusHandler() connectionStatusHandler {
return connectionStatusHandler{
connectionStatusMut: sync.NewRWMutex(),
connectionStatus: make(map[string]ConnectionStatusEntry),
connectionStatus: make(map[string]ConnectionStatusEntry),
}
}
@@ -1082,24 +1060,6 @@ func urlsToStrings(urls []*url.URL) []string {
return strings
}
var (
warningLimiters = make(map[protocol.DeviceID]*rate.Limiter)
warningLimitersMut = sync.NewMutex()
)
func warningFor(dev protocol.DeviceID, msg string) {
warningLimitersMut.Lock()
defer warningLimitersMut.Unlock()
lim, ok := warningLimiters[dev]
if !ok {
lim = rate.NewLimiter(rate.Every(perDeviceWarningIntv), 1)
warningLimiters[dev] = lim
}
if lim.Allow() {
l.Warnln(msg)
}
}
func tlsTimedHandshake(tc *tls.Conn) error {
tc.SetDeadline(time.Now().Add(tlsHandshakeTimeout))
defer tc.SetDeadline(time.Time{})
@@ -1156,7 +1116,7 @@ func (s *service) dialParallel(ctx context.Context, deviceID protocol.DeviceID,
for _, prio := range priorities {
tgts := dialTargetBuckets[prio]
res := make(chan internalConn, len(tgts))
wg := stdsync.WaitGroup{}
wg := sync.WaitGroup{}
for _, tgt := range tgts {
sema.Take(1)
wg.Add(1)
@@ -1215,7 +1175,7 @@ func (s *service) validateIdentity(c internalConn, expectedID protocol.DeviceID)
// connection.
certs := cs.PeerCertificates
if cl := len(certs); cl != 1 {
l.Infof("Got peer certificate list of length %d != 1 from peer at %s; protocol error", cl, c)
slog.Warn("Got peer certificate list of incorrect length", slog.Int("length", cl), slogutil.Address(c.RemoteAddr()))
c.Close()
return fmt.Errorf("expected 1 certificate, got %d", cl)
}
@@ -1364,7 +1324,7 @@ func (s *service) desiredConnectionsToDevice(deviceID protocol.DeviceID) int {
// connected to and how many connections we have to each device. It also
// tracks how many connections they are willing to use.
type deviceConnectionTracker struct {
connectionsMut stdsync.Mutex
connectionsMut sync.Mutex
connections map[protocol.DeviceID][]protocol.Connection // current connections
wantConnections map[protocol.DeviceID]int // number of connections they want
}
lib/connections/structs.go
+5
View File
@@ -11,6 +11,7 @@ import (
"crypto/tls"
"fmt"
"io"
"log/slog"
"net"
"net/url"
"time"
@@ -152,6 +153,10 @@ func (c internalConn) String() string {
return fmt.Sprintf("%s-%s/%s/%s/%s-P%d-%s", c.LocalAddr(), c.RemoteAddr(), c.Type(), c.Crypto(), t, c.Priority(), c.connectionID)
}
func (c internalConn) LogValue() slog.Value {
return slog.GroupValue(slog.String("local", c.LocalAddr().String()), slog.String("remote", c.RemoteAddr().String()), slog.String("type", c.Type()), slog.Bool("lan", c.isLocal), slog.String("crypto", c.Crypto()), slog.Int("prio", c.priority), slog.String("id", c.ConnectionID()))
}
type dialerFactory interface {
New(config.OptionsConfiguration, *tls.Config, *registry.Registry, *lanChecker) genericDialer
AlwaysWAN() bool
lib/connections/tcp_listen.go
+11 -7
View File
@@ -9,11 +9,14 @@ package connections
import (
"context"
"crypto/tls"
"errors"
"log/slog"
"net"
"net/url"
"sync"
"time"
"github.com/syncthing/syncthing/internal/slogutil"
"github.com/syncthing/syncthing/lib/config"
"github.com/syncthing/syncthing/lib/connections/registry"
"github.com/syncthing/syncthing/lib/dialer"
@@ -50,7 +53,7 @@ type tcpListener struct {
func (t *tcpListener) serve(ctx context.Context) error {
tcaddr, err := net.ResolveTCPAddr(t.uri.Scheme, t.uri.Host)
if err != nil {
l.Infoln("Listen (BEP/tcp):", err)
slog.WarnContext(ctx, "Failed to listen (TCP)", slogutil.Error(err))
return err
}
@@ -60,7 +63,7 @@ func (t *tcpListener) serve(ctx context.Context) error {
listener, err := lc.Listen(context.TODO(), t.uri.Scheme, tcaddr.String())
if err != nil {
l.Infoln("Listen (BEP/tcp):", err)
slog.WarnContext(ctx, "Failed to listen (TCP)", slogutil.Error(err))
return err
}
defer listener.Close()
@@ -74,8 +77,8 @@ func (t *tcpListener) serve(ctx context.Context) error {
t.registry.Register(t.uri.Scheme, tcaddr)
defer t.registry.Unregister(t.uri.Scheme, tcaddr)
l.Infof("TCP listener (%v) starting", tcaddr)
defer l.Infof("TCP listener (%v) shutting down", tcaddr)
slog.InfoContext(ctx, "TCP listener starting", slogutil.Address(tcaddr))
defer slog.InfoContext(ctx, "TCP listener shutting down", slogutil.Address(tcaddr))
var ipVersion nat.IPVersion
if t.uri.Scheme == "tcp4" {
@@ -121,8 +124,9 @@ func (t *tcpListener) serve(ctx context.Context) error {
default:
}
if err != nil {
if err, ok := err.(*net.OpError); !ok || !err.Timeout() {
l.Warnln("Listen (BEP/tcp): Accepting connection:", err)
var ne *net.OpError
if ok := errors.As(err, &ne); !ok || !ne.Timeout() {
slog.WarnContext(ctx, "Failed to accept TCP connection", slogutil.Error(err))
acceptFailures++
if acceptFailures > maxAcceptFailures {
@@ -152,7 +156,7 @@ func (t *tcpListener) serve(ctx context.Context) error {
tc := tls.Server(conn, t.tlsCfg)
if err := tlsTimedHandshake(tc); err != nil {
l.Infoln("Listen (BEP/tcp): TLS handshake:", err)
slog.WarnContext(ctx, "Failed TLS handshake", slogutil.Address(tc.RemoteAddr()), slogutil.Error(err))
tc.Close()
continue
}