Run integration tests against an external agent process (#552)

* Run integration tests against an external agent process

* Fix lint errors

Author: tallclair

.golangci.yaml

@@ -0,0 +1,20 @@
+run:
+  timeout: 3m
+linters:
+  disable-all: true
+  enable:
+  - gofmt
+  - revive
+  - gosec
+  - govet
+  - unused
+issues:
+  fix: true
+  exclude-rules:
+    # Don't run security checks on test files
+    - path: _test\.go
+      linters:
+        - gosec
+    - path: ^tests/
+      linters:
+        - gosec

Makefile

@@ -60,6 +60,10 @@ test:
 	go test -race -covermode=atomic -coverprofile=konnectivity.out ./... && go tool cover -html=konnectivity.out -o=konnectivity.html
 	cd konnectivity-client && go test -race -covermode=atomic -coverprofile=client.out ./... && go tool cover -html=client.out -o=client.html
 
+.PHONY: test-integration
+test-integration: build
+	go test -race ./tests -agent-path $(PWD)/bin/proxy-agent
+
 ## --------------------------------------
 ## Binaries
 ## --------------------------------------
@@ -91,7 +95,7 @@ bin/proxy-server: bin $(SOURCE)
 .PHONY: lint
 lint:
 	curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(INSTALL_LOCATION) v$(GOLANGCI_LINT_VERSION)
-	$(INSTALL_LOCATION)/golangci-lint run --no-config --disable-all --enable=gofmt,revive,gosec,govet,unused --fix --verbose --timeout 3m
+	$(INSTALL_LOCATION)/golangci-lint run --verbose
 
 ## --------------------------------------
 ## Go

cmd/agent/app/server.go

@@ -132,9 +132,9 @@ func (a *Agent) runHealthServer(o *options.GrpcProxyAgentOptions, cs agent.Readi
 
 		// Always be verbose if the check has failed
 		if len(failedChecks) > 0 {
-			klog.V(0).Infoln("%s check failed: \n%v", strings.Join(failedChecks, ","), individualCheckOutput.String())
+			klog.V(0).Infof("%s check failed: \n%v", strings.Join(failedChecks, ","), individualCheckOutput.String())
 			w.WriteHeader(http.StatusServiceUnavailable)
-			fmt.Fprintf(w, individualCheckOutput.String())
+			fmt.Fprint(w, individualCheckOutput.String())
 			return
 		}
 

pkg/server/server_test.go

@@ -442,7 +442,7 @@ func TestServerProxyNoBackend(t *testing.T) {
 	}
 	baseServerProxyTestWithoutBackend(t, validate)
 
-	if err := metricstest.ExpectServerDialFailure(metrics.DialFailureNoAgent, 1); err != nil {
+	if err := metricstest.DefaultTester.ExpectServerDialFailure(metrics.DialFailureNoAgent, 1); err != nil {
 		t.Error(err)
 	}
 }
@@ -664,14 +664,14 @@ func closeRspPkt(connectID int64, errMsg string) *client.Packet {
 
 func assertEstablishedConnsMetric(t testing.TB, expect int) {
 	t.Helper()
-	if err := metricstest.ExpectServerEstablishedConns(expect); err != nil {
+	if err := metricstest.DefaultTester.ExpectServerEstablishedConns(expect); err != nil {
 		t.Errorf("Expected %d %s metric: %v", expect, "established_connections", err)
 	}
 }
 
 func assertReadyBackendsMetric(t testing.TB, expect int) {
 	t.Helper()
-	if err := metricstest.ExpectServerReadyBackends(expect); err != nil {
+	if err := metricstest.DefaultTester.ExpectServerReadyBackends(expect); err != nil {
 		t.Errorf("Expected %d %s metric: %v", expect, "ready_backend_connections", err)
 	}
 }

pkg/testing/metrics/metrics.go

@@ -58,54 +58,81 @@ const (
 konnectivity_network_proxy_agent_open_endpoint_connections %d`
 )
 
-func ExpectServerDialFailures(expected map[server.DialFailureReason]int) error {
+var DefaultTester = &Tester{}
+var _ ServerTester = DefaultTester
+var _ AgentTester = DefaultTester
+
+type Tester struct {
+	// Endpoint is the metrics endpoint to scrape metrics from. If it is empty, the in-process
+	// DefaultGatherer is used.
+	Endpoint string
+}
+
+type ServerTester interface {
+	ExpectServerDialFailures(map[server.DialFailureReason]int) error
+	ExpectServerDialFailure(server.DialFailureReason, int) error
+	ExpectServerPendingDials(int) error
+	ExpectServerReadyBackends(int) error
+	ExpectServerEstablishedConns(int) error
+}
+
+type AgentTester interface {
+	ExpectAgentDialFailures(map[agent.DialFailureReason]int) error
+	ExpectAgentDialFailure(agent.DialFailureReason, int) error
+	ExpectAgentEndpointConnections(int) error
+}
+
+func (t *Tester) ExpectServerDialFailures(expected map[server.DialFailureReason]int) error {
 	expect := serverDialFailureHeader + "\n"
 	for r, v := range expected {
 		expect += fmt.Sprintf(serverDialFailureSample+"\n", r, v)
 	}
-	return ExpectMetric(server.Namespace, server.Subsystem, "dial_failure_count", expect)
+	return t.ExpectMetric(server.Namespace, server.Subsystem, "dial_failure_count", expect)
 }
 
-func ExpectServerDialFailure(reason server.DialFailureReason, count int) error {
-	return ExpectServerDialFailures(map[server.DialFailureReason]int{reason: count})
+func (t *Tester) ExpectServerDialFailure(reason server.DialFailureReason, count int) error {
+	return t.ExpectServerDialFailures(map[server.DialFailureReason]int{reason: count})
 }
 
-func ExpectServerPendingDials(v int) error {
+func (t *Tester) ExpectServerPendingDials(v int) error {
 	expect := serverPendingDialsHeader + "\n"
 	expect += fmt.Sprintf(serverPendingDialsSample+"\n", v)
-	return ExpectMetric(server.Namespace, server.Subsystem, "pending_backend_dials", expect)
+	return t.ExpectMetric(server.Namespace, server.Subsystem, "pending_backend_dials", expect)
 }
 
-func ExpectServerReadyBackends(v int) error {
+func (t *Tester) ExpectServerReadyBackends(v int) error {
 	expect := serverReadyBackendsHeader + "\n"
 	expect += fmt.Sprintf(serverReadyBackendsSample+"\n", v)
-	return ExpectMetric(server.Namespace, server.Subsystem, "ready_backend_connections", expect)
+	return t.ExpectMetric(server.Namespace, server.Subsystem, "ready_backend_connections", expect)
 }
 
-func ExpectServerEstablishedConns(v int) error {
+func (t *Tester) ExpectServerEstablishedConns(v int) error {
 	expect := serverEstablishedConnsHeader + "\n"
 	expect += fmt.Sprintf(serverEstablishedConnsSample+"\n", v)
-	return ExpectMetric(server.Namespace, server.Subsystem, "established_connections", expect)
+	return t.ExpectMetric(server.Namespace, server.Subsystem, "established_connections", expect)
 }
 
-func ExpectAgentDialFailures(expected map[agent.DialFailureReason]int) error {
+func (t *Tester) ExpectAgentDialFailures(expected map[agent.DialFailureReason]int) error {
 	expect := agentDialFailureHeader + "\n"
 	for r, v := range expected {
 		expect += fmt.Sprintf(agentDialFailureSample+"\n", r, v)
 	}
-	return ExpectMetric(agent.Namespace, agent.Subsystem, "endpoint_dial_failure_total", expect)
+	return t.ExpectMetric(agent.Namespace, agent.Subsystem, "endpoint_dial_failure_total", expect)
 }
 
-func ExpectAgentDialFailure(reason agent.DialFailureReason, count int) error {
-	return ExpectAgentDialFailures(map[agent.DialFailureReason]int{reason: count})
+func (t *Tester) ExpectAgentDialFailure(reason agent.DialFailureReason, count int) error {
+	return t.ExpectAgentDialFailures(map[agent.DialFailureReason]int{reason: count})
 }
 
-func ExpectAgentEndpointConnections(count int) error {
+func (t *Tester) ExpectAgentEndpointConnections(count int) error {
 	expect := fmt.Sprintf(agentEndpointConnections+"\n", count)
-	return ExpectMetric(agent.Namespace, agent.Subsystem, "open_endpoint_connections", expect)
+	return t.ExpectMetric(agent.Namespace, agent.Subsystem, "open_endpoint_connections", expect)
 }
 
-func ExpectMetric(namespace, subsystem, name, expected string) error {
+func (t *Tester) ExpectMetric(namespace, subsystem, name, expected string) error {
 	fqName := prometheus.BuildFQName(namespace, subsystem, name)
-	return promtest.GatherAndCompare(prometheus.DefaultGatherer, strings.NewReader(expected), fqName)
+	if t.Endpoint == "" {
+		return promtest.GatherAndCompare(prometheus.DefaultGatherer, strings.NewReader(expected), fqName)
+	}
+	return promtest.ScrapeAndCompare(t.Endpoint, strings.NewReader(expected), fqName)
 }

tests/framework/agent.go

@@ -21,17 +21,24 @@ import (
 	"fmt"
 	"log"
 	"net"
+	"net/http"
+	"os"
+	"os/exec"
 	"path/filepath"
 	"strconv"
 	"sync"
 	"testing"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/spf13/pflag"
 	"k8s.io/apimachinery/pkg/util/wait"
 
 	agentapp "sigs.k8s.io/apiserver-network-proxy/cmd/agent/app"
 	agentopts "sigs.k8s.io/apiserver-network-proxy/cmd/agent/app/options"
 	"sigs.k8s.io/apiserver-network-proxy/pkg/agent"
+	agentmetrics "sigs.k8s.io/apiserver-network-proxy/pkg/agent/metrics"
+	metricstest "sigs.k8s.io/apiserver-network-proxy/pkg/testing/metrics"
 )
 
 type AgentOpts struct {
@@ -47,6 +54,7 @@ type Agent interface {
 	GetConnectedServerCount() (int, error)
 	Ready() bool
 	Stop()
+	Metrics() metricstest.AgentTester
 }
 type InProcessAgentRunner struct{}
 
@@ -106,6 +114,99 @@ func (a *inProcessAgent) Ready() bool {
 	return checkReadiness(a.healthAddr)
 }
 
+func (a *inProcessAgent) Metrics() metricstest.AgentTester {
+	return metricstest.DefaultTester
+}
+
+type ExternalAgentRunner struct {
+	ExecutablePath string
+}
+
+func (r *ExternalAgentRunner) Start(t testing.TB, opts AgentOpts) (Agent, error) {
+	o, err := agentOptions(t, opts)
+	if err != nil {
+		return nil, err
+	}
+
+	args := []string{}
+	o.Flags().VisitAll(func(f *pflag.Flag) {
+		args = append(args, fmt.Sprintf("--%s=%s", f.Name, f.Value.String()))
+	})
+
+	cmd := exec.Command(r.ExecutablePath, args...)
+	cmd.Stdout = os.Stdout // Forward stdout & stderr
+	cmd.Stderr = os.Stderr
+	if err := cmd.Start(); err != nil {
+		return nil, err
+	}
+
+	healthAddr := net.JoinHostPort(o.HealthServerHost, strconv.Itoa(o.HealthServerPort))
+	a := &externalAgent{
+		cmd:        cmd,
+		healthAddr: healthAddr,
+		adminAddr:  net.JoinHostPort(o.AdminBindAddress, strconv.Itoa(o.AdminServerPort)),
+		agentID:    opts.AgentID,
+		metrics:    &metricstest.Tester{Endpoint: fmt.Sprintf("http://%s/metrics", healthAddr)},
+	}
+	t.Cleanup(a.Stop)
+
+	a.waitForLiveness()
+	return a, nil
+}
+
+type externalAgent struct {
+	healthAddr, adminAddr string
+	agentID               string
+	cmd                   *exec.Cmd
+	metrics               *metricstest.Tester
+
+	stopOnce sync.Once
+}
+
+func (a *externalAgent) Stop() {
+	a.stopOnce.Do(func() {
+		if err := a.cmd.Process.Kill(); err != nil {
+			log.Fatalf("Error stopping agent process: %v", err)
+		}
+	})
+}
+
+var (
+	agentConnectedServerCountMetric = prometheus.BuildFQName(agentmetrics.Namespace, agentmetrics.Subsystem, "open_server_connections")
+)
+
+func (a *externalAgent) GetConnectedServerCount() (int, error) {
+	return readIntGauge(a.metrics.Endpoint, agentConnectedServerCountMetric)
+}
+
+func (a *externalAgent) Ready() bool {
+	resp, err := http.Get(fmt.Sprintf("http://%s/readyz", a.healthAddr))
+	if err != nil {
+		return false
+	}
+	resp.Body.Close()
+	return resp.StatusCode == http.StatusOK
+}
+
+func (a *externalAgent) Metrics() metricstest.AgentTester {
+	return a.metrics
+}
+
+func (a *externalAgent) waitForLiveness() error {
+	err := wait.PollImmediate(100*time.Millisecond, wait.ForeverTestTimeout, func() (bool, error) {
+		resp, err := http.Get(fmt.Sprintf("http://%s/healthz", a.healthAddr))
+		if err != nil {
+			return false, nil
+		}
+		resp.Body.Close()
+		return resp.StatusCode == http.StatusOK, nil
+	})
+	if err != nil {
+		return fmt.Errorf("timed out waiting for agent liveness check")
+	}
+	return nil
+}
+
 func agentOptions(t testing.TB, opts AgentOpts) (*agentopts.GrpcProxyAgentOptions, error) {
 	o := agentopts.NewGrpcProxyAgentOptions()
 

tests/framework/metrics.go

@@ -0,0 +1,63 @@
+/*
+Copyright 2023 The Kubernetes Authors.
+
+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 framework
+
+import (
+	"fmt"
+	"net/http"
+
+	dto "github.com/prometheus/client_model/go"
+	"github.com/prometheus/common/expfmt"
+)
+
+func readIntGauge(metricsURL, metricName string) (int, error) {
+	metrics, err := scrapeMetrics(metricsURL)
+	if err != nil {
+		return 0, err
+	}
+
+	metric, found := metrics[metricName]
+	if !found {
+		return 0, fmt.Errorf("missing %s metric", metricName)
+	}
+
+	vals := metric.GetMetric()
+	if len(vals) == 0 {
+		return 0, fmt.Errorf("missing value for %s metric", metricName)
+	}
+
+	return int(vals[0].GetGauge().GetValue()), nil
+}
+
+func scrapeMetrics(metricsURL string) (map[string]*dto.MetricFamily, error) {
+	resp, err := http.Get(metricsURL)
+	if err != nil {
+		return nil, fmt.Errorf("failed to scrape metrics from %s: %w", metricsURL, err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("unexpected %q status scraping metrics from %s", resp.Status, metricsURL)
+	}
+
+	var parser expfmt.TextParser
+	mf, err := parser.TextToMetricFamilies(resp.Body)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse metrics: %w", err)
+	}
+	return mf, nil
+}