feat(timesync): query servers in parallel

This commit is contained in:
Siyuan Miao 2025-04-14 06:33:14 +02:00
parent 73b83557af
commit e470371d23
6 changed files with 175 additions and 75 deletions

View File

@ -1,22 +1,37 @@
--- version: "2"
linters: linters:
enable: enable:
- forbidigo - forbidigo
- goimports - misspell
- misspell - whitespace
# - revive settings:
- whitespace forbidigo:
forbid:
issues: - pattern: ^fmt\.Print.*$
exclude-rules: msg: Do not commit print statements. Use logger package.
- path: _test.go - pattern: ^log\.(Fatal|Panic|Print)(f|ln)?.*$
linters: msg: Do not commit log statements. Use logger package.
- errcheck exclusions:
generated: lax
linters-settings: presets:
forbidigo: - comments
forbid: - common-false-positives
- p: ^fmt\.Print.*$ - legacy
msg: Do not commit print statements. Use logger package. - std-error-handling
- p: ^log\.(Fatal|Panic|Print)(f|ln)?.*$ rules:
msg: Do not commit log statements. Use logger package. - linters:
- errcheck
path: _test.go
paths:
- third_party$
- builtin$
- examples$
formatters:
enable:
- goimports
exclusions:
generated: lax
paths:
- third_party$
- builtin$
- examples$

View File

@ -91,7 +91,7 @@ cd "${REMOTE_PATH}"
chmod +x jetkvm_app_debug chmod +x jetkvm_app_debug
# Run the application in the background # Run the application in the background
PION_LOG_TRACE=jetkvm,cloud,websocket,native ./jetkvm_app_debug PION_LOG_TRACE=jetkvm,cloud,websocket,native,jsonrpc ./jetkvm_app_debug
EOF EOF
echo "Deployment complete." echo "Deployment complete."

View File

@ -1,18 +1,98 @@
package timesync package timesync
import ( import (
"context"
"errors"
"math/rand"
"net/http" "net/http"
"time" "time"
) )
func (t *TimeSync) queryAllHttpTime() (now *time.Time) {
chunkSize := 4
httpUrls := t.httpUrls
// shuffle the http urls to avoid always querying the same servers
rand.Shuffle(len(httpUrls), func(i, j int) { httpUrls[i], httpUrls[j] = httpUrls[j], httpUrls[i] })
for i := 0; i < len(httpUrls); i += chunkSize {
chunk := httpUrls[i:min(i+chunkSize, len(httpUrls))]
results := t.queryMultipleHttp(chunk, timeSyncTimeout)
if results != nil {
return results
}
}
return nil
}
func (t *TimeSync) queryMultipleHttp(urls []string, timeout time.Duration) (now *time.Time) {
results := make(chan *time.Time, len(urls))
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
for _, url := range urls {
go func(url string) {
scopedLogger := t.l.With().
Str("http_url", url).
Logger()
startTime := time.Now()
now, err, response := queryHttpTime(
ctx,
url,
timeout,
)
duration := time.Since(startTime)
var status int
if response != nil {
status = response.StatusCode
}
if err == nil {
requestId := response.Header.Get("X-Request-Id")
if requestId != "" {
requestId = response.Header.Get("X-Msedge-Ref")
}
if requestId == "" {
requestId = response.Header.Get("Cf-Ray")
}
scopedLogger.Info().
Str("time", now.Format(time.RFC3339)).
Int("status", status).
Str("request_id", requestId).
Str("time_taken", duration.String()).
Msg("HTTP server returned time")
cancel()
results <- now
} else if !errors.Is(err, context.Canceled) {
scopedLogger.Warn().
Str("error", err.Error()).
Int("status", status).
Msg("failed to query HTTP server")
}
}(url)
}
return <-results
}
func queryHttpTime( func queryHttpTime(
ctx context.Context,
url string, url string,
timeout time.Duration, timeout time.Duration,
) (now *time.Time, err error, response *http.Response) { ) (now *time.Time, err error, response *http.Response) {
client := http.Client{ client := http.Client{
Timeout: timeout, Timeout: timeout,
} }
resp, err := client.Head(url) req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
if err != nil {
return nil, err, nil
}
resp, err := client.Do(req)
if err != nil { if err != nil {
return nil, err, nil return nil, err, nil
} }
@ -23,32 +103,3 @@ func queryHttpTime(
} }
return &parsedTime, nil, resp return &parsedTime, nil, resp
} }
func (t *TimeSync) queryAllHttpTime() (now *time.Time) {
for _, url := range t.httpUrls {
now, err, response := queryHttpTime(url, timeSyncTimeout)
var status string
if response != nil {
status = response.Status
}
scopedLogger := t.l.With().
Str("http_url", url).
Str("status", status).
Logger()
if err == nil {
scopedLogger.Info().
Str("time", now.Format(time.RFC3339)).
Msg("HTTP server returned time")
return now
} else {
scopedLogger.Error().
Str("error", err.Error()).
Msg("failed to query HTTP server")
}
}
return nil
}

View File

@ -1,38 +1,61 @@
package timesync package timesync
import ( import (
"math/rand/v2"
"time" "time"
"github.com/beevik/ntp" "github.com/beevik/ntp"
) )
func (t *TimeSync) queryNetworkTime() (now *time.Time) { func (t *TimeSync) queryNetworkTime() (now *time.Time) {
for _, server := range t.ntpServers { chunkSize := 4
now, err, response := queryNtpServer(server, timeSyncTimeout) ntpServers := t.ntpServers
scopedLogger := t.l.With(). // shuffle the ntp servers to avoid always querying the same servers
Str("server", server). rand.Shuffle(len(ntpServers), func(i, j int) { ntpServers[i], ntpServers[j] = ntpServers[j], ntpServers[i] })
Logger()
if err == nil { for i := 0; i < len(ntpServers); i += chunkSize {
scopedLogger.Info(). chunk := ntpServers[i:min(i+chunkSize, len(ntpServers))]
Str("time", now.Format(time.RFC3339)). results := t.queryMultipleNTP(chunk, timeSyncTimeout)
Str("reference", response.ReferenceString()). if results != nil {
Str("rtt", response.RTT.String()). return results
Str("clockOffset", response.ClockOffset.String()).
Uint8("stratum", response.Stratum).
Msg("NTP server returned time")
return now
} else {
scopedLogger.Error().
Str("error", err.Error()).
Msg("failed to query NTP server")
} }
} }
return nil return nil
} }
func (t *TimeSync) queryMultipleNTP(servers []string, timeout time.Duration) (now *time.Time) {
results := make(chan *time.Time, len(servers))
for _, server := range servers {
go func(server string) {
scopedLogger := t.l.With().
Str("server", server).
Logger()
now, err, response := queryNtpServer(server, timeout)
if err == nil {
scopedLogger.Info().
Str("time", now.Format(time.RFC3339)).
Str("reference", response.ReferenceString()).
Str("rtt", response.RTT.String()).
Str("clockOffset", response.ClockOffset.String()).
Uint8("stratum", response.Stratum).
Msg("NTP server returned time")
results <- now
} else {
scopedLogger.Warn().
Str("error", err.Error()).
Msg("failed to query NTP server")
}
}(server)
}
return <-results
}
func queryNtpServer(server string, timeout time.Duration) (now *time.Time, err error, response *ntp.Response) { func queryNtpServer(server string, timeout time.Duration) (now *time.Time, err error, response *ntp.Response) {
resp, err := ntp.QueryWithOptions(server, ntp.QueryOptions{Timeout: timeout}) resp, err := ntp.QueryWithOptions(server, ntp.QueryOptions{Timeout: timeout})
if err != nil { if err != nil {

View File

@ -10,12 +10,24 @@ import (
var ( var (
timeSync *timesync.TimeSync timeSync *timesync.TimeSync
defaultNTPServers = []string{ defaultNTPServers = []string{
"time.cloudflare.com",
"time.apple.com", "time.apple.com",
"time.aws.com",
"time.windows.com",
"time.google.com",
"162.159.200.123", // time.cloudflare.com
"0.pool.ntp.org",
"1.pool.ntp.org",
"2.pool.ntp.org",
"3.pool.ntp.org",
} }
defaultHTTPUrls = []string{ defaultHTTPUrls = []string{
"http://apple.com", "http://www.gstatic.com/generate_204",
"http://cloudflare.com", "http://cp.cloudflare.com/",
"http://edge-http.microsoft.com/captiveportal/generate_204",
// Firefox, Apple, and Microsoft have inconsistent results, so we don't use it
// "http://detectportal.firefox.com/",
// "http://www.apple.com/library/test/success.html",
// "http://www.msftconnecttest.com/connecttest.txt",
} }
builtTimestamp string builtTimestamp string
) )

View File

@ -84,10 +84,9 @@ export default function SettingsNetworkRoute() {
notifications.error("Failed to renew lease: " + resp.error.message); notifications.error("Failed to renew lease: " + resp.error.message);
} else { } else {
notifications.success("DHCP lease renewed"); notifications.success("DHCP lease renewed");
getNetworkState();
} }
}); });
}, [send, getNetworkState]); }, [send]);
useEffect(() => { useEffect(() => {
getNetworkState(); getNetworkState();
@ -320,7 +319,7 @@ export default function SettingsNetworkRoute() {
</GridCard> </GridCard>
)} )}
</div> </div>
<div className="space-y-4"> <div className="space-y-4 hidden">
<SettingsItem <SettingsItem
title="LLDP" title="LLDP"
description="Control which TLVs will be sent over Link Layer Discovery Protocol" description="Control which TLVs will be sent over Link Layer Discovery Protocol"