diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8134829f7e..ab47696b74 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -348,6 +348,12 @@ jobs: set -euo pipefail # shellcheck disable=SC2046 go test -v -race -tags infrastructure -run '^${{ matrix.test }}$' $(git grep -l '^//go:build infrastructure' | xargs grep -l '^func ${{ matrix.test }}' | xargs -I {} dirname ./{}) + - name: Archive Report + uses: actions/upload-artifact@v4 + if: always() # Always upload the report even on failure + with: + name: kube-report-${{ matrix.test }} + path: /tmp/ftl-kube-report/ integration-success: name: Integration Success needs: [integration-run] diff --git a/backend/controller/scaling/kube_scaling_integration_test.go b/backend/controller/scaling/kube_scaling_integration_test.go index 69e1b104cc..fdf904e10a 100644 --- a/backend/controller/scaling/kube_scaling_integration_test.go +++ b/backend/controller/scaling/kube_scaling_integration_test.go @@ -34,7 +34,7 @@ func TestKubeScaling(t *testing.T) { in.Call("echo", "echo", "Bob", func(t testing.TB, response string) { assert.Equal(t, "Hello, Bob!!!", response) }), - in.VerifyKubeState(func(ctx context.Context, t testing.TB, namespace string, client *kubernetes.Clientset) { + in.VerifyKubeState(func(ctx context.Context, t testing.TB, namespace string, client kubernetes.Clientset) { deps, err := client.AppsV1().Deployments(namespace).List(ctx, v1.ListOptions{}) assert.NoError(t, err) for _, dep := range deps.Items { @@ -55,7 +55,12 @@ func TestKubeScaling(t *testing.T) { func(t testing.TB, ic in.TestContext) { // Hit the verb constantly to test rolling updates. go func() { - defer routineStopped.Done() + defer func() { + if r := recover(); r != nil { + failure.Store(fmt.Errorf("panic in verb: %v", r)) + } + routineStopped.Done() + }() for !done.Load() { in.Call("echo", "echo", "Bob", func(t testing.TB, response string) { if !strings.Contains(response, "Bob") { @@ -76,7 +81,7 @@ func TestKubeScaling(t *testing.T) { err := failure.Load() assert.NoError(t, err) }, - in.VerifyKubeState(func(ctx context.Context, t testing.TB, namespace string, client *kubernetes.Clientset) { + in.VerifyKubeState(func(ctx context.Context, t testing.TB, namespace string, client kubernetes.Clientset) { deps, err := client.AppsV1().Deployments(namespace).List(ctx, v1.ListOptions{}) assert.NoError(t, err) depCount := 0 diff --git a/go.mod b/go.mod index 910471e5b5..8be52d5ca6 100644 --- a/go.mod +++ b/go.mod @@ -81,6 +81,7 @@ require ( k8s.io/apimachinery v0.31.1 k8s.io/client-go v0.31.1 modernc.org/sqlite v1.33.1 + sigs.k8s.io/yaml v1.4.0 ) require ( @@ -155,7 +156,6 @@ require ( modernc.org/libc v1.55.3 // indirect sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect sigs.k8s.io/structured-merge-diff/v4 v4.4.1 // indirect - sigs.k8s.io/yaml v1.4.0 // indirect ) require ( diff --git a/internal/integration/actions.go b/internal/integration/actions.go index 13b3ba4e5a..3f93b7b045 100644 --- a/internal/integration/actions.go +++ b/internal/integration/actions.go @@ -227,7 +227,7 @@ func Deploy(module string) Action { if ic.Provisioner != nil { args = append(args, "--use-provisioner", "--provisioner-endpoint=http://localhost:8893") } - if ic.kubeClient != nil { + if ic.kubeClient.Ok() { args = append(args, "--build-env", "GOOS=linux", "--build-env", "GOARCH=amd64", "--build-env", "CGO_ENABLED=0") } args = append(args, module) @@ -395,9 +395,9 @@ func Call[Req any, Resp any](module, verb string, request Req, check func(t test } // VerifyKubeState lets you test the current kube state -func VerifyKubeState(check func(ctx context.Context, t testing.TB, namespace string, client *kubernetes.Clientset)) Action { +func VerifyKubeState(check func(ctx context.Context, t testing.TB, namespace string, client kubernetes.Clientset)) Action { return func(t testing.TB, ic TestContext) { - check(ic.Context, t, ic.kubeNamespace, ic.kubeClient) + check(ic.Context, t, ic.kubeNamespace, ic.kubeClient.MustGet()) } } diff --git a/internal/integration/harness.go b/internal/integration/harness.go index 209965afad..1c450fa4cb 100644 --- a/internal/integration/harness.go +++ b/internal/integration/harness.go @@ -7,6 +7,7 @@ import ( "context" "errors" "fmt" + "io" "os" "path/filepath" "runtime" @@ -22,6 +23,11 @@ import ( "github.com/otiai10/copy" "k8s.io/client-go/kubernetes" + kubecore "k8s.io/api/core/v1" + kubemeta "k8s.io/apimachinery/pkg/apis/meta/v1" + + "sigs.k8s.io/yaml" + "github.com/TBD54566975/ftl/backend/controller/scaling/k8sscaling" ftlv1 "github.com/TBD54566975/ftl/backend/protos/xyz/block/ftl/v1" "github.com/TBD54566975/ftl/backend/protos/xyz/block/ftl/v1/console/pbconsoleconnect" @@ -33,13 +39,15 @@ import ( "github.com/TBD54566975/ftl/internal/rpc" ) +const dumpPath = "/tmp/ftl-kube-report" + func (i TestContext) integrationTestTimeout() time.Duration { timeout := optional.Zero(os.Getenv("FTL_INTEGRATION_TEST_TIMEOUT")).Default("5s") d, err := time.ParseDuration(timeout) if err != nil { panic(err) } - if i.kubeClient != nil { + if i.kubeClient.Ok() { // kube can be slow, give it some time return d * 5 } @@ -277,9 +285,10 @@ func run(t *testing.T, actionsOrOptions ...ActionOrOption) { Verbs: verbs, realT: t, language: language, - kubeClient: kubeClient, kubeNamespace: kubeNamespace, + kubeClient: optional.Ptr(kubeClient), } + defer ic.dumpKubePods() if opts.startController || opts.kube { ic.Controller = controller @@ -354,7 +363,7 @@ type TestContext struct { // The Language under test language string // Set if the test is running on kubernetes - kubeClient *kubernetes.Clientset + kubeClient optional.Option[kubernetes.Clientset] kubeNamespace string Controller ftlv1connect.ControllerServiceClient @@ -465,3 +474,54 @@ func startProcess(ctx context.Context, t testing.TB, args ...string) context.Con }) return ctx } + +func (i TestContext) dumpKubePods() { + if client, ok := i.kubeClient.Get(); ok { + _ = os.RemoveAll(dumpPath) // #nosec + list, err := client.CoreV1().Pods(i.kubeNamespace).List(i, kubemeta.ListOptions{}) + if err == nil { + for _, pod := range list.Items { + Infof("Dumping logs for pod %s", pod.Name) + podPath := filepath.Join(dumpPath, pod.Name) + err := os.MkdirAll(podPath, 0755) // #nosec + if err != nil { + Infof("Error creating directory %s: %v", podPath, err) + continue + } + podYaml, err := yaml.Marshal(pod) + if err != nil { + Infof("Error marshalling pod %s: %v", pod.Name, err) + continue + } + err = os.WriteFile(filepath.Join(podPath, "pod.yaml"), podYaml, 0644) // #nosec + if err != nil { + Infof("Error writing pod %s: %v", pod.Name, err) + continue + } + for _, container := range pod.Spec.Containers { + path := filepath.Join(dumpPath, pod.Name, container.Name+".log") + req := client.CoreV1().Pods(i.kubeNamespace).GetLogs(pod.Name, &kubecore.PodLogOptions{Container: container.Name}) + podLogs, err := req.Stream(context.Background()) + defer func() { + _ = podLogs.Close() + }() + if err != nil { + Infof("Error getting logs for pod %s: %v", pod.Name, err) + continue + } + buf := new(bytes.Buffer) + _, err = io.Copy(buf, podLogs) + if err != nil { + Infof("Error copying logs for pod %s: %v", pod.Name, err) + continue + } + str := buf.String() + err = os.WriteFile(path, []byte(str), 0644) // #nosec + if err != nil { + Infof("Error writing logs for pod %s: %v", pod.Name, err) + } + } + } + } + } +}