Test User
294 lines
8.7 KiB
Go
294 lines
8.7 KiB
Go
package client
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"math"
|
|
"net/http"
|
|
"net/url"
|
|
"os/exec"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"speedtest/internal/probe"
|
|
)
|
|
|
|
type Iperf3Options struct {
|
|
ServerURL string
|
|
Target string
|
|
Duration time.Duration
|
|
Timeout time.Duration
|
|
Protocol string
|
|
Reverse bool
|
|
BandwidthMbps float64
|
|
Iperf3Path string
|
|
OnSample func(probe.Sample)
|
|
}
|
|
|
|
type createIperf3Request struct {
|
|
DurationSeconds int `json:"durationSeconds"`
|
|
Protocol string `json:"protocol"`
|
|
Reverse bool `json:"reverse"`
|
|
BandwidthMbps float64 `json:"bandwidthMbps,omitempty"`
|
|
}
|
|
|
|
type createIperf3Response struct {
|
|
ID string `json:"id"`
|
|
Host string `json:"host"`
|
|
Port int `json:"port"`
|
|
Protocol string `json:"protocol"`
|
|
Reverse bool `json:"reverse"`
|
|
DurationSeconds int `json:"durationSeconds"`
|
|
BandwidthMbps float64 `json:"bandwidthMbps"`
|
|
CompleteURL string `json:"completeUrl"`
|
|
}
|
|
|
|
func RunIperf3(ctx context.Context, options Iperf3Options) (probe.Summary, error) {
|
|
options, server, err := normalizeIperf3Options(options)
|
|
if err != nil {
|
|
return probe.Summary{}, err
|
|
}
|
|
httpClient := http.Client{Timeout: transferTimeout(options.Timeout)}
|
|
|
|
session, err := createIperf3Session(ctx, &httpClient, server, options)
|
|
if err != nil {
|
|
return probe.Summary{}, err
|
|
}
|
|
|
|
startedAt := time.Now().UTC()
|
|
raw, err := runIperf3Command(ctx, options, session, server)
|
|
finishedAt := time.Now().UTC()
|
|
if err != nil {
|
|
failedSample := probe.Sample{
|
|
At: finishedAt,
|
|
Kind: iperf3SampleKind(session.Reverse),
|
|
Success: false,
|
|
Error: err.Error(),
|
|
}
|
|
completeCtx, cancel := context.WithTimeout(context.Background(), cancelRequestTimeout(options.Timeout))
|
|
defer cancel()
|
|
if _, completeErr := completeIperf3Session(completeCtx, &httpClient, server, session.CompleteURL, startedAt, finishedAt, []probe.Sample{failedSample}); completeErr != nil {
|
|
return probe.Summary{}, fmt.Errorf("%w; also failed to complete iperf3 session: %v", err, completeErr)
|
|
}
|
|
return probe.Summary{}, err
|
|
}
|
|
sample, err := sampleFromIperf3JSON(raw, session.Reverse)
|
|
if err != nil {
|
|
return probe.Summary{}, err
|
|
}
|
|
sample.At = finishedAt
|
|
if options.OnSample != nil {
|
|
options.OnSample(sample)
|
|
}
|
|
return completeIperf3Session(ctx, &httpClient, server, session.CompleteURL, startedAt, finishedAt, []probe.Sample{sample})
|
|
}
|
|
|
|
func normalizeIperf3Options(options Iperf3Options) (Iperf3Options, *url.URL, error) {
|
|
if strings.TrimSpace(options.ServerURL) == "" {
|
|
return Iperf3Options{}, nil, errors.New("server URL is required")
|
|
}
|
|
server, err := url.Parse(strings.TrimRight(options.ServerURL, "/"))
|
|
if err != nil || server.Scheme == "" || server.Host == "" || (server.Scheme != "http" && server.Scheme != "https") {
|
|
return Iperf3Options{}, nil, errors.New("server URL must be a valid http or https URL")
|
|
}
|
|
if strings.TrimSpace(options.Target) == "" {
|
|
options.Target = server.Host
|
|
}
|
|
if options.Duration <= 0 {
|
|
options.Duration = 30 * time.Second
|
|
}
|
|
if options.Timeout <= 0 {
|
|
options.Timeout = 3 * time.Second
|
|
}
|
|
protocol := strings.ToLower(strings.TrimSpace(options.Protocol))
|
|
if protocol == "" {
|
|
protocol = "tcp"
|
|
}
|
|
if protocol != "tcp" && protocol != "udp" {
|
|
return Iperf3Options{}, nil, errors.New("protocol must be tcp or udp")
|
|
}
|
|
options.Protocol = protocol
|
|
if strings.TrimSpace(options.Iperf3Path) == "" {
|
|
options.Iperf3Path = "iperf3"
|
|
}
|
|
return options, server, nil
|
|
}
|
|
|
|
func createIperf3Session(ctx context.Context, httpClient *http.Client, server *url.URL, options Iperf3Options) (createIperf3Response, error) {
|
|
payload := createIperf3Request{
|
|
DurationSeconds: int(math.Max(1, math.Ceil(options.Duration.Seconds()))),
|
|
Protocol: options.Protocol,
|
|
Reverse: options.Reverse,
|
|
BandwidthMbps: options.BandwidthMbps,
|
|
}
|
|
body, err := json.Marshal(payload)
|
|
if err != nil {
|
|
return createIperf3Response{}, err
|
|
}
|
|
request, err := http.NewRequestWithContext(ctx, http.MethodPost, resolve(server, "/api/iperf3/sessions"), bytes.NewReader(body))
|
|
if err != nil {
|
|
return createIperf3Response{}, err
|
|
}
|
|
request.Header.Set("Content-Type", "application/json")
|
|
response, err := httpClient.Do(request)
|
|
if err != nil {
|
|
return createIperf3Response{}, err
|
|
}
|
|
defer response.Body.Close()
|
|
|
|
if response.StatusCode < 200 || response.StatusCode >= 300 {
|
|
message := decodeError(response.Body)
|
|
if message == "" {
|
|
message = response.Status
|
|
}
|
|
return createIperf3Response{}, fmt.Errorf("create iperf3 session: %s", message)
|
|
}
|
|
var session createIperf3Response
|
|
if err := json.NewDecoder(response.Body).Decode(&session); err != nil {
|
|
return createIperf3Response{}, err
|
|
}
|
|
return session, nil
|
|
}
|
|
|
|
func runIperf3Command(ctx context.Context, options Iperf3Options, session createIperf3Response, server *url.URL) ([]byte, error) {
|
|
timeout := options.Duration + options.Timeout
|
|
commandCtx, cancel := context.WithTimeout(ctx, timeout)
|
|
defer cancel()
|
|
|
|
host := session.Host
|
|
if host == "" {
|
|
host = server.Hostname()
|
|
}
|
|
args := []string{"-c", host, "-p", strconv.Itoa(session.Port), "-t", strconv.Itoa(session.DurationSeconds), "-J"}
|
|
if session.Protocol == "udp" {
|
|
args = append(args, "-u")
|
|
}
|
|
if session.Reverse {
|
|
args = append(args, "-R")
|
|
}
|
|
if session.BandwidthMbps > 0 {
|
|
args = append(args, "-b", formatMbps(session.BandwidthMbps)+"M")
|
|
}
|
|
|
|
cmd := exec.CommandContext(commandCtx, options.Iperf3Path, args...)
|
|
var stderr bytes.Buffer
|
|
cmd.Stderr = &stderr
|
|
output, err := cmd.Output()
|
|
if err != nil {
|
|
message := strings.TrimSpace(stderr.String())
|
|
if message == "" {
|
|
message = strings.TrimSpace(string(output))
|
|
}
|
|
if message == "" {
|
|
message = err.Error()
|
|
}
|
|
return nil, fmt.Errorf("iperf3 failed: %s", message)
|
|
}
|
|
return output, nil
|
|
}
|
|
|
|
func iperf3SampleKind(reverse bool) string {
|
|
if reverse {
|
|
return "download"
|
|
}
|
|
return "upload"
|
|
}
|
|
|
|
func sampleFromIperf3JSON(raw []byte, reverse bool) (probe.Sample, error) {
|
|
type sum struct {
|
|
Bytes int64 `json:"bytes"`
|
|
BitsPerSec float64 `json:"bits_per_second"`
|
|
LostPercent float64 `json:"lost_percent"`
|
|
JitterMs float64 `json:"jitter_ms"`
|
|
Seconds float64 `json:"seconds"`
|
|
Retransmits int64 `json:"retransmits"`
|
|
Omitted bool `json:"omitted"`
|
|
Sender bool `json:"sender"`
|
|
Receiver bool `json:"receiver"`
|
|
LostPackets int64 `json:"lost_packets"`
|
|
TotalPackets int64 `json:"packets"`
|
|
Datagrams int64 `json:"datagrams"`
|
|
OutOfOrder int64 `json:"out_of_order"`
|
|
PacketLatency float64 `json:"mean_rtt"`
|
|
}
|
|
type iperfJSON struct {
|
|
End struct {
|
|
SumSent sum `json:"sum_sent"`
|
|
SumReceived sum `json:"sum_received"`
|
|
Sum sum `json:"sum"`
|
|
} `json:"end"`
|
|
}
|
|
var payload iperfJSON
|
|
if err := json.Unmarshal(raw, &payload); err != nil {
|
|
return probe.Sample{}, err
|
|
}
|
|
|
|
selected := payload.End.SumSent
|
|
kind := "upload"
|
|
if reverse {
|
|
selected = payload.End.SumReceived
|
|
kind = "download"
|
|
}
|
|
if selected.BitsPerSec == 0 {
|
|
selected = payload.End.Sum
|
|
}
|
|
if selected.BitsPerSec == 0 && payload.End.SumReceived.BitsPerSec > 0 {
|
|
selected = payload.End.SumReceived
|
|
}
|
|
if selected.BitsPerSec == 0 {
|
|
return probe.Sample{}, errors.New("iperf3 JSON does not include throughput")
|
|
}
|
|
|
|
return probe.Sample{
|
|
Kind: kind,
|
|
Success: true,
|
|
Bytes: selected.Bytes,
|
|
Mbps: round2(selected.BitsPerSec / 1000000),
|
|
}, nil
|
|
}
|
|
|
|
func completeIperf3Session(ctx context.Context, httpClient *http.Client, server *url.URL, path string, startedAt time.Time, finishedAt time.Time, samples []probe.Sample) (probe.Summary, error) {
|
|
payload := map[string]any{
|
|
"startedAt": startedAt,
|
|
"finishedAt": finishedAt,
|
|
"samples": samples,
|
|
}
|
|
body, err := json.Marshal(payload)
|
|
if err != nil {
|
|
return probe.Summary{}, err
|
|
}
|
|
request, err := http.NewRequestWithContext(ctx, http.MethodPost, resolve(server, path), bytes.NewReader(body))
|
|
if err != nil {
|
|
return probe.Summary{}, err
|
|
}
|
|
request.Header.Set("Content-Type", "application/json")
|
|
response, err := httpClient.Do(request)
|
|
if err != nil {
|
|
return probe.Summary{}, err
|
|
}
|
|
defer response.Body.Close()
|
|
if response.StatusCode < 200 || response.StatusCode >= 300 {
|
|
message := decodeError(response.Body)
|
|
if message == "" {
|
|
message = response.Status
|
|
}
|
|
return probe.Summary{}, fmt.Errorf("complete iperf3 session: %s", message)
|
|
}
|
|
var completed completeResponse
|
|
if err := json.NewDecoder(response.Body).Decode(&completed); err != nil {
|
|
return probe.Summary{}, err
|
|
}
|
|
return completed.Summary, nil
|
|
}
|
|
|
|
func formatMbps(mbps float64) string {
|
|
if mbps == math.Trunc(mbps) {
|
|
return strconv.FormatInt(int64(mbps), 10)
|
|
}
|
|
return strings.TrimRight(strings.TrimRight(fmt.Sprintf("%.2f", mbps), "0"), ".")
|
|
}
|