Skip to content

Commit 3adec6d

Browse files
authored
Merge pull request #2301 from vyncint/fix-e2e-cluster-concurrency
fix(e2e): give every kind cluster its own kubeconfig
2 parents 31e84e1 + b951ff5 commit 3adec6d

10 files changed

Lines changed: 211 additions & 324 deletions

File tree

.github/workflows/e2e.yaml

Lines changed: 0 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -76,9 +76,6 @@ jobs:
7676
needs: build
7777
strategy:
7878
fail-fast: false
79-
# matrix:
80-
# SHARD: [0]
81-
# SHARDS: [1]
8279

8380
steps:
8481
- name: Free Disk Space (Ubuntu)
@@ -132,9 +129,6 @@ jobs:
132129

133130
- name: Run e2e tests
134131
run: make test-e2e-ci
135-
# env:
136-
# SHARD: ${{ matrix.SHARD }}
137-
# SHARDS: ${{ matrix.SHARDS }}
138132

139133
- name: Run coverage report
140134
uses: vladopajic/go-test-coverage@a93b868a4cbcbf18dc3781650fad241f0020e609 # v2.18.8

CLAUDE.md

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -155,7 +155,7 @@ v1alpha1 is the legacy API; conversion functions exist to v1beta1 (which is the
155155

156156
- Unit/integration tests use `envtest` (embedded Kubernetes API server + etcd); no cluster needed
157157
- E2E tests in `e2e/` use KIND: one directory per suite, each its own Go package and test binary, provisioning its own KIND cluster. Run a single suite with `make test-e2e E2E_TEST=<suite-dir>`. Shared helpers live in `e2e/common/` and `e2e/internal/` and are excluded from suite selection
158-
- E2E knobs (all `make` overrides): `E2E_TEST_TIMEOUT` (per suite binary, default 20m), `E2E_SUITE_PARALLEL` (clusters one suite binary builds at once, default 2 — raising it starves the aggregators on a 4-vCPU runner), `KIND_COMMAND_TIMEOUT` (per kind invocation; derived from `E2E_TEST_TIMEOUT` when unset)
158+
- E2E knobs (all `make` overrides): `E2E_TEST_TIMEOUT` (per suite binary, default 20m), `E2E_SUITE_PARALLEL` (clusters one suite binary builds at once, default 2 — raising it starves the aggregators on a 4-vCPU runner), `E2E_CLUSTERS` (suite binaries at once, default 4; holds peak clusters at 5, which otherwise follows the core count), `KIND_COMMAND_TIMEOUT` (per kind invocation; derived from `E2E_TEST_TIMEOUT` when unset)
159159
- Coverage profile config in `.testcoverage.yml`; tool: `go-test-coverage`
160160
- Test framework: Ginkgo + Gomega for BDD-style tests; testify for unit tests
161161

Makefile

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,10 @@ E2E_TEST_TIMEOUT ?= 20m
5858
# otherwise follows the core count and starves the aggregators.
5959
E2E_SUITE_PARALLEL ?= 2
6060

61+
# Suite binaries running at once. Pinned, peak clusters holds at 5; left to -p
62+
# it follows the core count, so eight cores would reach 10 and sixteen 15.
63+
E2E_CLUSTERS ?= 4
64+
6165
TEST_COV_DIR := $(shell mkdir -p build/_test_coverage && realpath build/_test_coverage)
6266

6367
CONTROLLER_GEN := ${BIN}/controller-gen
@@ -247,7 +251,7 @@ test-e2e-nodeps:
247251
KIND_IMAGE="$(KIND_IMAGE)" \
248252
PROJECT_DIR="$(PWD)" \
249253
E2E_TEST_COV_DIR=${TEST_COV_DIR} \
250-
go test -count=1 -v -parallel ${E2E_SUITE_PARALLEL} -timeout ${E2E_TEST_TIMEOUT} $$(go list ./${E2E_TEST}/... | grep -vE '/e2e/(common|internal)(/|$$)')
254+
go test -count=1 -v -p ${E2E_CLUSTERS} -parallel ${E2E_SUITE_PARALLEL} -timeout ${E2E_TEST_TIMEOUT} $$(go list ./${E2E_TEST}/... | grep -vE '/e2e/(common|internal)(/|$$)')
251255
go tool covdata textfmt -i=${TEST_COV_DIR}/covdatafiles -o ${TEST_COV_DIR}/coverage_e2e.out
252256
@echo "--- E2E test coverage report"
253257
go tool covdata percent -i=${TEST_COV_DIR}/covdatafiles

controllers/logging/logging_controller_test.go

Lines changed: 9 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1485,15 +1485,18 @@ func TestWatchNamespaces(t *testing.T) {
14851485
continue
14861486
}
14871487

1488-
g.Eventually(func() ReturnVal {
1488+
// The expectation is recomputed on every poll rather than passed in as an
1489+
// argument, which Go would evaluate once before polling starts. The "full
1490+
// list" case reads the live namespace set, and envtest never finishes
1491+
// deleting a namespace, so a value snapshotted up front goes stale as the
1492+
// rest of the package creates more and the poll can never match it.
1493+
g.Eventually(func(poll gomega.Gomega) {
14891494
n, e := model.UniqueWatchNamespaces(context.TODO(), mgr.GetClient(), c.logging)
1490-
return ReturnVal{
1495+
poll.Expect(ReturnVal{
14911496
namespaces: n,
14921497
err: e,
1493-
}
1494-
}, timeout).Should(gomega.Equal(
1495-
c.expectedResult(),
1496-
))
1498+
}).To(gomega.Equal(c.expectedResult()))
1499+
}, timeout).Should(gomega.Succeed())
14971500
}
14981501
}
14991502

e2e/common/cluster.go

Lines changed: 27 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@ import (
2222
"os/exec"
2323
"strings"
2424
"testing"
25+
"time"
2526

2627
"emperror.dev/errors"
2728
"github.com/spf13/cast"
@@ -35,6 +36,9 @@ import (
3536
"github.com/kube-logging/logging-operator/e2e/internal/kind"
3637
)
3738

39+
// clusterStopTimeout bounds the wait for cluster.Start to return after cancel.
40+
const clusterStopTimeout = time.Minute
41+
3842
type Cluster interface {
3943
cluster.Cluster
4044
LoadImages(images ...string) error
@@ -68,15 +72,25 @@ func WithCluster(name string, t *testing.T, fn func(*testing.T, Cluster), before
6872
RequireNoError(t, err)
6973

7074
ctx, cancel := context.WithCancel(context.Background())
75+
startErr := make(chan error, 1)
7176
go func() {
72-
RequireNoError(t, cluster.Start(ctx))
77+
startErr <- cluster.Start(ctx)
7378
}()
7479

7580
defer func() {
7681
assert.NoError(t, beforeCleanup(t, cluster))
7782
assert.NoError(t, cluster.Cleanup())
7883
cancel()
79-
RequireNoError(t, DeleteTestCluster(name))
84+
85+
// Checked here, not in the goroutine: FailNow is undefined off the test one.
86+
select {
87+
case err := <-startErr:
88+
assert.NoError(t, err, "starting the cluster")
89+
case <-time.After(clusterStopTimeout):
90+
assert.Fail(t, "cluster.Start did not return after cancellation")
91+
}
92+
93+
assert.NoError(t, DeleteTestCluster(name))
8094
}()
8195

8296
fn(t, cluster)
@@ -119,9 +133,17 @@ func GetTestCluster(clusterName string, opts ...cluster.Option) (Cluster, error)
119133
}
120134

121135
func DeleteTestCluster(clusterName string) error {
122-
return errors.WrapIfWithDetails(kindCLI.DeleteCluster(kind.DeleteClusterOptions{
123-
Name: clusterName,
124-
}), "deleting kind cluster", "clusterName", clusterName)
136+
kubeconfig, err := clusterKubeconfigPath(clusterName)
137+
if err != nil {
138+
return err
139+
}
140+
if err := kindCLI.DeleteCluster(kind.DeleteClusterOptions{
141+
Name: clusterName,
142+
Kubeconfig: kubeconfig,
143+
}); err != nil {
144+
return errors.WrapIfWithDetails(err, "deleting kind cluster", "clusterName", clusterName)
145+
}
146+
return removeClusterKubeconfig(clusterName)
125147
}
126148

127149
func CmdEnv(cmd *exec.Cmd, c Cluster) *exec.Cmd {

e2e/common/helpers.go

Lines changed: 1 addition & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -17,13 +17,10 @@ package common
1717
import (
1818
"context"
1919
"fmt"
20-
"os"
21-
"sync/atomic"
2220
"testing"
2321
"time"
2422

2523
"emperror.dev/errors"
26-
"github.com/spf13/cast"
2724
"github.com/stretchr/testify/assert"
2825
corev1 "k8s.io/api/core/v1"
2926
"k8s.io/apimachinery/pkg/api/resource"
@@ -48,8 +45,6 @@ const (
4845
NodeExporterTag = "local"
4946
)
5047

51-
var sequence uint32
52-
5348
func RequireNoError(t *testing.T, err error) {
5449
if err != nil {
5550
assert.Fail(t, fmt.Sprintf("Received unexpected error:\n%#v %+v", err, errors.GetDetails(err)))
@@ -58,14 +53,6 @@ func RequireNoError(t *testing.T, err error) {
5853
}
5954

6055
func Initialize(t *testing.T) {
61-
localSeq := atomic.AddUint32(&sequence, 1)
62-
shards := cast.ToUint32(os.Getenv("SHARDS"))
63-
shard := cast.ToUint32(os.Getenv("SHARD"))
64-
if shards > 0 {
65-
if localSeq%shards != shard {
66-
t.Skipf("skipping %s as sequence %d not in shard %d", t.Name(), localSeq, shard)
67-
}
68-
}
6956
t.Parallel()
7057
}
7158

@@ -228,7 +215,7 @@ func LoggingTenant(
228215
Spec: v1beta1.LoggingSpec{
229216
LoggingRef: "tenant",
230217
ControlNamespace: nsTenant,
231-
WatchNamespaces: []string{"tenant"},
218+
WatchNamespaces: []string{nsTenant},
232219
FluentdSpec: &v1beta1.FluentdSpec{
233220
Image: v1beta1.ImageSpec{
234221
Repository: FluentdImageRepo,

e2e/common/kind.go

Lines changed: 63 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,10 @@ package common
1616

1717
import (
1818
"fmt"
19+
"os"
20+
"path/filepath"
1921
"strings"
22+
"sync"
2023

2124
"emperror.dev/errors"
2225

@@ -30,17 +33,73 @@ const KindClusterCreationTimeout = "3m"
3033

3134
var kindCLI = kind.New()
3235

36+
// kubeconfigDir is one 0700 directory per run, so the paths in it are unguessable.
37+
var kubeconfigDir = sync.OnceValues(func() (string, error) {
38+
return os.MkdirTemp("", "e2e-kubeconfig-*")
39+
})
40+
41+
// clusterKubeconfigPath keeps each cluster's kind bookkeeping in its own file.
42+
// kind locks the kubeconfig it updates and the lock is non-blocking, so
43+
// clusters sharing one fail outright rather than wait.
44+
func clusterKubeconfigPath(name string) (string, error) {
45+
dir, err := kubeconfigDir()
46+
if err != nil {
47+
return "", errors.WrapIf(err, "creating the kubeconfig directory")
48+
}
49+
// Reasserted on every lookup, not left to MkdirTemp: kind recreates a missing
50+
// parent itself, at 0755, so a directory that goes away comes back readable.
51+
// MkdirAll returns nil without touching the mode of a directory that already
52+
// exists, so the Chmod is what actually puts the 0700 back.
53+
if err := os.MkdirAll(dir, 0o700); err != nil {
54+
return "", errors.WrapIfWithDetails(err, "creating the kubeconfig directory", "path", dir)
55+
}
56+
if err := os.Chmod(dir, 0o700); err != nil {
57+
return "", errors.WrapIfWithDetails(err, "restoring the kubeconfig directory mode", "path", dir)
58+
}
59+
return filepath.Join(dir, "kind-"+name+".kubeconfig"), nil
60+
}
61+
62+
// removeClusterKubeconfig drops the file and the lock kind leaves beside it,
63+
// which kind itself does not. The directory stays for the run: removing it with
64+
// one cluster only let kind recreate it at 0755 for the next.
65+
func removeClusterKubeconfig(name string) error {
66+
path, err := clusterKubeconfigPath(name)
67+
if err != nil {
68+
return err
69+
}
70+
if err := removeIfExists(path); err != nil {
71+
return err
72+
}
73+
return removeIfExists(path + ".lock")
74+
}
75+
76+
func removeIfExists(path string) error {
77+
if err := os.Remove(path); err != nil && !os.IsNotExist(err) {
78+
return errors.WrapIfWithDetails(err, "removing kubeconfig", "path", path)
79+
}
80+
return nil
81+
}
82+
3383
func KindClusterKubeconfig(name string) ([]byte, error) {
84+
kubeconfig, err := clusterKubeconfigPath(name)
85+
if err != nil {
86+
return nil, err
87+
}
88+
3489
create := kind.CreateClusterOptions{
35-
Name: name,
36-
Wait: KindClusterCreationTimeout,
90+
Name: name,
91+
Wait: KindClusterCreationTimeout,
92+
Kubeconfig: kubeconfig,
3793
}
3894

39-
err := kindCLI.CreateCluster(create)
95+
err = kindCLI.CreateCluster(create)
4096
if err != nil && isClusterAlreadyExistsError(err) {
4197
// Adopting a leftover would hand the suite an unknown operator and data.
4298
fmt.Printf("kind cluster %q already exists, recreating it\n", name)
43-
if err := kindCLI.DeleteCluster(kind.DeleteClusterOptions{Name: name}); err != nil {
99+
if err := kindCLI.DeleteCluster(kind.DeleteClusterOptions{
100+
Name: name,
101+
Kubeconfig: kubeconfig,
102+
}); err != nil {
44103
return nil, errors.WrapIfWithDetails(err, "deleting a leftover kind cluster", "clusterName", name)
45104
}
46105
err = kindCLI.CreateCluster(create)

e2e/common/kind_test.go

Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,93 @@
1+
// Copyright © 2026 Kube logging authors
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package common
16+
17+
import (
18+
"os"
19+
"path/filepath"
20+
"testing"
21+
22+
"github.com/stretchr/testify/require"
23+
)
24+
25+
func TestClusterKubeconfigPath(t *testing.T) {
26+
alpha, err := clusterKubeconfigPath("alpha")
27+
require.NoError(t, err)
28+
beta, err := clusterKubeconfigPath("beta")
29+
require.NoError(t, err)
30+
31+
require.NotEqual(t, alpha, beta)
32+
require.Equal(t, filepath.Dir(alpha), filepath.Dir(beta))
33+
34+
again, err := clusterKubeconfigPath("alpha")
35+
require.NoError(t, err)
36+
require.Equal(t, alpha, again, "create and delete have to name the same file")
37+
38+
dir := filepath.Dir(alpha)
39+
require.NotEqual(t, os.TempDir(), dir, "a path in the shared temp directory is guessable")
40+
41+
info, err := os.Stat(dir)
42+
require.NoError(t, err)
43+
require.Equal(t, os.FileMode(0o700), info.Mode().Perm())
44+
}
45+
46+
func TestRemoveClusterKubeconfig(t *testing.T) {
47+
path := stubKubeconfig(t, "gamma")
48+
lock := path + ".lock"
49+
require.NoError(t, os.WriteFile(lock, nil, 0o600))
50+
51+
require.NoError(t, removeClusterKubeconfig("gamma"))
52+
53+
require.NoFileExists(t, path)
54+
require.NoFileExists(t, lock)
55+
}
56+
57+
func TestRemoveClusterKubeconfigToleratesMissingFiles(t *testing.T) {
58+
require.NoError(t, removeClusterKubeconfig("never-created"))
59+
}
60+
61+
// kind recreates a missing parent directory itself, at 0755, so a lookup after
62+
// anything has taken the directory away has to put the 0700 back. Putting a
63+
// 0755 directory in place is the state that needs fixing: removing it instead
64+
// would only prove that MkdirAll creates a fresh one at the mode it is given.
65+
func TestClusterKubeconfigPathRestoresTheDirectory(t *testing.T) {
66+
dir := filepath.Dir(mustPath(t, "first"))
67+
require.NoError(t, os.RemoveAll(dir))
68+
require.NoError(t, os.MkdirAll(dir, 0o755))
69+
70+
path := stubKubeconfig(t, "second")
71+
require.FileExists(t, path)
72+
73+
info, err := os.Stat(filepath.Dir(path))
74+
require.NoError(t, err)
75+
require.Equal(t, os.FileMode(0o700), info.Mode().Perm())
76+
}
77+
78+
func mustPath(t *testing.T, name string) string {
79+
t.Helper()
80+
81+
path, err := clusterKubeconfigPath(name)
82+
require.NoError(t, err)
83+
return path
84+
}
85+
86+
func stubKubeconfig(t *testing.T, name string) string {
87+
t.Helper()
88+
89+
path, err := clusterKubeconfigPath(name)
90+
require.NoError(t, err)
91+
require.NoError(t, os.WriteFile(path, nil, 0o600))
92+
return path
93+
}

e2e/internal/kind/commands.go

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -111,7 +111,10 @@ func (k *Kind) CreateCluster(options CreateClusterOptions) error {
111111
case errors.Is(err, ErrTimeout):
112112
// kind removes a half-built cluster when one of its own actions fails,
113113
// but not when we kill it, so the leftovers have to go explicitly.
114-
cleanupErr := k.deleteCluster(k.CleanupTimeout, DeleteClusterOptions{Name: options.Name})
114+
cleanupErr := k.deleteCluster(k.CleanupTimeout, DeleteClusterOptions{
115+
Name: options.Name,
116+
Kubeconfig: options.Kubeconfig,
117+
})
115118
if cleanupErr != nil {
116119
return fmt.Errorf("%w; deleting the partial cluster also failed: %w", err, cleanupErr)
117120
}
@@ -133,7 +136,14 @@ func (k *Kind) DeleteCluster(options DeleteClusterOptions) error {
133136
}
134137

135138
func (k *Kind) deleteCluster(timeout time.Duration, options DeleteClusterOptions) error {
136-
return k.run(timeout, options.AppendToArgs([]string{"delete", "cluster"}), nil)
139+
cmderr := &bytes.Buffer{}
140+
err := k.run(timeout, options.AppendToArgs([]string{"delete", "cluster"}), func(cmd *exec.Cmd) {
141+
cmd.Stderr = io.MultiWriter(os.Stderr, cmderr)
142+
})
143+
if err != nil && cmderr.Len() > 0 {
144+
return fmt.Errorf("%w: %s", err, strings.TrimSpace(cmderr.String()))
145+
}
146+
return err
137147
}
138148

139149
func (k *Kind) GetKubeconfig(options GetKubeconfigOptions) ([]byte, error) {

0 commit comments

Comments
 (0)