Skip to content

Commit d16b15f

Browse files
authored
Merge pull request kcp-dev#3984 from ntnn/fix-cache-flakes
Reset REST mapper on retr, log errors and use embeddedetcd.Server.Stopped
2 parents 810d85a + c66dbac commit d16b15f

6 files changed

Lines changed: 85 additions & 14 deletions

File tree

go.mod

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@ require (
2222
github.com/kcp-dev/apimachinery/v2 v2.29.1-0.20260223112726-38863b7c2c8e
2323
github.com/kcp-dev/client-go v0.28.1-0.20260223113551-9948318ac6d7
2424
github.com/kcp-dev/code-generator/v3 v3.0.0-00010101000000-000000000000
25-
github.com/kcp-dev/embeddedetcd v1.1.0
25+
github.com/kcp-dev/embeddedetcd v1.1.1-0.20260402110232-2cc5c5cce35e
2626
github.com/kcp-dev/logicalcluster/v3 v3.0.5
2727
github.com/kcp-dev/sdk v0.0.0
2828
github.com/kcp-dev/virtual-workspace-framework v0.0.0

go.sum

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -119,8 +119,8 @@ github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8Hm
119119
github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y=
120120
github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
121121
github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
122-
github.com/kcp-dev/embeddedetcd v1.1.0 h1:4u5BwZdD43rMnZc3VOpj/VS/+WgJDknv1wuyy4rVkzM=
123-
github.com/kcp-dev/embeddedetcd v1.1.0/go.mod h1:KNR9s+3UcXtNamapwLH7M/8ZMZCHpIU5YalOAEjOwJg=
122+
github.com/kcp-dev/embeddedetcd v1.1.1-0.20260402110232-2cc5c5cce35e h1:nOcf6Id7wozcS2pwoNNmuKqIozAAn/K+0wu3z7xGxE0=
123+
github.com/kcp-dev/embeddedetcd v1.1.1-0.20260402110232-2cc5c5cce35e/go.mod h1:bv5D/GGLuILI1Y5tZzbBvzOjkdixKS9lM+bbN1vWZTE=
124124
github.com/kcp-dev/kubernetes v0.0.0-20260317180031-228ac1412525 h1:Ix6xWkzxeB3zD9d5BcBC1cjz2ELh7/tb3btrZdooJWQ=
125125
github.com/kcp-dev/kubernetes v0.0.0-20260317180031-228ac1412525/go.mod h1:OWamAtYc/9r2ICXEDBDdH6Oh6X0m6UX6QYsdn2O3QE0=
126126
github.com/kcp-dev/kubernetes/staging/src/k8s.io/api v0.0.0-20260317180031-228ac1412525 h1:3TIuylu5r8EeWAJCQxqTZrCM0sXqeCCjGGCb66O/8Lo=

staging/src/github.com/kcp-dev/sdk/testing/server/ready.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -53,7 +53,7 @@ func WaitForReady(ctx context.Context, cfg *rest.Config) error {
5353

5454
func waitForEndpoint(ctx context.Context, client *rest.RESTClient, endpoint string) error {
5555
var lastError error
56-
if err := wait.PollUntilContextTimeout(ctx, 100*time.Millisecond, wait.ForeverTestTimeout, true, func(ctx context.Context) (bool, error) {
56+
if err := wait.PollUntilContextTimeout(ctx, 100*time.Millisecond, time.Minute, true, func(ctx context.Context) (bool, error) {
5757
req := rest.NewRequest(client).RequestURI(endpoint)
5858
if _, err := req.Do(ctx).Raw(); err != nil {
5959
lastError = fmt.Errorf("error contacting %s: failed components: %v", req.URL(), unreadyComponentsFromError(err))

test/e2e/cache/replication_api_cache_test.go

Lines changed: 12 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -103,14 +103,22 @@ func TestCacheServerReplicationAPI(t *testing.T) {
103103

104104
t.Logf("Create a instances object in workspace %q", rootOrg)
105105
kcptestinghelpers.Eventually(t, func() (bool, string) {
106-
err = helpers.CreateResourceFromFS(ctx, dynamicClusterClient.Cluster(rootOrg), mapper, nil, "assets/instances.yaml", testFiles)
107-
return err == nil, ""
106+
err := helpers.CreateResourceFromFS(ctx, dynamicClusterClient.Cluster(rootOrg), mapper, nil, "assets/instances.yaml", testFiles)
107+
if err != nil {
108+
mapper.Reset()
109+
return false, err.Error()
110+
}
111+
return true, ""
108112
}, wait.ForeverTestTimeout, 100*time.Millisecond, "waiting for instances object to be created")
109113

110114
t.Logf("Create a publishedResource in workspace %q", rootOrg)
111115
kcptestinghelpers.Eventually(t, func() (bool, string) {
112-
err = helpers.CreateResourceFromFS(ctx, dynamicClusterClient.Cluster(rootOrg), mapper, nil, "assets/published-resource-instances.yaml", testFiles)
113-
return err == nil, ""
116+
err := helpers.CreateResourceFromFS(ctx, dynamicClusterClient.Cluster(rootOrg), mapper, nil, "assets/published-resource-instances.yaml", testFiles)
117+
if err != nil {
118+
mapper.Reset()
119+
return false, err.Error()
120+
}
121+
return true, ""
114122
}, wait.ForeverTestTimeout, 100*time.Millisecond, "waiting for publishedResource to be created")
115123

116124
t.Logf("Wait for replication api to be ready and reporting 7 replicas")

test/e2e/reconciler/cache/helpers.go

Lines changed: 40 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -93,19 +93,56 @@ func StartStandaloneCacheServer(ctx context.Context, t *testing.T, dataDir strin
9393
cacheServerCompletedConfig, err := cacheServerConfig.Complete()
9494
require.NoError(t, err)
9595

96+
// When the test ends the cache server is stopped.
97+
// For the cache server to stop gracefully etcd must be available until it is shut down.
98+
// For etcd to then gracefully stop the files on disk must be present until etcd is shut down.
99+
// If the cache server/etcd are bounded by the context the server
100+
// will shut down in parallel while the test harness is deleting
101+
// files, which leads to spurious opaque test errors (sometimes in
102+
// _other_ tests).
103+
// So the shutdown _must_ be:
104+
// 1. Stop cache server
105+
// 2. Stop etcd
106+
// 3. Let test cleanup run
107+
serverCtx, serverCancel := context.WithCancel(ctx)
108+
etcdCtx, etcdCancel := context.WithCancel(context.Background())
109+
96110
if cacheServerCompletedConfig.EmbeddedEtcd.Config != nil {
97111
t.Logf("Starting embedded etcd for the cache server")
98-
require.NoError(t, embeddedetcd.NewServer(cacheServerCompletedConfig.EmbeddedEtcd).Run(ctx))
112+
113+
etcdserver := embeddedetcd.NewServer(cacheServerCompletedConfig.EmbeddedEtcd)
114+
115+
if err := etcdserver.Run(etcdCtx); err != nil {
116+
etcdCancel()
117+
t.Fatalf("failed to start embedded etcd: %v", err)
118+
}
119+
120+
t.Cleanup(func() {
121+
<-etcdCtx.Done()
122+
if err := wait.PollUntilContextTimeout(context.Background(), time.Second, wait.ForeverTestTimeout, true,
123+
func(ctx context.Context) (bool, error) {
124+
return etcdserver.Stopped(), nil
125+
},
126+
); err != nil {
127+
t.Fatal(err)
128+
}
129+
})
99130
}
131+
100132
cacheServer, err := cacheserver.NewServer(cacheServerCompletedConfig)
101133
require.NoError(t, err)
102-
preparedCachedServer, err := cacheServer.PrepareRun(ctx)
134+
preparedCachedServer, err := cacheServer.PrepareRun(serverCtx)
103135
require.NoError(t, err)
104136
start := time.Now()
137+
105138
t.Logf("Starting the cache server")
106139
go func() {
107-
assert.NoError(t, preparedCachedServer.Run(ctx))
140+
// stop etcd server after cache server is done shutting down
141+
defer etcdCancel()
142+
assert.NoError(t, preparedCachedServer.Run(serverCtx))
108143
}()
144+
// Stop cache server before starting other cleanup functions
145+
t.Cleanup(serverCancel)
109146

110147
cacheServerCertificatePath := path.Join(dataDir, "cache", "apiserver.crt")
111148
kcptestinghelpers.Eventually(t, func() (bool, string) {

test/integration/framework/server.go

Lines changed: 29 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -19,12 +19,16 @@ package framework
1919
import (
2020
"context"
2121
"fmt"
22+
"log"
2223
"sync"
24+
"time"
2325

2426
"github.com/spf13/pflag"
27+
"github.com/stretchr/testify/require"
2528

2629
"k8s.io/apimachinery/pkg/runtime"
2730
utilerrors "k8s.io/apimachinery/pkg/util/errors"
31+
"k8s.io/apimachinery/pkg/util/wait"
2832
"k8s.io/client-go/rest"
2933
"k8s.io/client-go/tools/clientcmd"
3034
clientcmdapi "k8s.io/client-go/tools/clientcmd/api"
@@ -65,6 +69,8 @@ type InProcessServer struct {
6569

6670
loadCfgOnce sync.Once
6771
ClientConfig clientcmd.ClientConfig
72+
73+
embeddedetcd *embeddedetcd.Server
6874
}
6975

7076
func NewInProcessServer(t kcptestingserver.TestingT, opts ...kcptestingserver.Option) *InProcessServer {
@@ -141,11 +147,22 @@ func (s *InProcessServer) Start(ctx context.Context, t kcptestingserver.TestingT
141147

142148
etcdCtx, etcdCancel := context.WithCancel(context.Background())
143149

144-
// the etcd server must be up before NewServer because storage decorators access it right away
150+
// When the test ends kcp is stopped.
151+
// For kcp to stop gracefully etcd must be available until kcp is shut down.
152+
// For etcd to then gracefully stop the files on disk must be present until etcd is shut down.
153+
// If kcp/etcd are bounded by the context the server will shut down
154+
// in parallel while the test harness is deleting files, which leads
155+
// to spurious opaque test errors (sometimes in _other_ tests).
156+
// So the shutdown _must_ be:
157+
// 1. Stop kcp
158+
// 2. Stop etcd
159+
// 3. Let test cleanup run
160+
145161
if completedConfig.EmbeddedEtcd.Config != nil {
146-
if err := embeddedetcd.NewServer(completedConfig.EmbeddedEtcd).Run(etcdCtx); err != nil {
162+
s.embeddedetcd = embeddedetcd.NewServer(completedConfig.EmbeddedEtcd)
163+
if err := s.embeddedetcd.Run(etcdCtx); err != nil {
147164
etcdCancel()
148-
t.Fatalf("failed to start embedded etcd: %v", err)
165+
require.NoError(t, err)
149166
}
150167
}
151168

@@ -177,6 +194,15 @@ func (s *InProcessServer) Stop() {
177194
s.cancel = nil
178195
})
179196
<-s.StopCh
197+
if s.embeddedetcd != nil {
198+
if err := wait.PollUntilContextTimeout(context.Background(), time.Second, wait.ForeverTestTimeout, true,
199+
func(ctx context.Context) (bool, error) {
200+
return s.embeddedetcd.Stopped(), nil
201+
},
202+
); err != nil {
203+
log.Printf("embedded etcd did not stop in timeout: %v", err)
204+
}
205+
}
180206
}
181207

182208
func (s *InProcessServer) Stopped() bool {

0 commit comments

Comments
 (0)