diff --git a/Dockerfile.loadtester b/Dockerfile.loadtester index d34ab58b1..345b6cd31 100644 --- a/Dockerfile.loadtester +++ b/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 @@ -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 diff --git a/pkg/loadtester/kubectl.go b/pkg/loadtester/kubectl.go new file mode 100644 index 000000000..f383d943e --- /dev/null +++ b/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 +} diff --git a/pkg/loadtester/server.go b/pkg/loadtester/server.go index ce1664458..5f7a3674e 100644 --- a/pkg/loadtester/server.go +++ b/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)