Skip to content

Commit 2dcb42f

Browse files
utkuozdemirutkuozdemir
committed
test: drop resource cache workarounds to validate shared watch stream
Bump the etcd state to the shared watch stream experiment and remove all reads that bypass the controller runtime resource cache, so the integration tests exercise cached reads everywhere. Throwaway change to validate the consistency fix, not meant to be merged. Unit tests are expected to fail because the in-memory state used by them does not implement the shared watch stream, so the integration jobs are not gated on them on this branch. Signed-off-by: Utku Ozdemir <utku.ozdemir@siderolabs.com>
1 parent eb53209 commit 2dcb42f

15 files changed

Lines changed: 22 additions & 138 deletions

File tree

‎.github/workflows/ci.yaml‎

Lines changed: 0 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -201,7 +201,6 @@ jobs:
201201
needs:
202202
- default
203203
- lint
204-
- unit-tests
205204
steps:
206205
- name: gather-system-info
207206
id: system-info
@@ -566,7 +565,6 @@ jobs:
566565
needs:
567566
- default
568567
- lint
569-
- unit-tests
570568
steps:
571569
- name: gather-system-info
572570
id: system-info
@@ -853,7 +851,6 @@ jobs:
853851
needs:
854852
- default
855853
- lint
856-
- unit-tests
857854
steps:
858855
- name: gather-system-info
859856
id: system-info
@@ -1208,7 +1205,6 @@ jobs:
12081205
needs:
12091206
- default
12101207
- lint
1211-
- unit-tests
12121208
steps:
12131209
- name: gather-system-info
12141210
id: system-info
@@ -1498,7 +1494,6 @@ jobs:
14981494
needs:
14991495
- default
15001496
- lint
1501-
- unit-tests
15021497
steps:
15031498
- name: gather-system-info
15041499
id: system-info
@@ -1585,7 +1580,6 @@ jobs:
15851580
needs:
15861581
- default
15871582
- lint
1588-
- unit-tests
15891583
steps:
15901584
- name: gather-system-info
15911585
id: system-info

‎go.mod‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -320,3 +320,5 @@ require (
320320
sigs.k8s.io/structured-merge-diff/v6 v6.4.0 // indirect
321321
sigs.k8s.io/yaml v1.6.0 // indirect
322322
)
323+
324+
replace github.com/cosi-project/state-etcd => github.com/smira/state-etcd v0.0.0-20260803154600-abed9e9567f0

‎go.sum‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -121,8 +121,6 @@ github.com/coreos/go-systemd/v22 v22.7.0 h1:LAEzFkke61DFROc7zNLX/WA2i5J8gYqe0rSj
121121
github.com/coreos/go-systemd/v22 v22.7.0/go.mod h1:xNUYtjHu2EDXbsxz1i41wouACIwT7Ybq9o0BQhMwD0w=
122122
github.com/cosi-project/runtime v1.16.2 h1:sgjOMrnaElrja93kkJJJIQYwd45PhpGl9cBci9yqn5c=
123123
github.com/cosi-project/runtime v1.16.2/go.mod h1:+GrSnmJjMfWMe6NubevwwXQf/v7afddDLeCbLonvvps=
124-
github.com/cosi-project/state-etcd v0.7.0 h1:7EYGKbGClLA0h910oMWxtw4NZh9qt6GebaGxqGm7zts=
125-
github.com/cosi-project/state-etcd v0.7.0/go.mod h1:hplOOGHR1Rt/T8fiuFcP16thoQDvprqP/aateuCv36o=
126124
github.com/cosi-project/state-sqlite v0.4.0 h1:SwL5/LlAwnMomRjMM72igPA1F6W/88zbae4cFxqz5vw=
127125
github.com/cosi-project/state-sqlite v0.4.0/go.mod h1:V20oy2Sfxla0zZ+SJSgjV20feg2xGARlvVPL4Z4KfRo=
128126
github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g=
@@ -490,6 +488,8 @@ github.com/siderolabs/tcpproxy v0.1.0/go.mod h1:onn6CPPj/w1UNqQ0U97oRPF0CqbrgEAp
490488
github.com/sirupsen/logrus v1.8.1/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
491489
github.com/sirupsen/logrus v1.9.4 h1:TsZE7l11zFCLZnZ+teH4Umoq5BhEIfIzfRDZ1Uzql2w=
492490
github.com/sirupsen/logrus v1.9.4/go.mod h1:ftWc9WdOfJ0a92nsE2jF5u5ZwH8Bv2zdeOC42RjbV2g=
491+
github.com/smira/state-etcd v0.0.0-20260803154600-abed9e9567f0 h1:6fUPxDRfNOymYwKosfYAj/6ZWbMbrO60COp4iQ9X1/A=
492+
github.com/smira/state-etcd v0.0.0-20260803154600-abed9e9567f0/go.mod h1:+I+tyOd7uBofnxloBn66Qc3P+oleRgD6qOeO+rkF/pU=
493493
github.com/soheilhy/cmux v0.1.5 h1:jjzc5WVemNEDTLwv9tlmemhC73tI08BNOIGwBOo10Js=
494494
github.com/soheilhy/cmux v0.1.5/go.mod h1:T7TcVDs9LWfQgPlPsdngu6I6QIoyIFZDDC6sNE1GqG0=
495495
github.com/sosodev/duration v1.4.0 h1:35ed0KiVFriGHHzZZJaZLgmTEEICIyt8Sx0RQfj9IjE=

‎internal/backend/runtime/omni/controllers/cleanup/id.go‎

Lines changed: 2 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -33,9 +33,7 @@ func (h *SameIDHandler[I, O]) FinalizerRemoval(ctx context.Context, r controller
3333
resource.VersionUndefined,
3434
)
3535

36-
// Use GetUncached to bypass the controller runtime cache. A cached miss here would cause the handler to skip cleanup permanently (the finalizer gets removed
37-
// and the orphaned resource is never retried).
38-
res, err := r.GetUncached(ctx, md)
36+
res, err := r.Get(ctx, md)
3937
if err != nil {
4038
if state.IsNotFoundError(err) {
4139
return nil
@@ -101,9 +99,7 @@ func IDHandleFunc[I, O generic.ResourceWithRD](getIDFunc GetIDFunc[I], blockIfOw
10199
resource.VersionUndefined,
102100
)
103101

104-
// Use GetUncached to bypass the controller runtime cache. A cached miss here would cause the handler to skip cleanup permanently (the finalizer gets removed
105-
// and the orphaned resource is never retried).
106-
res, err := r.GetUncached(ctx, md)
102+
res, err := r.Get(ctx, md)
107103
if err != nil {
108104
if state.IsNotFoundError(err) {
109105
return nil

‎internal/backend/runtime/omni/controllers/omni/machine_set_node.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -223,7 +223,7 @@ func (ctrl *MachineSetNodeController) MapInput(
223223
}
224224

225225
func (ctrl *MachineSetNodeController) getAllMachineSetNodes(ctx context.Context, r controller.QRuntime, opts ...state.ListOption) (safe.List[*omni.MachineSetNode], error) {
226-
items, err := r.ListUncached(
226+
items, err := r.List(
227227
ctx, resource.NewMetadata(resources.DefaultNamespace, omni.MachineSetNodeType, "", resource.VersionUndefined),
228228
opts...,
229229
)

‎internal/backend/runtime/omni/controllers/omni/machineconfig/reconciliation_context.go‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,6 @@ import (
3232
"github.com/siderolabs/omni/client/pkg/omni/resources/infra"
3333
"github.com/siderolabs/omni/client/pkg/omni/resources/omni"
3434
siderolinkres "github.com/siderolabs/omni/client/pkg/omni/resources/siderolink"
35-
"github.com/siderolabs/omni/internal/backend/runtime/omni/controllers/uncached"
3635
"github.com/siderolabs/omni/internal/backend/talos/lifecycle"
3736
)
3837

@@ -267,8 +266,7 @@ func BuildReconciliationContext(ctx context.Context, r controller.Reader,
267266
return nil, errors.New("failed to get machine set name from the machine config resource")
268267
}
269268

270-
// A stale allow value could apply a config while updates are blocked.
271-
machineSetConfigStatus, err := safe.ReaderGetByID[*omni.MachineSetConfigStatus](ctx, uncached.Reader(r), machineSetName)
269+
machineSetConfigStatus, err := safe.ReaderGetByID[*omni.MachineSetConfigStatus](ctx, r, machineSetName)
272270
if err != nil && !state.IsNotFoundError(err) {
273271
return nil, err
274272
}

‎internal/backend/runtime/omni/controllers/omni/machineconfig/status.go‎

Lines changed: 5 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,6 @@ import (
4343
"github.com/siderolabs/omni/internal/backend/runtime/omni/controllers/helpers"
4444
"github.com/siderolabs/omni/internal/backend/runtime/omni/controllers/omni/internal/mappers"
4545
talosutils "github.com/siderolabs/omni/internal/backend/runtime/omni/controllers/omni/internal/talos"
46-
"github.com/siderolabs/omni/internal/backend/runtime/omni/controllers/uncached"
4746
"github.com/siderolabs/omni/internal/backend/runtime/talos"
4847
"github.com/siderolabs/omni/internal/backend/talos/lifecycle"
4948
)
@@ -895,11 +894,9 @@ func (ctrl *StatusController) etcdForfeitHook(rc *ReconciliationContext, cluster
895894
}
896895
}
897896

898-
// listClusterMachines lists the cluster's ClusterMachines uncached, so callers see the freshest finalizer set.
897+
// listClusterMachines lists the cluster's ClusterMachines.
899898
func (ctrl *StatusController) listClusterMachines(ctx context.Context, r controller.ReaderWriter, clusterName string) ([]resource.Resource, error) {
900-
qruntime := r.(controller.QRuntime) //nolint:forcetypeassert,errcheck
901-
902-
list, err := qruntime.ListUncached(
899+
list, err := r.List(
903900
ctx,
904901
omni.NewClusterMachine("").Metadata(),
905902
state.WithLabelQuery(resource.LabelEqual(omni.LabelCluster, clusterName)),
@@ -1253,9 +1250,7 @@ func (ctrl *StatusController) shouldReset(
12531250
return false, fmt.Errorf("finalizer: failed to get cluster status '%s': %w", clusterName, err)
12541251
}
12551252

1256-
// Read the Machine uncached: a cached read can briefly still report a Machine that is being torn
1257-
// down or destroyed as present and running, so the reset decision below could be made on stale data.
1258-
machine, err := safe.ReaderGetByID[*omni.Machine](ctx, uncached.Reader(r), machineID)
1253+
machine, err := safe.ReaderGetByID[*omni.Machine](ctx, r, machineID)
12591254
if err != nil {
12601255
if state.IsNotFoundError(err) {
12611256
return false, nil
@@ -1459,15 +1454,12 @@ func (ctrl *StatusController) acquireConfigUpdateLock(ctx context.Context, r con
14591454
return errors.New("failed to get machine set name from the cluster machine")
14601455
}
14611456

1462-
// A stale strategy could exceed the configured update parallelism.
1463-
machineSetConfigStatus, err := safe.ReaderGetByID[*omni.MachineSetConfigStatus](ctx, uncached.Reader(r), machineSetName)
1457+
machineSetConfigStatus, err := safe.ReaderGetByID[*omni.MachineSetConfigStatus](ctx, r, machineSetName)
14641458
if err != nil {
14651459
return err
14661460
}
14671461

1468-
qruntime := r.(controller.QRuntime) //nolint:forcetypeassert,errcheck
1469-
1470-
clusterMachines, err := qruntime.ListUncached(
1462+
clusterMachines, err := r.List(
14711463
ctx,
14721464
omni.NewClusterMachine("").Metadata(),
14731465
state.WithLabelQuery(resource.LabelEqual(omni.LabelMachineSet, machineSetName)),

‎internal/backend/runtime/omni/controllers/omni/maintenance_config_status.go‎

Lines changed: 2 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -41,7 +41,6 @@ import (
4141
siderolinkres "github.com/siderolabs/omni/client/pkg/omni/resources/siderolink"
4242
"github.com/siderolabs/omni/client/pkg/siderolink"
4343
"github.com/siderolabs/omni/internal/backend/runtime/omni/controllers/omni/internal/imagefactoryauth"
44-
"github.com/siderolabs/omni/internal/backend/runtime/omni/controllers/uncached"
4544
"github.com/siderolabs/omni/internal/backend/runtime/talos"
4645
)
4746

@@ -278,8 +277,7 @@ func (helper *maintenanceConfigStatusControllerHelper) transform(ctx context.Con
278277
}
279278

280279
// wait for the machine's incoming config to be extracted before applying anything, otherwise we could overwrite (and lose) the config the machine arrived with before it is preserved.
281-
// read uncached, to avoid acting on a stale (or not yet visible) extraction status.
282-
extractionStatus, err := safe.ReaderGetByID[*omni.MachineConfigExtractionStatus](ctx, uncached.Reader(r), link.Metadata().ID())
280+
extractionStatus, err := safe.ReaderGetByID[*omni.MachineConfigExtractionStatus](ctx, r, link.Metadata().ID())
283281
if err != nil && !state.IsNotFoundError(err) {
284282
return err
285283
}
@@ -288,8 +286,7 @@ func (helper *maintenanceConfigStatusControllerHelper) transform(ctx context.Con
288286
return xerrors.NewTaggedf[qtransform.SkipReconcileTag]("machine config has not been extracted yet")
289287
}
290288

291-
// read machine config patches uncached, to avoid acting on a stale (or not yet visible) just-extracted preserved config patch
292-
machinePatches, err := getMachinePatches(ctx, uncached.Reader(r), link.Metadata().ID())
289+
machinePatches, err := getMachinePatches(ctx, r, link.Metadata().ID())
293290
if err != nil {
294291
return fmt.Errorf("error collecting machine config patches: %w", err)
295292
}

‎internal/backend/runtime/omni/controllers/omni/node_unique_token_cleanup.go‎

Lines changed: 1 addition & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,6 @@ import (
2121
"github.com/siderolabs/omni/client/pkg/cosi/helpers"
2222
"github.com/siderolabs/omni/client/pkg/omni/resources"
2323
"github.com/siderolabs/omni/client/pkg/omni/resources/siderolink"
24-
"github.com/siderolabs/omni/internal/backend/runtime/omni/controllers/uncached"
2524
)
2625

2726
// NodeUniqueTokenCleanupControllerName is the name of the NodeUniqueTokenCleanupController.
@@ -102,10 +101,7 @@ func (ctrl *NodeUniqueTokenCleanupController) Reconcile(ctx context.Context,
102101
}
103102

104103
func (ctrl *NodeUniqueTokenCleanupController) reconcileRunning(ctx context.Context, r controller.QRuntime, logger *zap.Logger, nodeUniqueToken *siderolink.NodeUniqueToken) error {
105-
// Read the link uncached. The cached input replica is fed by a separate watch stream and can lag
106-
// real state, so a link that already exists may not yet be visible when the token's reconcile fires.
107-
// Acting on a stale "no link" view here would wrongly tear down a token that actually has a link.
108-
link, err := safe.ReaderGetByID[*siderolink.Link](ctx, uncached.Reader(r), nodeUniqueToken.Metadata().ID())
104+
link, err := safe.ReaderGetByID[*siderolink.Link](ctx, r, nodeUniqueToken.Metadata().ID())
109105
if err != nil && !state.IsNotFoundError(err) {
110106
return err
111107
}

‎internal/backend/runtime/omni/controllers/omni/node_unique_token_cleanup_test.go‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -43,8 +43,7 @@ func (suite *MachineStatusSnapshotControllerSuite) TestNodeUniqueTokenCleanup()
4343

4444
link := siderolink.NewLink(token.Metadata().ID(), &specs.SiderolinkSpec{})
4545

46-
// Create the link before recreating the token. The controller reads the link uncached, so an
47-
// existing link is always observed and the token is never treated as orphaned, even on a busy runner.
46+
// Create the link before recreating the token, so the token is never treated as orphaned.
4847
suite.Require().NoError(suite.state.Create(ctx, link))
4948
suite.Require().NoError(suite.state.Create(ctx, token))
5049

0 commit comments

Comments
 (0)