Add persistent HTTP client to protocol_loadtest.

Summary:
Currently there is a wrk client, and a seq_tests command line tool. The wrk client is persistent (i.e. doesn't stop requesting until its killed), whereas the command line tool runs one off tests for data loss. This diff adds a new client that reuses the pkg code from the seq_tests command line tool, to continually run data loss tests with http load.

This diff also updates the seq_tests code to ensure that the specified number of connections is used exactly, where before it only specified the number of workers.

Test Plan: Ran skaffold to deploy the server. Ran skaffold to deploy the new client. Saw that the new client successfully requests from the server.

Reviewers: nserrino, zasgar, vihang, philkuz

Reviewed By: nserrino

Signed-off-by: James Bartlett <jamesbartlett@pixielabs.ai>

Differential Revision: https://phab.corp.pixielabs.ai/D12196

GitOrigin-RevId: 13844908dfd2726ff5f905ecec92fa4b1e643540

src/e2e_test/protocol_loadtest/client/BUILD.bazel

@@ -0,0 +1,57 @@
+# Copyright 2018- The Pixie 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.
+#
+# SPDX-License-Identifier: Apache-2.0
+
+load("@io_bazel_rules_docker//container:container.bzl", "container_push")
+load("@io_bazel_rules_go//go:def.bzl", "go_binary", "go_library")
+load("//bazel:go_image_alias.bzl", "go_image")
+
+go_library(
+    name = "client_lib",
+    srcs = ["client.go"],
+    importpath = "px.dev/pixie/src/e2e_test/protocol_loadtest/client",
+    visibility = ["//visibility:private"],
+    deps = [
+        "//src/e2e_test/vizier/seq_tests/client/pkg/httpclient",
+        "@com_github_sirupsen_logrus//:logrus",
+        "@com_github_spf13_pflag//:pflag",
+        "@com_github_spf13_viper//:viper",
+    ],
+)
+
+go_binary(
+    name = "client",
+    embed = [":client_lib"],
+    visibility = ["//visibility:public"],
+)
+
+go_image(
+    name = "protocol_loadtest_client_image",
+    binary = ":client",
+    importpath = "px.dev/pixie",
+    visibility = [
+        "//src/e2e_test:__subpackages__",
+    ],
+)
+
+container_push(
+    name = "push_protocol_loadtest_client_image",
+    format = "Docker",
+    image = ":protocol_loadtest_client_image",
+    registry = "gcr.io",
+    repository = "pixie-oss/pixie-dev/src/e2e_test/protocol_loadtest/client/protocol_loadtest_client",
+    tag = "{STABLE_BUILD_TAG}",
+    tags = ["manual"],
+)

src/e2e_test/protocol_loadtest/client/client.go

@@ -0,0 +1,72 @@
+/*
+ * Copyright 2018- The Pixie 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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package main
+
+import (
+	"fmt"
+
+	log "github.com/sirupsen/logrus"
+	"github.com/spf13/pflag"
+	"github.com/spf13/viper"
+
+	"px.dev/pixie/src/e2e_test/vizier/seq_tests/client/pkg/httpclient"
+)
+
+func init() {
+	pflag.String("http_host", "", "The host of the http server")
+	pflag.String("http_path", "", "The request path to request from the http server")
+	pflag.Int("http_port", 0, "Port of the http server")
+
+	pflag.Int("num_connections", 0, "Number of simulataneous connections for the seq load generator")
+	pflag.Int("req_size", 16, "Size of the request in bytes")
+	pflag.Int("resp_size", 16, "Size of the response in bytes")
+	pflag.Int("num_messages", 1000, "Num messages per conn per loop")
+}
+
+func main() {
+	viper.AutomaticEnv()
+	viper.BindPFlags(pflag.CommandLine)
+
+	host := viper.GetString("http_host")
+	path := viper.GetString("http_path")
+	port := viper.GetInt("http_port")
+	addr := fmt.Sprintf("http://%s:%d%s", host, port, path)
+
+	numConns := viper.GetInt("num_connections")
+	numMessagesPerConn := viper.GetInt("num_messages")
+	reqSize := viper.GetInt("req_size")
+	respSize := viper.GetInt("resp_size")
+
+	numMessages := numMessagesPerConn * numConns
+
+	seqNum := 0
+	for {
+		log.Infof("Started loadtest with %d conns, %d messages, %d req_size, %d resp_size", numConns, numMessages, reqSize, respSize)
+		c := httpclient.New(addr, seqNum, numMessages, numConns, reqSize, respSize)
+		err := c.Run()
+		if err != nil {
+			log.WithError(err).Error("failed to run seq client")
+		}
+		err = c.PrintStats()
+		if err != nil {
+			log.WithError(err).Error("failed to print stats")
+		}
+		seqNum += numMessages + 1
+	}
+}

src/e2e_test/protocol_loadtest/k8s/client/client_deployment.yaml

@@ -0,0 +1,31 @@
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: client
+  namespace: px-protocol-loadtest
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      name: client
+  template:
+    metadata:
+      labels:
+        name: client
+    spec:
+      containers:
+      - name: app
+        image: gcr.io/pixie-oss/pixie-dev/src/e2e_test/protocol_loadtest/client/protocol_loadtest_client_image:latest
+        env:
+        - name: HTTP_PORT
+          value: "8080"
+        - name: HTTP_SSL_PORT
+          value: "8081"
+        - name: HTTP_HOST
+          value: "server.px-protocol-loadtest.svc.cluster.local"
+        - name: HTTP_PATH
+          value: "/"
+        envFrom:
+        - configMapRef:
+            name: px-protocol-loadtest-config

src/e2e_test/protocol_loadtest/k8s/client/kustomization.yaml

@@ -0,0 +1,7 @@
+---
+apiVersion: kustomize.config.k8s.io/v1beta1
+kind: Kustomization
+namespace: px-protocol-loadtest
+resources:
+- client_deployment.yaml
+- loadtest_config.yaml

src/e2e_test/protocol_loadtest/k8s/client/loadtest_config.yaml

@@ -0,0 +1,7 @@
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: px-protocol-loadtest-config
+data:
+  NUM_CONNECTIONS: "99"

src/e2e_test/protocol_loadtest/skaffold_client.yaml

@@ -0,0 +1,17 @@
+---
+apiVersion: skaffold/v2alpha3
+kind: Config
+build:
+  artifacts:
+  - image: gcr.io/pixie-oss/pixie-dev/src/e2e_test/protocol_loadtest/client/protocol_loadtest_client_image
+    context: .
+    bazel:
+      target: //src/e2e_test/protocol_loadtest/client:protocol_loadtest_client_image.tar
+  tagPolicy:
+    dateTime: {}
+  local:
+    push: true
+deploy:
+  kustomize:
+    paths:
+    - src/e2e_test/protocol_loadtest/k8s/client

src/e2e_test/vizier/seq_tests/client/pkg/httpclient/client.go

@@ -20,10 +20,13 @@ package httpclient
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
 	"fmt"
 	"io"
+	"net"
 	"net/http"
+	"net/url"
 	"sync"
 	"time"
 
@@ -110,12 +113,37 @@ func (c *HTTPSeqClient) PrintStats() error {
 func (c *HTTPSeqClient) worker(wg *sync.WaitGroup, jobs <-chan int, results chan<- error) {
 	defer wg.Done()
 
+	d := net.Dialer{
+		Timeout:   30 * time.Second,
+		KeepAlive: 30 * time.Second,
+	}
+	u, err := url.Parse(c.addr)
+	if err != nil {
+		results <- err
+		return
+	}
+	conn, err := d.DialContext(context.Background(), "tcp", u.Host)
+	if err != nil {
+		results <- err
+		return
+	}
+	defer conn.Close()
+
+	t := &http.Transport{
+		DialContext: func(context.Context, string, string) (net.Conn, error) {
+			return conn, nil
+		},
+	}
+	client := &http.Client{
+		Transport: t,
+	}
+
 	for j := range jobs {
-		results <- makeSingleRequest(c.addr, j, c.reqSize, c.respSize)
+		results <- makeSingleRequest(client, c.addr, j, c.reqSize, c.respSize)
 	}
 }
 
-func makeSingleRequest(addr string, seqID, reqSize, respSize int) error {
+func makeSingleRequest(client *http.Client, addr string, seqID, reqSize, respSize int) error {
 	body := struct {
 		SeqID          int    `json:"seq_id"`
 		RespSize       int    `json:"resp_size"`
@@ -132,7 +160,6 @@ func makeSingleRequest(addr string, seqID, reqSize, respSize int) error {
 		return err
 	}
 
-	client := &http.Client{}
 	req, err := http.NewRequest("POST", addr, &buf)
 	if err != nil {
 		return err