Merge pull request #1485 from mumubin/support-kubectl

feat: loadtester support kubectl type

Dockerfile.loadtester

@@ -10,6 +10,10 @@ RUN HELM3_VERSION=3.11.0 && \
 curl -sSL "https://get.helm.sh/helm-v${HELM3_VERSION}-linux-${TARGETARCH}.tar.gz" | tar xvz && \
 chmod +x linux-${TARGETARCH}/helm && mv linux-${TARGETARCH}/helm /usr/local/bin/helm
 
+RUN KUBECTL_VERSION=v1.28.0 &&  \
+curl -LO "https://dl.k8s.io/release/${KUBECTL_VERSION}/bin/linux/${TARGETARCH}/kubectl" && \
+chmod +x kubectl && mv kubectl /usr/local/bin/kubectl
+
 RUN GRPC_HEALTH_PROBE_VERSION=v0.4.12 && \
 wget -qO /usr/local/bin/grpc_health_probe https://github.com/grpc-ecosystem/grpc-health-probe/releases/download/${GRPC_HEALTH_PROBE_VERSION}/grpc_health_probe-linux-${TARGETARCH} && \
 chmod +x /usr/local/bin/grpc_health_probe
@@ -41,7 +45,7 @@ ARG TARGETPLATFORM
 
 RUN addgroup -S app && \
 adduser -S -g app app && \
-apk --no-cache add ca-certificates curl jq libgcc wrk hey
+apk --no-cache add ca-certificates curl jq libgcc wrk hey git
 
 WORKDIR /home/app
 
@@ -51,6 +55,7 @@ RUN ln -s /opt/bats/bin/bats /usr/local/bin/
 COPY --from=builder /usr/local/bin/helm /usr/local/bin/
 COPY --from=builder /usr/local/bin/ghz /usr/local/bin/
 COPY --from=builder /usr/local/bin/grpc_health_probe /usr/local/bin/
+COPY --from=builder /usr/local/bin/kubectl /usr/local/bin/
 
 ADD https://raw.githubusercontent.com/grpc/grpc-proto/master/grpc/health/v1/health.proto /tmp/ghz/health.proto
 

pkg/loadtester/kubectl.go

@@ -0,0 +1,42 @@
+package loadtester
+
+import (
+	"context"
+	"fmt"
+	"os/exec"
+	"strings"
+)
+
+const TaskTypeKubectl = "kubectl"
+
+type KubectlTask struct {
+	TaskBase
+	command      string
+	logCmdOutput bool
+}
+
+func (task *KubectlTask) Hash() string {
+	return hash(task.canary + task.command)
+}
+
+func (task *KubectlTask) Run(ctx context.Context) (*TaskRunResult, error) {
+	kubectlCmd := fmt.Sprintf("%s %s", TaskTypeKubectl, task.command)
+	task.logger.With("canary", task.canary).Infof("running command %v", kubectlCmd)
+
+	cmd := exec.CommandContext(ctx, TaskTypeKubectl, strings.Fields(task.command)...)
+	out, err := cmd.CombinedOutput()
+	if err != nil {
+		task.logger.With("canary", task.canary).Errorf("command failed %s %v %s", task.command, err, out)
+		return &TaskRunResult{false, out}, fmt.Errorf("command %s failed: %s: %w", task.command, out, err)
+	} else {
+		if task.logCmdOutput {
+			task.logger.With("canary", task.canary).Info(string(out))
+		}
+		task.logger.With("canary", task.canary).Infof("command finished %v", kubectlCmd)
+	}
+	return &TaskRunResult{true, out}, nil
+}
+
+func (task *KubectlTask) String() string {
+	return task.command
+}

pkg/loadtester/server.go

@@ -362,7 +362,32 @@ func HandleNewTask(logger *zap.SugaredLogger, taskRunner TaskRunnerInterface, au
 				}
 				return
 			}
+			//run kubectl cmd
+			if typ == TaskTypeKubectl {
+				kubectl := KubectlTask{
+					command:      payload.Metadata["cmd"],
+					logCmdOutput: true,
+					TaskBase: TaskBase{
+						canary: fmt.Sprintf("%s.%s", payload.Name, payload.Namespace),
+						logger: logger,
+					},
+				}
 
+				ctx, cancel := context.WithTimeout(context.Background(), taskRunner.Timeout())
+				defer cancel()
+
+				result, err := kubectl.Run(ctx)
+				if !result.ok {
+					w.WriteHeader(http.StatusInternalServerError)
+					w.Write([]byte(err.Error()))
+					return
+				}
+				w.WriteHeader(http.StatusOK)
+				if rtnCmdOutput {
+					w.Write(result.out)
+				}
+				return
+			}
 			// run concord job (blocking task)
 			if typ == TaskTypeConcord {
 				concord, err := NewConcordTask(payload.Metadata, fmt.Sprintf("%s.%s", payload.Name, payload.Namespace), logger)