SKUPPER-2086: add connection rate test to Benchdog

README.md

@@ -49,6 +49,8 @@ Sample output:
 
 [**pgbench**](pgbench) - PostgreSQL
 
+[**cbench**](cbench) - Connection Rate
+
 [**wrk**](wrk) - HTTP/1.1
 
 ## Configuration

cbench/README.md

@@ -0,0 +1,53 @@
+# Benchdog : cbench
+
+A containerized test measuring Skupper connection rate.
+
+The Cbench client repeatedly makes and then closes connections with the server. After making each connection, it waits to receive a small message back from the server. The duration that the client records is the time between its request for a connection, and its receipt of the message from the server. 
+
+This is a slight over estimate of the connection time, but this is the only way we can know that the connection has gone all the way through the router network to the target server.
+
+Each test lasts 15 seconds. At the end of each test, the client prints out statistics for that test.
+
+
+## Testing with Skupper
+
+### Server namespace:
+
+```
+    kubectl create namespace cbench-server
+    kubectl config set-context --current --namespace cbench-server
+    skupper init
+    kubectl create deployment cbench-server --image quay.io/skupper/benchdog-cbench-server
+    skupper expose deployment/cbench-server --port 5800
+    skupper token create ~/token.yaml
+```
+
+### Client namespace:
+
+```
+    kubectl create namespace cbench-client
+    kubectl config set-context --current --namespace=cbench-client
+    skupper init
+    skupper link create ~/token.yaml
+    skupper link status      # Make sure it's connected
+    kubectl get services     # Look for cbench-server
+    kubectl run --env="CBENCH_HOST=cbench-server" --env="CBENCH_PORT=5800" --image quay.io/skupper/benchdog-cbench-client cbench-client
+    kubectl logs cbench-client  # To see the output
+```
+
+## Controlling number of client threads
+
+The client main program does not itself create connections to the server. Instead, it launches one or more threads, each of which repeatedly makes and closes connections as quickly as possible in a loop. Each separate test can use a different number of threads. 
+
+The number of tests that are run, and the number of threads used in each test, is controlled by the N\_CLIENTS\_LIST environment variable. If it is not present, its value defaults to "1 2 5". You can override the default by giving it a new value in the kubectl run command like so:
+
+```
+    kubectl run --env="CBENCH_HOST=cbench-server" --env="CBENCH_PORT=5800" --env N_CLIENTS_LIST="1 2 3 4" --image quay.io/skupper/benchdog-cbench-client cbench-client
+```
+
+It will take a little over 15 seconds to run each test, so you may need to look at the logs several times to get all of your output.
+
+
+
+
+

cbench/client/Dockerfile

@@ -0,0 +1,28 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you 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.
+#
+
+FROM registry.fedoraproject.org/fedora-minimal
+COPY .  /cbench/client
+WORKDIR /cbench/client
+RUN microdnf -y --setopt=install_weak_deps=False install gcc python
+RUN gcc -o client client.c
+RUN chmod +x ./r_client
+RUN chmod +x ./collect.py
+CMD ["./r_client"]
+

cbench/client/client.c

@@ -0,0 +1,171 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.
+ */
+
+#define _GNU_SOURCE
+
+#include <errno.h>
+#include <netdb.h>
+#include <pthread.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/socket.h>
+#include <sys/time.h>
+#include <unistd.h>
+
+typedef struct thread_context {
+    int id;
+    uint16_t port;
+    char * host_name;
+    char * output_dir;
+} thread_context_t;
+
+
+double first_timestamp = 0;
+
+double timestamp() {
+    struct timeval t;
+    gettimeofday(&t, 0);
+    double ts = t.tv_sec + ((double) t.tv_usec) / 1000000.0;
+    return ts - first_timestamp;
+}
+
+
+void * client ( void * data ) {
+    thread_context_t * ctx = (thread_context_t *) data;
+    char * output_dir = ctx->output_dir;
+    char * host_name  = ctx->host_name;
+    int id            = ctx->id;
+    uint16_t port_number          = ctx->port;
+
+    // Get the output file
+    char transfers_file[256];
+    snprintf(transfers_file, 256, "%s/connections.%d.data", output_dir, id);
+
+    FILE* transfers = fopen(transfers_file, "w");
+    if (!transfers) goto bailout;
+
+    int connection_count = 0;
+    while (1) {
+        struct sockaddr_in router_addr;
+        struct hostent * router;
+        int sockfd;
+
+        if ( (sockfd = socket(AF_INET, SOCK_STREAM, 0)) < 0 ) {
+            perror ( "Socket creation failed" );
+            exit ( 1 );
+        }
+
+        if ((router = gethostbyname(host_name)) == NULL) {
+            fprintf(stderr, "No such host: %s\n", host_name);
+            exit ( 1 );
+        }
+
+        // Set router address and port
+        bzero ( (char *) & router_addr, sizeof(router_addr) );
+        router_addr.sin_family = AF_INET;
+        bcopy ((char *)router->h_addr,
+               (char *)&router_addr.sin_addr.s_addr,
+               router->h_length);
+        router_addr.sin_port = htons(port_number);
+
+        double start_time = timestamp();
+        // Connect to router
+        if (connect(sockfd, (struct sockaddr *)&router_addr, sizeof(router_addr)) < 0) {
+            perror("Connection failed");
+            exit(1);
+        }
+
+        // Receive message from server, and measure the time it took
+        // to make this connection.
+        // This means we are overestimating the connect time, but this 
+        // is the only way we can be sure that the connection has gone 
+        // all the way through the router network to the server.
+        char buffer[1024];
+        ssize_t n = recv ( sockfd, buffer, sizeof(buffer) - 1, 0 );
+        close ( sockfd );
+        if (n <= 0) {
+          fprintf ( stderr, "message failure: recv returned %zd\n", n );
+          goto bailout;
+        }
+
+        // Measure the connection time and store it in the output file.
+        double duration = timestamp() - start_time;
+        ++ connection_count;
+        fprintf(transfers, "%.6lf,%6lf\n", start_time, duration);
+        fflush(transfers);
+    }
+
+bailout:
+
+    fclose(transfers);
+    if (errno) {
+        fprintf(stderr, "client: ERROR! %s\n", strerror(errno));
+    }
+}
+
+
+int main(size_t argc, char** argv) {
+    if (argc != 3) {
+        fprintf(stderr, "Usage: client JOBS OUTPUT-DIR\n");
+        return 1;
+    }
+    int jobs = atoi(argv[1]);
+    char* output_dir = argv[2];
+
+    // These two are environment variables because
+    // I want to be able to supply them from the 
+    // 'kubectl run' command.
+    char* host     = getenv("CBENCH_HOST");
+    char* port_str = getenv("CBENCH_PORT");
+
+    if (! (host && port_str)) {
+        fprintf(stderr, "need host and port\n");
+        exit(1);
+    }
+    uint16_t port_number = atoi(port_str);
+
+    thread_context_t contexts       [jobs];
+    pthread_t        client_threads [jobs];
+
+    // This will be Time Zero for all other timestamps.
+    first_timestamp = timestamp();
+
+    // Start all the threads that will make connections
+    // as fast as they can.
+    // Send each one its own context.
+    for (int i = 0; i < jobs; i++) {
+        contexts[i] = (thread_context_t) {
+            .id = i,
+            .port = port_number,
+            .host_name = host,
+            .output_dir = output_dir,
+        };
+        pthread_create(client_threads+i, NULL, & client, contexts+i);
+    }
+
+    for(int i = 0; i < jobs; i++) {
+        pthread_join(client_threads[i], NULL);
+    }
+
+    exit(errno);
+}
+
+
+

cbench/client/collect.py

@@ -0,0 +1,77 @@
+#! /usr/bin/python
+
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you 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.
+#
+
+import sys as _sys
+
+durations = []
+last_time = 0.0
+
+print_header = _sys.argv[1]
+n_clients = _sys.argv[2]
+
+for file_name in _sys.argv[3:] :
+    with open(file_name) as f:
+      lines = f.readlines()
+      for line in lines : 
+          # format:   request_time,connection_complete_time
+          # example:  0.027601,0.000699
+          words = line.strip().split(',')
+          if len(words) < 2 :
+              continue
+          # The file line are in chron order, so this will 
+          # always have the last-recorded time in it.
+          last_time = words[0]
+          duration  = words[1]
+          durations.append ( float(duration) )
+
+
+sorted_durations = sorted(durations, key = lambda x:float(x))
+duration_sum     = sum(sorted_durations)
+average_dur      = duration_sum / len(durations)
+
+pos_50 = int(len(durations) / 2)
+pos_99 = int(len(durations) * 0.99)
+dur_50 = sorted_durations[pos_50]
+dur_99 = sorted_durations[pos_99]
+
+# Convert to msec
+dur_50      *= 1000
+dur_99      *= 1000
+average_dur *= 1000
+
+cnx_per_sec = len(durations) / float(last_time)
+cps         = f"{cnx_per_sec:.2f}  cnx/s"
+avg         = f"{average_dur:.2f} ms"
+d_50        = f"{dur_50:.2f} ms"
+d_99        = f"{dur_99:.2f} ms"
+
+col_1 = "CLIENTS"
+col_2 = "THROUGHPUT" 
+col_3 = "LATENCY AVG"
+col_4 = "LATENCY P50"
+col_5 = "LATENCY P99"
+
+if print_header == "print_header" :
+    print ( f"{col_1:>11}{col_2:>22}{col_3:>20}{col_4:>20}{col_5:>20}" )
+print ( f"{n_clients:>11}{cps:>22}{avg:>20}{d_50:>20}{d_99:>20}" )
+
+
+