[WIP] benchmarking: add the egress-tunnels workload

cmd/benchmarking/egress/internal/cmd/actor/classify.go

@@ -0,0 +1,95 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package actor
+
+import (
+	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"net/url"
+	"strings"
+	"syscall"
+)
+
+// maxOtherLen bounds the message kept for an error classify does not
+// recognize, so one odd error cannot flood the report.
+const maxOtherLen = 80
+
+// classify names the outcome of a request in a few low-cardinality classes.
+// It returns "" for a success: no error and HTTP 200.
+func classify(err error, status int) string {
+	if err == nil {
+		if status == 200 {
+			return ""
+		}
+		return fmt.Sprintf("HTTP %d", status)
+	}
+	var netErr net.Error
+	var dnsErr *net.DNSError
+	var opErr *net.OpError
+	switch {
+	case errors.As(err, &dnsErr):
+		if dnsErr.IsNotFound {
+			return "dns: not found"
+		}
+		if dnsErr.IsTimeout {
+			return "dns: timeout"
+		}
+		return "dns: error"
+	case errors.As(err, new(x509.UnknownAuthorityError)):
+		return "tls: unknown authority"
+	case errors.As(err, new(x509.HostnameError)):
+		return "tls: hostname mismatch"
+	case errors.As(err, new(x509.CertificateInvalidError)):
+		return "tls: invalid certificate"
+	case errors.As(err, new(tls.RecordHeaderError)),
+		// net/http replaces the record error with this when the server
+		// answered in plain HTTP, without wrapping it.
+		strings.Contains(err.Error(), "server gave HTTP response to HTTPS client"):
+		return "tls: not a TLS server"
+	case errors.As(err, &opErr) && opErr.Op == "remote error":
+		// crypto/tls reports an alert from the peer this way; its message,
+		// such as "tls: handshake failure", names the alert.
+		return "tls: peer alert: " + strings.TrimPrefix(opErr.Err.Error(), "tls: ")
+	case errors.Is(err, context.DeadlineExceeded), errors.As(err, &netErr) && netErr.Timeout():
+		return "timeout"
+	case errors.Is(err, syscall.ECONNREFUSED):
+		return "connection refused"
+	case errors.Is(err, syscall.ECONNRESET):
+		return "connection reset"
+	case errors.Is(err, io.EOF), errors.Is(err, io.ErrUnexpectedEOF):
+		return "EOF"
+	}
+	// *url.Error repeats the method and URL, and *net.OpError the addresses
+	// and the local port; both only add cardinality.
+	var urlErr *url.Error
+	if errors.As(err, &urlErr) {
+		err = urlErr.Err
+	}
+	if errors.As(err, &opErr) {
+		return "net: " + opErr.Op + ": " + opErr.Err.Error()
+	}
+	msg := err.Error()
+	if len(msg) > maxOtherLen {
+		// Cut on a character boundary, or one message could yield two
+		// classes depending on where its bytes fall.
+		msg = strings.ToValidUTF8(msg[:maxOtherLen], "")
+	}
+	return "other: " + msg
+}

cmd/benchmarking/egress/internal/cmd/actor/classify_test.go

@@ -0,0 +1,90 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package actor
+
+import (
+	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"errors"
+	"io"
+	"net"
+	"net/url"
+	"strings"
+	"syscall"
+	"testing"
+)
+
+type timeoutErr struct{}
+
+func (timeoutErr) Error() string   { return "i/o timeout" }
+func (timeoutErr) Timeout() bool   { return true }
+func (timeoutErr) Temporary() bool { return true }
+
+func TestClassify(t *testing.T) {
+	wrap := func(err error) error { return &url.Error{Op: "Get", URL: "http://x/", Err: err} }
+	for _, tc := range []struct {
+		name   string
+		err    error
+		status int
+		want   string
+	}{
+		{"success", nil, 200, ""},
+		{"http status", nil, 503, "HTTP 503"},
+		{"deadline", wrap(context.DeadlineExceeded), 0, "timeout"},
+		{"net timeout", wrap(&net.OpError{Op: "dial", Err: timeoutErr{}}), 0, "timeout"},
+		{"refused", wrap(&net.OpError{Op: "dial", Err: syscall.ECONNREFUSED}), 0, "connection refused"},
+		{"reset", wrap(&net.OpError{Op: "read", Err: syscall.ECONNRESET}), 0, "connection reset"},
+		{"eof", wrap(io.EOF), 0, "EOF"},
+		{"unexpected eof", wrap(io.ErrUnexpectedEOF), 0, "EOF"},
+		{"dns not found", wrap(&net.DNSError{Err: "no such host", Name: "x", IsNotFound: true}), 0, "dns: not found"},
+		{"dns other", wrap(&net.DNSError{Err: "server misbehaving", Name: "x"}), 0, "dns: error"},
+		{"dns timeout", wrap(&net.DNSError{Err: "i/o timeout", Name: "x", IsTimeout: true}), 0, "dns: timeout"},
+		{"long other is cut", wrap(errors.New(strings.Repeat("x", maxOtherLen+20))), 0, "other: " + strings.Repeat("x", maxOtherLen)},
+		{"cut keeps whole characters", wrap(errors.New(strings.Repeat("a", maxOtherLen-1) + "é!")), 0, "other: " + strings.Repeat("a", maxOtherLen-1)},
+		{"other strips url", wrap(errors.New("boom")), 0, "other: boom"},
+		{"unknown authority", wrap(&tls.CertificateVerificationError{Err: x509.UnknownAuthorityError{}}), 0, "tls: unknown authority"},
+		{"hostname mismatch", wrap(&tls.CertificateVerificationError{Err: x509.HostnameError{Host: "x"}}), 0, "tls: hostname mismatch"},
+		{"expired", wrap(&tls.CertificateVerificationError{Err: x509.CertificateInvalidError{Reason: x509.Expired}}), 0, "tls: invalid certificate"},
+		{"not tls", wrap(tls.RecordHeaderError{Msg: "first record does not look like a TLS handshake"}), 0, "tls: not a TLS server"},
+		{"plain http answer", wrap(errors.New("http: server gave HTTP response to HTTPS client")), 0, "tls: not a TLS server"},
+		{"peer alert", wrap(&net.OpError{Op: "remote error", Err: errors.New("tls: handshake failure")}), 0, "tls: peer alert: handshake failure"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := classify(tc.err, tc.status); got != tc.want {
+				t.Errorf("classify() = %q, want %q", got, tc.want)
+			}
+		})
+	}
+
+	long := classify(errors.New(strings.Repeat("x", 500)), 0)
+	if len(long) != len("other: ")+maxOtherLen {
+		t.Errorf("unrecognized error not truncated: %d bytes", len(long))
+	}
+}
+
+func TestClassifyDropsAddressesFromNetErrors(t *testing.T) {
+	t.Parallel()
+	err := &url.Error{Op: "Get", URL: "http://10.0.0.1:8080/", Err: &net.OpError{
+		Op: "write", Net: "tcp",
+		Source: &net.TCPAddr{IP: net.IPv4(10, 0, 0, 2), Port: 51234},
+		Addr:   &net.TCPAddr{IP: net.IPv4(10, 0, 0, 1), Port: 8080},
+		Err:    syscall.EPIPE,
+	}}
+	got := classify(err, 0)
+	if want := "net: write: broken pipe"; got != want {
+		t.Errorf("classify(%v) = %q, want %q", err, got, want)
+	}
+}

cmd/benchmarking/egress/internal/cmd/actor/recorder.go

@@ -0,0 +1,94 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package actor is the workload of the egress scale test: the subcommand
+// that holds one tunnel per endpoint open through the egress gateway and
+// counts what happened.
+package actor
+
+import (
+	"context"
+	"fmt"
+
+	"go.opentelemetry.io/otel/metric"
+	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+	"go.opentelemetry.io/otel/sdk/resource"
+
+	"go.ate.dev/substrate/cmd/benchmarking/egress/internal/contract"
+)
+
+// Recorder holds one loop's four distributions as OTel histograms on a
+// manual reader in delta temporality: Collect returns what was recorded since
+// the previous Collect, so each call is one window. Nothing exports; the
+// points travel inside the loop's Stats.
+type Recorder struct {
+	reader   *sdkmetric.ManualReader
+	provider *sdkmetric.MeterProvider
+	hists    map[contract.Distribution]metric.Float64Histogram
+}
+
+// NewRecorder creates the four instruments on a private MeterProvider.
+func NewRecorder() (*Recorder, error) {
+	reader := sdkmetric.NewManualReader(sdkmetric.WithTemporalitySelector(
+		func(sdkmetric.InstrumentKind) metricdata.Temporality { return metricdata.DeltaTemporality }))
+	provider := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader), sdkmetric.WithResource(resource.Empty()))
+	meter := provider.Meter("benchmark-egress")
+	r := &Recorder{reader: reader, provider: provider, hists: make(map[contract.Distribution]metric.Float64Histogram, len(contract.DistributionNames))}
+	for _, name := range contract.DistributionNames {
+		h, err := meter.Float64Histogram(string(name), metric.WithUnit("s"), metric.WithExplicitBucketBoundaries(contract.Bounds...))
+		if err != nil {
+			_ = provider.Shutdown(context.Background())
+			return nil, fmt.Errorf("creating the %s histogram: %w", name, err)
+		}
+		r.hists[name] = h
+	}
+	return r, nil
+}
+
+// Record adds one sample, in seconds, to the named distribution.
+func (r *Recorder) Record(ctx context.Context, name contract.Distribution, seconds float64) {
+	r.hists[name].Record(ctx, seconds)
+}
+
+// Collect returns each distribution's samples since the previous Collect,
+// keyed by name. A distribution with no new sample is absent.
+func (r *Recorder) Collect(ctx context.Context) (map[contract.Distribution]contract.Point, error) {
+	var rm metricdata.ResourceMetrics
+	if err := r.reader.Collect(ctx, &rm); err != nil {
+		return nil, err
+	}
+	out := make(map[contract.Distribution]contract.Point, len(contract.DistributionNames))
+	for _, sm := range rm.ScopeMetrics {
+		for _, m := range sm.Metrics {
+			h, ok := m.Data.(metricdata.Histogram[float64])
+			if !ok || len(h.DataPoints) == 0 {
+				continue
+			}
+			out[contract.Distribution(m.Name)] = pointOf(h.DataPoints[0])
+		}
+	}
+	return out, nil
+}
+
+// Shutdown releases the provider. Record is a no-op afterwards.
+func (r *Recorder) Shutdown(ctx context.Context) error { return r.provider.Shutdown(ctx) }
+
+// pointOf copies the merge fields out of an SDK data point.
+func pointOf(dp metricdata.HistogramDataPoint[float64]) contract.Point {
+	p := contract.Point{Bounds: dp.Bounds, BucketCounts: dp.BucketCounts, Count: dp.Count, Sum: dp.Sum}
+	p.Min, _ = dp.Min.Value()
+	p.Max, _ = dp.Max.Value()
+	return p
+}