Something went wrong. Try again.
forked from tailscale/tailscale
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199// Copyright (c) Tailscale Inc & contributors// SPDX-License-Identifier: BSD-3-Clause
package prober
import ( "bytes" "context" "expvar" "fmt" "io" "net" "net/http" "net/netip" "time"
"github.com/prometheus/client_golang/prometheus" "tailscale.com/net/netutil")
const maxHTTPBody = 4 << 20 // MiB
// NewProbeTransport returns a fresh *http.Transport for a single probe run (so// we never reuse a past connection).//// If dialAddr is valid (its zero value means "no override"), every connection is// dialed to dialAddr instead of resolving the request URL's host, while SNI, the// Host header, and TLS certificate validation continue to derive from the URL// host. This is the HTTP analog of TLSWithIP: it lets a probe target a specific// backend that serves a given hostname (e.g. one particular Funnel ingress node).//// Custom probe classes that dial a specific backend should use this rather than// reconstructing the dial override, so the SNI/Host/cert semantics stay// identical across probes.func NewProbeTransport(dialAddr netip.AddrPort) *http.Transport { tr := netutil.NewDefaultTransport() if dialAddr.IsValid() { dst := dialAddr.String() // Reuse the transport's own dialer (preserving its Timeout/KeepAlive and // any future tuning); only substitute the dial target so connections go // to dialAddr instead of the resolved URL host. dial := tr.DialContext tr.DialContext = func(ctx context.Context, network, _ string) (net.Conn, error) { return dial(ctx, network, dst) } } return tr}
// HTTP returns a ProbeClass that healthchecks an HTTP URL.//// The probe function sends a GET request for url, expects an HTTP 200// response, and verifies that want is present in the response// body.func HTTP(url, wantText string) ProbeClass { return httpProbe(url, netip.AddrPort{}, wantText)}
// HTTPWithDialAddr is like HTTP, but dials dialAddr (an ip:port) instead of the// URL's host. SNI, the Host header, and TLS certificate validation still use the// URL host, so this probes a specific backend serving the URL's hostname.func HTTPWithDialAddr(url string, dialAddr netip.AddrPort, wantText string) ProbeClass { return httpProbe(url, dialAddr, wantText)}
func httpProbe(url string, dialAddr netip.AddrPort, wantText string) ProbeClass { return ProbeClass{ Probe: func(ctx context.Context) error { return probeHTTP(ctx, url, []byte(wantText), dialAddr) }, Class: "http", }}
func probeHTTP(ctx context.Context, url string, want []byte, dialAddr netip.AddrPort) error { req, err := http.NewRequestWithContext(ctx, "GET", url, nil) if err != nil { return fmt.Errorf("constructing request: %w", err) }
// Get a completely new transport each time, so we don't reuse a // past connection. tr := NewProbeTransport(dialAddr) defer tr.CloseIdleConnections() c := &http.Client{ Transport: tr, }
resp, err := c.Do(req) if err != nil { return fmt.Errorf("fetching %q: %w", url, err) } defer resp.Body.Close() if resp.StatusCode != 200 { return fmt.Errorf("fetching %q: status code %d, want 200", url, resp.StatusCode) }
bs, err := io.ReadAll(io.LimitReader(resp.Body, maxHTTPBody)) if err != nil { return fmt.Errorf("reading body of %q: %w", url, err) }
if !bytes.Contains(bs, want) { // Log response body, but truncate it if it's too large; the limit // has been chosen arbitrarily. if maxlen := 300; len(bs) > maxlen { bs = bs[:maxlen] } return fmt.Errorf("body of %q does not contain %q (got: %q)", url, want, string(bs)) }
return nil}
// HTTPBandwidth returns a ProbeClass that downloads size bytes from url and// records how long the transfer took, for bandwidth measurement. It issues a// GET, expects an HTTP 200 response, and reads size bytes from the body.//// Because the transfer is measured at the receiver (this prober reads and times// the body it pulls), the recorded byte count and duration are exact even on a// truncated response. This probe does not carry a direction label; callers that// run it alongside an upload probe can attach one at registration time (e.g.// Labels{"direction": "down"}).//// size must be positive. A non-positive size reads nothing from the body, so// the probe records a zero-byte transfer and trivially succeeds.func HTTPBandwidth(url string, size int64) ProbeClass { return httpBandwidthProbe(url, size, netip.AddrPort{})}
// HTTPBandwidthWithDialAddr is like HTTPBandwidth, but dials dialAddr (an// ip:port) instead of the URL's host, while SNI/Host/cert validation still use// the URL host. It measures download bandwidth from a specific backend serving// the URL's hostname.func HTTPBandwidthWithDialAddr(url string, size int64, dialAddr netip.AddrPort) ProbeClass { return httpBandwidthProbe(url, size, dialAddr)}
func httpBandwidthProbe(url string, size int64, dialAddr netip.AddrPort) ProbeClass { var transferTimeSeconds expvar.Float var totalBytesTransferred expvar.Float return ProbeClass{ Probe: func(ctx context.Context) error { return probeHTTPBandwidth(ctx, url, size, dialAddr, &transferTimeSeconds, &totalBytesTransferred) }, Class: "http_bw", Metrics: HTTPBandwidthMetrics(size, &transferTimeSeconds, &totalBytesTransferred), }}
// HTTPBandwidthMetrics returns the Metrics function for an "http_bw" bandwidth// probe, exposing the configured payload size and the running transfer// time/bytes accumulators. It is shared so probes that measure bandwidth// differently (e.g. a receiver-reported upload probe) still emit an identical// metric set and can be compared under a common direction label.func HTTPBandwidthMetrics(size int64, transferTimeSeconds, totalBytesTransferred *expvar.Float) func(prometheus.Labels) []prometheus.Metric { return func(lb prometheus.Labels) []prometheus.Metric { return []prometheus.Metric{ prometheus.MustNewConstMetric(prometheus.NewDesc("http_bw_probe_size_bytes", "Payload size of the bandwidth prober", nil, lb), prometheus.GaugeValue, float64(size)), prometheus.MustNewConstMetric(prometheus.NewDesc("http_bw_transfer_time_seconds_total", "Time it took to transfer data", nil, lb), prometheus.CounterValue, transferTimeSeconds.Value()), prometheus.MustNewConstMetric(prometheus.NewDesc("http_bw_bytes_total", "Amount of data transferred", nil, lb), prometheus.CounterValue, totalBytesTransferred.Value()), } }}
func probeHTTPBandwidth(ctx context.Context, url string, size int64, dialAddr netip.AddrPort, transferTimeSeconds, totalBytesTransferred *expvar.Float) error { req, err := http.NewRequestWithContext(ctx, "GET", url, nil) if err != nil { return fmt.Errorf("constructing request: %w", err) }
// Get a completely new transport each time, so we don't reuse a // past connection. tr := NewProbeTransport(dialAddr) defer tr.CloseIdleConnections() c := &http.Client{ Transport: tr, }
resp, err := c.Do(req) if err != nil { return fmt.Errorf("fetching %q: %w", url, err) } defer resp.Body.Close() if resp.StatusCode != 200 { return fmt.Errorf("fetching %q: status code %d, want 200", url, resp.StatusCode) } start := time.Now() n, err := io.CopyN(io.Discard, resp.Body, size) // Measure transfer time and bytes transferred irrespective of whether // it succeeded or failed. transferTimeSeconds.Add(time.Since(start).Seconds()) totalBytesTransferred.Add(float64(n)) if err != nil { return fmt.Errorf("reading body of %q: %w", url, err) } return nil}