refactor(measurex): allow to configure timeouts and max-snapshot-size (#645)

This diff lightly refactors the code in measurex to allow a user
to configure all possible timeouts and the max-snapshot-size.

There is currently a little bit of tension between setting timeouts
inside of measurex and the watchdog timeouts inside of netxlite.

This tension has been documented.

Let us repeat the issue also in this commit message. If you are
using a masurex.Measurer configured with very large timeouts and
the underlying netxlite implementation uses shorter whatchdog
timeouts, then you are going to see shorter than expected timeouts.

Ideally, we would like to have just a single timeout but there is
no way to ask the context "hey, can you tell me if you already have
a configured timeout?".

It may be that the right solution is to modify netxlite to have
some sort of root/library object with this configuration.

If that's the case, then a Measurer could be refactored as follows:

- create the underlying netxlite "library"

- initialize the timeouts desired by the Measurer

- create a Dialer, of whatever is needed

- use it

Now this is not possible because netxlite timeouts are internal
static settings rather than attributes of a structure.

Anyway, for now I'm happy with this just being documented.

(I suspect this issue will need to be addresses when we'll write
unit tests for measurex; at that time a proper solution should
come out naturally due to the unit tests constraints.)

I'm working on this refactoring, BTW, to facilitate rewriting `tor`
using measurex (see https://github.com/ooni/probe/issues/1688).

internal/measurex/http.go

@@ -37,21 +37,43 @@ import (
 // HTTP events into the WritableDB.
 func (mx *Measurer) WrapHTTPTransport(
 	db WritableDB, txp model.HTTPTransport) *HTTPTransportDB {
-	return WrapHTTPTransport(mx.Begin, db, txp)
+	return WrapHTTPTransport(mx.Begin, db, txp, mx.httpMaxBodySnapshotSize())
 }
 
-// We only read a small snapshot of the body to keep measurements
+// DefaultHTTPMaxBodySnapshotSize is the default size used when
-// lean, since we're mostly interested in TLS interference nowadays
+// saving HTTP body snapshots. We only save a small snapshot of the
-// but we'll also allow for reading more bytes from the conn.
+// body to keep measurements lean, since we're mostly interested
-const httpMaxBodySnapshot = 1 << 11
+// in TLS interference nowadays and much less in full bodies.
+const DefaultHTTPMaxBodySnapshotSize = 1 << 11
 
+// httpMaxBodySnapshotSize selects the maximum body snapshot size.
+func (mx *Measurer) httpMaxBodySnapshotSize() int64 {
+	if mx.HTTPMaxBodySnapshotSize > 0 {
+		return mx.HTTPMaxBodySnapshotSize
+	}
+	return DefaultHTTPMaxBodySnapshotSize
+}
+
+// WrapHTTPTransport creates a new model.HTTPTransport instance
+// using the following configuration:
+//
+// - begin is the conventional "zero time" indicating the
+// moment when the measurement begun;
+//
+// - db is the writable DB into which to write the measurement;
+//
+// - txp is the underlying transport to use;
+//
+// - maxBodySnapshotSize is the max size of the response body snapshot
+// to save: we'll truncate bodies larger than that.
 func WrapHTTPTransport(
-	begin time.Time, db WritableDB, txp model.HTTPTransport) *HTTPTransportDB {
+	begin time.Time, db WritableDB, txp model.HTTPTransport,
+	maxBodySnapshotSize int64) *HTTPTransportDB {
 	return &HTTPTransportDB{
 		HTTPTransport:       txp,
 		Begin:               begin,
 		DB:                  db,
-		MaxBodySnapshotSize: httpMaxBodySnapshot,
+		MaxBodySnapshotSize: maxBodySnapshotSize,
 	}
 }
 

internal/measurex/measurer.go

@@ -31,9 +31,30 @@ type Measurer struct {
 	// Begin is when we started measuring (this field is MANDATORY).
 	Begin time.Time
 
+	// DNSLookupTimeout is the OPTIONAL timeout for performing
+	// a DNS lookup. If not set, we use a default value.
+	//
+	// Note that the underlying network implementation MAY use a
+	// shorter-than-you-selected watchdog timeout. In such a case,
+	// the shorter watchdog timeout will prevail.
+	DNSLookupTimeout time.Duration
+
 	// HTTPClient is the MANDATORY HTTP client for the WCTH.
 	HTTPClient model.HTTPClient
 
+	// HTTPMaxBodySnapshotSize is the OPTIONAL maximum size,
+	// in bytes, of the response body snapshot we save. If this field
+	// is zero or negative, we'll use a small default value.
+	HTTPMaxBodySnapshotSize int64
+
+	// HTTPRoundTripTimeout is the OPTIONAL timeout for performing
+	// an HTTP round trip. If not set, we use a default value.
+	//
+	// Note that the underlying network implementation MAY use a
+	// shorter-than-you-selected watchdog timeout. In such a case,
+	// the shorter watchdog timeout will prevail.
+	HTTPRoundTripTimeout time.Duration
+
 	// Logger is the MANDATORY logger to use.
 	Logger model.Logger
 
@@ -42,9 +63,33 @@ type Measurer struct {
 	// is not set, we'll not be using any helper.
 	MeasureURLHelper MeasureURLHelper
 
+	// QUICHandshakeTimeout is the OPTIONAL timeout for performing
+	// a QUIC handshake. If not set, we use a default value.
+	//
+	// Note that the underlying network implementation MAY use a
+	// shorter-than-you-selected watchdog timeout. In such a case,
+	// the shorter watchdog timeout will prevail.
+	QUICHandshakeTimeout time.Duration
+
 	// Resolvers is the MANDATORY list of resolvers.
 	Resolvers []*ResolverInfo
 
+	// TCPConnectTimeout is the OPTIONAL timeout for performing
+	// a tcp connect. If not set, we use a default value.
+	//
+	// Note that the underlying network implementation MAY use a
+	// shorter-than-you-selected watchdog timeout. In such a case,
+	// the shorter watchdog timeout will prevail.
+	TCPconnectTimeout time.Duration
+
+	// TLSHandshakeTimeout is the OPTIONAL timeout for performing
+	// a tls handshake. If not set, we use a default value.
+	//
+	// Note that the underlying network implementation MAY use a
+	// shorter-than-you-selected watchdog timeout. In such a case,
+	// the shorter watchdog timeout will prevail.
+	TLSHandshakeTimeout time.Duration
+
 	// TLSHandshaker is the MANDATORY TLS handshaker.
 	TLSHandshaker model.TLSHandshaker
 }
@@ -53,9 +98,14 @@ type Measurer struct {
 // instance using the most default settings.
 func NewMeasurerWithDefaultSettings() *Measurer {
 	return &Measurer{
-		Begin:      time.Now(),
+		Begin:                   time.Now(),
-		HTTPClient: &http.Client{},
+		DNSLookupTimeout:        0,
-		Logger:     log.Log,
+		HTTPClient:              &http.Client{},
+		HTTPMaxBodySnapshotSize: 0,
+		HTTPRoundTripTimeout:    0,
+		Logger:                  log.Log,
+		MeasureURLHelper:        nil,
+		QUICHandshakeTimeout:    0,
 		Resolvers: []*ResolverInfo{{
 			Network: "system",
 			Address: "",
@@ -63,13 +113,26 @@ func NewMeasurerWithDefaultSettings() *Measurer {
 			Network: "udp",
 			Address: "8.8.4.4:53",
 		}},
-		TLSHandshaker: netxlite.NewTLSHandshakerStdlib(log.Log),
+		TCPconnectTimeout:   0,
+		TLSHandshakeTimeout: 0,
+		TLSHandshaker:       netxlite.NewTLSHandshakerStdlib(log.Log),
 	}
 }
 
+// DefaultDNSLookupTimeout is the default DNS lookup timeout.
+const DefaultDNSLookupTimeout = 4 * time.Second
+
+// dnsLookupTimeout selects the correct DNS lookup timeout.
+func (mx *Measurer) dnsLookupTimeout() time.Duration {
+	if mx.DNSLookupTimeout > 0 {
+		return mx.DNSLookupTimeout
+	}
+	return DefaultDNSLookupTimeout
+}
+
 // LookupHostSystem performs a LookupHost using the system resolver.
 func (mx *Measurer) LookupHostSystem(ctx context.Context, domain string) *DNSMeasurement {
-	const timeout = 4 * time.Second
+	timeout := mx.dnsLookupTimeout()
 	ol := NewOperationLogger(mx.Logger, "LookupHost %s with getaddrinfo", domain)
 	ctx, cancel := context.WithTimeout(ctx, timeout)
 	defer cancel()
@@ -87,7 +150,7 @@ func (mx *Measurer) LookupHostSystem(ctx context.Context, domain string) *DNSMea
 // lookupHostForeign performs a LookupHost using a "foreign" resolver.
 func (mx *Measurer) lookupHostForeign(
 	ctx context.Context, domain string, r model.Resolver) *DNSMeasurement {
-	const timeout = 4 * time.Second
+	timeout := mx.dnsLookupTimeout()
 	ol := NewOperationLogger(mx.Logger, "LookupHost %s with %s", domain, r.Network())
 	ctx, cancel := context.WithTimeout(ctx, timeout)
 	defer cancel()
@@ -113,7 +176,7 @@ func (mx *Measurer) lookupHostForeign(
 // Returns a DNSMeasurement.
 func (mx *Measurer) LookupHostUDP(
 	ctx context.Context, domain, address string) *DNSMeasurement {
-	const timeout = 4 * time.Second
+	timeout := mx.dnsLookupTimeout()
 	ol := NewOperationLogger(mx.Logger, "LookupHost %s with %s/udp", domain, address)
 	ctx, cancel := context.WithTimeout(ctx, timeout)
 	defer cancel()
@@ -141,7 +204,7 @@ func (mx *Measurer) LookupHostUDP(
 // Returns a DNSMeasurement.
 func (mx *Measurer) LookupHTTPSSvcUDP(
 	ctx context.Context, domain, address string) *DNSMeasurement {
-	const timeout = 4 * time.Second
+	timeout := mx.dnsLookupTimeout()
 	ol := NewOperationLogger(mx.Logger, "LookupHTTPSvc %s with %s/udp", domain, address)
 	ctx, cancel := context.WithTimeout(ctx, timeout)
 	defer cancel()
@@ -160,7 +223,7 @@ func (mx *Measurer) LookupHTTPSSvcUDP(
 // except that it uses a "foreign" resolver.
 func (mx *Measurer) lookupHTTPSSvcUDPForeign(
 	ctx context.Context, domain string, r model.Resolver) *DNSMeasurement {
-	const timeout = 4 * time.Second
+	timeout := mx.dnsLookupTimeout()
 	ol := NewOperationLogger(mx.Logger, "LookupHTTPSvc %s with %s", domain, r.Address())
 	ctx, cancel := context.WithTimeout(ctx, timeout)
 	defer cancel()
@@ -196,10 +259,21 @@ func (mx *Measurer) TCPConnect(ctx context.Context, address string) *EndpointMea
 	}
 }
 
+// DefaultTCPConnectTimeout is the default TCP connect timeout.
+const DefaultTCPConnectTimeout = 15 * time.Second
+
+// tcpConnectTimeout selects the correct TCP connect timeout.
+func (mx *Measurer) tcpConnectTimeout() time.Duration {
+	if mx.TCPconnectTimeout > 0 {
+		return mx.TCPconnectTimeout
+	}
+	return DefaultTCPConnectTimeout
+}
+
 // TCPConnectWithDB is like TCPConnect but does not create a new measurement,
 // rather it just stores the events inside of the given DB.
 func (mx *Measurer) TCPConnectWithDB(ctx context.Context, db WritableDB, address string) (Conn, error) {
-	const timeout = 10 * time.Second
+	timeout := mx.tcpConnectTimeout()
 	ol := NewOperationLogger(mx.Logger, "TCPConnect %s", address)
 	ctx, cancel := context.WithTimeout(ctx, timeout)
 	defer cancel()
@@ -255,6 +329,17 @@ func (mx *Measurer) TLSConnectAndHandshake(ctx context.Context,
 	}
 }
 
+// DefaultTLSHandshakeTimeout is the default TLS handshake timeout.
+const DefaultTLSHandshakeTimeout = 10 * time.Second
+
+// tlsHandshakeTimeout selects the correct TLS handshake timeout.
+func (mx *Measurer) tlsHandshakeTimeout() time.Duration {
+	if mx.TLSHandshakeTimeout > 0 {
+		return mx.TLSHandshakeTimeout
+	}
+	return DefaultTLSHandshakeTimeout
+}
+
 // TLSConnectAndHandshakeWithDB is like TLSConnectAndHandshake but
 // uses the given DB instead of creating a new Measurement.
 func (mx *Measurer) TLSConnectAndHandshakeWithDB(ctx context.Context,
@@ -263,7 +348,7 @@ func (mx *Measurer) TLSConnectAndHandshakeWithDB(ctx context.Context,
 	if err != nil {
 		return nil, err
 	}
-	const timeout = 10 * time.Second
+	timeout := mx.tlsHandshakeTimeout()
 	ol := NewOperationLogger(mx.Logger,
 		"TLSHandshake %s with sni=%s", address, config.ServerName)
 	ctx, cancel := context.WithTimeout(ctx, timeout)
@@ -315,12 +400,23 @@ func (mx *Measurer) QUICHandshake(ctx context.Context, address string,
 	}
 }
 
+// DefaultQUICHandshakeTimeout is the default QUIC handshake timeout.
+const DefaultQUICHandshakeTimeout = 10 * time.Second
+
+// quicHandshakeTimeout selects the correct QUIC handshake timeout.
+func (mx *Measurer) quicHandshakeTimeout() time.Duration {
+	if mx.QUICHandshakeTimeout > 0 {
+		return mx.QUICHandshakeTimeout
+	}
+	return DefaultQUICHandshakeTimeout
+}
+
 // QUICHandshakeWithDB is like QUICHandshake but uses the given
 // db to store events rather than creating a temporary one and
 // use it to generate a new Measuremet.
 func (mx *Measurer) QUICHandshakeWithDB(ctx context.Context, db WritableDB,
 	address string, config *tls.Config) (quic.EarlySession, error) {
-	const timeout = 10 * time.Second
+	timeout := mx.quicHandshakeTimeout()
 	ol := NewOperationLogger(mx.Logger,
 		"QUICHandshake %s with sni=%s", address, config.ServerName)
 	ctx, cancel := context.WithTimeout(ctx, timeout)
@@ -495,6 +591,8 @@ func (mx *Measurer) httpEndpointGetQUIC(ctx context.Context,
 	return mx.httpClientDo(ctx, clnt, epnt)
 }
 
+// HTTPClientGET performs a GET operation of the given URL
+// using the given HTTP client instance.
 func (mx *Measurer) HTTPClientGET(
 	ctx context.Context, clnt model.HTTPClient, URL *url.URL) (*http.Response, error) {
 	return mx.httpClientDo(ctx, clnt, &HTTPEndpoint{
@@ -508,6 +606,17 @@ func (mx *Measurer) HTTPClientGET(
 	})
 }
 
+// DefaultHTTPRoundTripTimeout is the default HTTP round-trip timeout.
+const DefaultHTTPRoundTripTimeout = 15 * time.Second
+
+// httpRoundTripTimeout selects the correct HTTP round-trip timeout.
+func (mx *Measurer) httpRoundTripTimeout() time.Duration {
+	if mx.HTTPRoundTripTimeout > 0 {
+		return mx.HTTPRoundTripTimeout
+	}
+	return DefaultHTTPRoundTripTimeout
+}
+
 func (mx *Measurer) httpClientDo(ctx context.Context,
 	clnt model.HTTPClient, epnt *HTTPEndpoint) (*http.Response, error) {
 	req, err := NewHTTPGetRequest(ctx, epnt.URL.String())
@@ -515,7 +624,7 @@ func (mx *Measurer) httpClientDo(ctx context.Context,
 		return nil, err
 	}
 	req.Header = epnt.Header.Clone() // must clone because of parallel usage
-	const timeout = 15 * time.Second
+	timeout := mx.httpRoundTripTimeout()
 	ol := NewOperationLogger(mx.Logger,
 		"%s %s with %s/%s", req.Method, req.URL.String(), epnt.Address, epnt.Network)
 	ctx, cancel := context.WithTimeout(ctx, timeout)

internal/measurex/tracing.go

@@ -26,13 +26,18 @@ import (
 // - resolver is the underlying resolver to use
 //
 // - handshake is the TLS handshaker to use
+//
+// - maxBodySnapshotSize is the max size of the response body snapshot
+// to save: we'll truncate bodies larger than that.
 func NewTracingHTTPTransport(logger model.Logger, begin time.Time, db WritableDB,
-	resolver model.Resolver, dialer model.Dialer, handshaker model.TLSHandshaker) *HTTPTransportDB {
+	resolver model.Resolver, dialer model.Dialer, handshaker model.TLSHandshaker,
+	maxBodySnapshotSize int64) *HTTPTransportDB {
 	resolver = WrapResolver(begin, db, resolver)
 	dialer = netxlite.WrapDialer(logger, resolver, WrapDialer(begin, db, dialer))
 	tlsDialer := netxlite.NewTLSDialer(dialer, handshaker)
 	return WrapHTTPTransport(
-		begin, db, netxlite.NewHTTPTransport(logger, dialer, tlsDialer))
+		begin, db, netxlite.NewHTTPTransport(logger, dialer, tlsDialer),
+		maxBodySnapshotSize)
 }
 
 // NewTracingHTTPTransportWithDefaultSettings creates a new
@@ -46,21 +51,27 @@ func NewTracingHTTPTransport(logger model.Logger, begin time.Time, db WritableDB
 //
 // - db is the DB in which to write events that will
 // eventually become the measurement
-//
 func NewTracingHTTPTransportWithDefaultSettings(
 	begin time.Time, logger model.Logger, db WritableDB) *HTTPTransportDB {
 	return NewTracingHTTPTransport(logger, begin, db,
 		netxlite.NewResolverStdlib(logger),
 		netxlite.NewDialerWithoutResolver(logger),
-		netxlite.NewTLSHandshakerStdlib(logger))
+		netxlite.NewTLSHandshakerStdlib(logger),
+		DefaultHTTPMaxBodySnapshotSize)
 }
 
-func (mx *Measurer) NewTracingHTTPTransportWithDefaultSettings(
+// NewTracingHTTPTransportWithDefaultSettings creates a new
-	logger model.Logger, db WritableDB) *HTTPTransportDB {
+// HTTP transport with tracing capabilities and default settings.
+//
+// Arguments:
+//
+// - db is the DB in which to write events that will
+// eventually become the measurement
+func (mx *Measurer) NewTracingHTTPTransportWithDefaultSettings(db WritableDB) *HTTPTransportDB {
 	return NewTracingHTTPTransport(
 		mx.Logger, mx.Begin, db, mx.NewResolverSystem(db, mx.Logger),
 		mx.NewDialerWithoutResolver(db, mx.Logger),
-		mx.TLSHandshaker)
+		mx.TLSHandshaker, mx.httpMaxBodySnapshotSize())
 }
 
 // UnmeasuredHTTPEndpoints returns the endpoints whose IP address