Skip to content

Commit 02ff20d

Browse files
teoincontattoteoincontatto
committed
fix: separate resources cache by resource that generates them to avoid stale resources
1 parent 3682da3 commit 02ff20d

9 files changed

Lines changed: 263 additions & 55 deletions

File tree

‎stackgres-k8s/src/operator/src/main/java/io/stackgres/operator/app/DefaultOperatorWatchersHandler.java‎

Lines changed: 183 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,8 @@
3232
import io.stackgres.common.crd.sgcluster.StackGresClusterBackupConfiguration;
3333
import io.stackgres.common.crd.sgcluster.StackGresClusterConfigurations;
3434
import io.stackgres.common.crd.sgcluster.StackGresClusterList;
35+
import io.stackgres.common.crd.sgcluster.StackGresClusterManagedScriptEntry;
36+
import io.stackgres.common.crd.sgcluster.StackGresClusterManagedSql;
3537
import io.stackgres.common.crd.sgconfig.StackGresConfig;
3638
import io.stackgres.common.crd.sgconfig.StackGresConfigList;
3739
import io.stackgres.common.crd.sgdbops.StackGresDbOps;
@@ -46,6 +48,8 @@
4648
import io.stackgres.common.crd.sgpooling.StackGresPoolingConfigList;
4749
import io.stackgres.common.crd.sgprofile.StackGresProfile;
4850
import io.stackgres.common.crd.sgprofile.StackGresProfileList;
51+
import io.stackgres.common.crd.sgscript.StackGresScript;
52+
import io.stackgres.common.crd.sgscript.StackGresScriptList;
4953
import io.stackgres.common.crd.sgshardedbackup.StackGresShardedBackup;
5054
import io.stackgres.common.crd.sgshardedbackup.StackGresShardedBackupList;
5155
import io.stackgres.common.crd.sgshardedcluster.StackGresShardedCluster;
@@ -57,6 +61,7 @@
5761
import io.stackgres.common.crd.sgstream.StackGresStream;
5862
import io.stackgres.common.crd.sgstream.StackGresStreamList;
5963
import io.stackgres.operator.common.ResourceWatcherFactory;
64+
import io.stackgres.operator.conciliation.DeployedResourcesCache;
6065
import io.stackgres.operator.conciliation.backup.BackupReconciliator;
6166
import io.stackgres.operator.conciliation.cluster.ClusterReconciliator;
6267
import io.stackgres.operator.conciliation.config.ConfigReconciliator;
@@ -107,6 +112,7 @@ public class DefaultOperatorWatchersHandler implements OperatorWatchersHandler {
107112
Collections.synchronizedMap(new HashMap<>());
108113
private final Map<String, StackGresStream> streams =
109114
Collections.synchronizedMap(new HashMap<>());
115+
private final DeployedResourcesCache deployedResourcesCache;
110116

111117
@Inject
112118
public DefaultOperatorWatchersHandler(
@@ -120,8 +126,8 @@ public DefaultOperatorWatchersHandler(
120126
ShardedBackupReconciliator shardedBackupReconciliatorCycle,
121127
ShardedDbOpsReconciliator shardedDbOpsReconciliatorCycle,
122128
StreamReconciliator streamReconciliatorCycle,
123-
ResourceWatcherFactory watcherFactory) {
124-
super();
129+
ResourceWatcherFactory watcherFactory,
130+
DeployedResourcesCache deployedResourcesCache) {
125131
this.client = client;
126132
this.configReconciliatorCycle = configReconciliatorCycle;
127133
this.clusterReconciliatorCycle = clusterReconciliatorCycle;
@@ -133,23 +139,28 @@ public DefaultOperatorWatchersHandler(
133139
this.shardedDbOpsReconciliatorCycle = shardedDbOpsReconciliatorCycle;
134140
this.streamReconciliatorCycle = streamReconciliatorCycle;
135141
this.watcherFactory = watcherFactory;
142+
this.deployedResourcesCache = deployedResourcesCache;
136143
}
137144

138145
@Override
139146
public void startWatchers() {
140147
monitors.addAll(createCustomResourceWatchers(
141148
StackGresConfig.class,
142149
StackGresConfigList.class,
143-
onCreateOrUpdate(
150+
onCreateOrUpdateAndOnDelete(
144151
reconcileConfig()
145-
.andThen(putConfig()))));
152+
.andThen(putConfig()),
153+
invalidateConfig()
154+
.andThen(removeConfig()))));
146155

147156
monitors.addAll(createCustomResourceWatchers(
148157
StackGresCluster.class,
149158
StackGresClusterList.class,
150-
onCreateOrUpdate(
159+
onCreateOrUpdateAndOnDelete(
151160
putCluster()
152-
.andThen(reconcileCluster()))));
161+
.andThen(reconcileCluster()),
162+
invalidateCluster()
163+
.andThen(removeCluster()))));
153164

154165
monitors.addAll(createCustomResourceWatchers(
155166
StackGresProfile.class,
@@ -180,54 +191,75 @@ public void startWatchers() {
180191
reconcileObjectStorageClusters()
181192
.andThen(reconcileObjectStorageShardedClusters()))));
182193

194+
monitors.addAll(createCustomResourceWatchers(
195+
StackGresScript.class,
196+
StackGresScriptList.class,
197+
onCreateOrUpdate(
198+
reconcileScriptClusters()
199+
.andThen(reconcileScriptShardedClusters()))));
200+
183201
monitors.addAll(createCustomResourceWatchers(
184202
StackGresBackup.class,
185203
StackGresBackupList.class,
186-
onCreateOrUpdate(
204+
onCreateOrUpdateAndOnDelete(
187205
putBackup()
188-
.andThen(reconcileBackup()))));
206+
.andThen(reconcileBackup()),
207+
invalidateBackup()
208+
.andThen(removeBackup()))));
189209

190210
monitors.addAll(createCustomResourceWatchers(
191211
StackGresDbOps.class,
192212
StackGresDbOpsList.class,
193-
onCreateOrUpdate(
213+
onCreateOrUpdateAndOnDelete(
194214
putDbOps()
195-
.andThen(reconcileDbOps()))));
215+
.andThen(reconcileDbOps()),
216+
invalidateDbOps()
217+
.andThen(removeDbOps()))));
196218

197219
monitors.addAll(createCustomResourceWatchers(
198220
StackGresDistributedLogs.class,
199221
StackGresDistributedLogsList.class,
200-
onCreateOrUpdate(
222+
onCreateOrUpdateAndOnDelete(
201223
putDistributedLogs()
202-
.andThen(reconcileDistributedLogs()))));
224+
.andThen(reconcileDistributedLogs()),
225+
invalidateDistributedLogs()
226+
.andThen(removeDistributedLogs()))));
203227

204228
monitors.addAll(createCustomResourceWatchers(
205229
StackGresShardedCluster.class,
206230
StackGresShardedClusterList.class,
207-
onCreateOrUpdate(
231+
onCreateOrUpdateAndOnDelete(
208232
putShardedCluster()
209-
.andThen(reconcileShardedCluster()))));
233+
.andThen(reconcileShardedCluster()),
234+
invalidateShardedCluster()
235+
.andThen(removeShardedCluster()))));
210236

211237
monitors.addAll(createCustomResourceWatchers(
212238
StackGresShardedBackup.class,
213239
StackGresShardedBackupList.class,
214-
onCreateOrUpdate(
240+
onCreateOrUpdateAndOnDelete(
215241
putShardedBackup()
216-
.andThen(reconcileShardedBackup()))));
242+
.andThen(reconcileShardedBackup()),
243+
invalidateShardedBackup()
244+
.andThen(removeShardedBackup()))));
217245

218246
monitors.addAll(createCustomResourceWatchers(
219247
StackGresShardedDbOps.class,
220248
StackGresShardedDbOpsList.class,
221-
onCreateOrUpdate(
249+
onCreateOrUpdateAndOnDelete(
222250
putShardedDbOps()
223-
.andThen(reconcileShardedDbOps()))));
251+
.andThen(reconcileShardedDbOps()),
252+
invalidateShardedDbOps()
253+
.andThen(removeShardedDbOps()))));
224254

225255
monitors.addAll(createCustomResourceWatchers(
226256
StackGresStream.class,
227257
StackGresStreamList.class,
228-
onCreateOrUpdate(
258+
onCreateOrUpdateAndOnDelete(
229259
putStream()
230-
.andThen(reconcileStream()))));
260+
.andThen(reconcileStream()),
261+
invalidateStream()
262+
.andThen(removeStream()))));
231263

232264
monitors.addAll(createWatchers(
233265
Endpoints.class,
@@ -300,6 +332,19 @@ private <T> BiConsumer<Action, T> onCreateOrUpdate(BiConsumer<Action, T> consume
300332
};
301333
}
302334

335+
private <T> BiConsumer<Action, T> onCreateOrUpdateAndOnDelete(
336+
BiConsumer<Action, T> consumerOnCreateOrUpdate,
337+
BiConsumer<Action, T> consumerOnDelete) {
338+
return (action, resource) -> {
339+
if (action == Action.ADDED || action == Action.MODIFIED) {
340+
consumerOnCreateOrUpdate.accept(action, resource);
341+
}
342+
if (action == Action.DELETED) {
343+
consumerOnDelete.accept(action, resource);
344+
}
345+
};
346+
}
347+
303348
private <T> BiConsumer<Action, T> onCreateOrUpdateOrDelete(BiConsumer<Action, T> consumer) {
304349
return (action, resource) -> {
305350
if (action == Action.ADDED || action == Action.MODIFIED || action == Action.DELETED) {
@@ -349,6 +394,43 @@ private BiConsumer<Action, StackGresStream> putStream() {
349394
return (action, stream) -> this.streams.put(resourceId(stream), stream);
350395
}
351396

397+
private BiConsumer<Action, StackGresConfig> removeConfig() {
398+
return (action, config) -> configs.remove(resourceId(config));
399+
}
400+
401+
private BiConsumer<Action, StackGresCluster> removeCluster() {
402+
return (action, cluster) -> clusters.remove(resourceId(cluster));
403+
}
404+
405+
private BiConsumer<Action, StackGresDistributedLogs> removeDistributedLogs() {
406+
return (action, distributedLogs) -> this.distributedLogs
407+
.remove(resourceId(distributedLogs));
408+
}
409+
410+
private BiConsumer<Action, StackGresBackup> removeBackup() {
411+
return (action, backup) -> backups.remove(resourceId(backup));
412+
}
413+
414+
private BiConsumer<Action, StackGresDbOps> removeDbOps() {
415+
return (action, dbOps) -> this.dbOps.remove(resourceId(dbOps));
416+
}
417+
418+
private BiConsumer<Action, StackGresShardedCluster> removeShardedCluster() {
419+
return (action, cluster) -> shardedClusters.remove(resourceId(cluster));
420+
}
421+
422+
private BiConsumer<Action, StackGresShardedBackup> removeShardedBackup() {
423+
return (action, backup) -> shardedBackups.remove(resourceId(backup));
424+
}
425+
426+
private BiConsumer<Action, StackGresShardedDbOps> removeShardedDbOps() {
427+
return (action, dbOps) -> shardedDbOps.remove(resourceId(dbOps));
428+
}
429+
430+
private BiConsumer<Action, StackGresStream> removeStream() {
431+
return (action, stream) -> this.streams.remove(resourceId(stream));
432+
}
433+
352434
private BiConsumer<Action, StackGresConfig> reconcileConfig() {
353435
return (action, config) -> Optional
354436
.ofNullable(configs.get(resourceId(config)))
@@ -445,6 +527,21 @@ private BiConsumer<Action, StackGresObjectStorage> reconcileObjectStorageCluster
445527
.forEach(cluster -> reconcileCluster().accept(action, cluster));
446528
}
447529

530+
private BiConsumer<Action, StackGresScript> reconcileScriptClusters() {
531+
return (action, script) -> synchronizedCopyOfValues(clusters)
532+
.stream()
533+
.filter(cluster -> Objects.equals(
534+
cluster.getMetadata().getNamespace(),
535+
script.getMetadata().getNamespace()))
536+
.filter(cluster -> Optional.ofNullable(cluster.getSpec().getManagedSql())
537+
.map(StackGresClusterManagedSql::getScripts)
538+
.stream()
539+
.flatMap(List::stream)
540+
.map(StackGresClusterManagedScriptEntry::getSgScript)
541+
.anyMatch(script.getMetadata().getName()::equals))
542+
.forEach(cluster -> reconcileCluster().accept(action, cluster));
543+
}
544+
448545
private BiConsumer<Action, StackGresProfile> reconcileInstanceProfileDistributedLogs() {
449546
return (action, instanceProfile) -> synchronizedCopyOfValues(distributedLogs)
450547
.stream()
@@ -559,6 +656,36 @@ private BiConsumer<Action, StackGresObjectStorage> reconcileObjectStorageSharded
559656
.forEach(shardedCluster -> reconcileShardedCluster().accept(action, shardedCluster));
560657
}
561658

659+
private BiConsumer<Action, StackGresScript> reconcileScriptShardedClusters() {
660+
return (action, script) -> synchronizedCopyOfValues(shardedClusters)
661+
.stream()
662+
.filter(cluster -> Objects.equals(
663+
cluster.getMetadata().getNamespace(),
664+
script.getMetadata().getNamespace()))
665+
.filter(cluster -> Optional.ofNullable(cluster.getSpec().getCoordinator().getManagedSql())
666+
.map(StackGresClusterManagedSql::getScripts)
667+
.stream()
668+
.flatMap(List::stream)
669+
.map(StackGresClusterManagedScriptEntry::getSgScript)
670+
.anyMatch(script.getMetadata().getName()::equals)
671+
|| Optional.ofNullable(cluster.getSpec().getShards().getManagedSql())
672+
.map(StackGresClusterManagedSql::getScripts)
673+
.stream()
674+
.flatMap(List::stream)
675+
.map(StackGresClusterManagedScriptEntry::getSgScript)
676+
.anyMatch(script.getMetadata().getName()::equals)
677+
|| Optional.ofNullable(cluster.getSpec().getShards().getOverrides())
678+
.stream()
679+
.flatMap(List::stream)
680+
.flatMap(override -> Optional.ofNullable(override.getManagedSql())
681+
.map(StackGresClusterManagedSql::getScripts)
682+
.stream())
683+
.flatMap(List::stream)
684+
.map(StackGresClusterManagedScriptEntry::getSgScript)
685+
.anyMatch(script.getMetadata().getName()::equals))
686+
.forEach(cluster -> reconcileShardedCluster().accept(action, cluster));
687+
}
688+
562689
private BiConsumer<Action, Endpoints> reconcileEndpointsShardedClusters() {
563690
String clusterScopeKey =
564691
StackGresContext.STACKGRES_KEY_PREFIX + StackGresContext.CLUSTER_SCOPE_KEY;
@@ -711,6 +838,42 @@ private BiConsumer<Action, Pod> reconcilePodStreams() {
711838
.forEach(stream -> reconcileStream().accept(action, stream));
712839
}
713840

841+
private BiConsumer<Action, StackGresConfig> invalidateConfig() {
842+
return (action, config) -> deployedResourcesCache.removeAll(config);
843+
}
844+
845+
private BiConsumer<Action, StackGresCluster> invalidateCluster() {
846+
return (action, cluster) -> deployedResourcesCache.removeAll(cluster);
847+
}
848+
849+
private BiConsumer<Action, StackGresDistributedLogs> invalidateDistributedLogs() {
850+
return (action, distributedlogs) -> deployedResourcesCache.removeAll(distributedlogs);
851+
}
852+
853+
private BiConsumer<Action, StackGresShardedCluster> invalidateShardedCluster() {
854+
return (action, shardedCluster) -> deployedResourcesCache.removeAll(shardedCluster);
855+
}
856+
857+
private BiConsumer<Action, StackGresDbOps> invalidateDbOps() {
858+
return (action, dbops) -> deployedResourcesCache.removeAll(dbops);
859+
}
860+
861+
private BiConsumer<Action, StackGresBackup> invalidateBackup() {
862+
return (action, backup) -> deployedResourcesCache.removeAll(backup);
863+
}
864+
865+
private BiConsumer<Action, StackGresShardedBackup> invalidateShardedBackup() {
866+
return (action, backup) -> deployedResourcesCache.removeAll(backup);
867+
}
868+
869+
private BiConsumer<Action, StackGresShardedDbOps> invalidateShardedDbOps() {
870+
return (action, dbOps) -> deployedResourcesCache.removeAll(dbOps);
871+
}
872+
873+
private BiConsumer<Action, StackGresStream> invalidateStream() {
874+
return (action, stream) -> deployedResourcesCache.removeAll(stream);
875+
}
876+
714877
@Override
715878
public void stopWatchers() {
716879
monitors.forEach(WatcherMonitor::close);

‎stackgres-k8s/src/operator/src/main/java/io/stackgres/operator/conciliation/AbstractConciliator.java‎

Lines changed: 7 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -102,12 +102,12 @@ public ReconciliationResult evalReconciliationState(T config) {
102102
var deployedOtherOwnerRequiredResources = deployedResourcesSnapshot.deployedResources().stream()
103103
.filter(deployedResource -> deployedResource.getMetadata().getOwnerReferences() != null
104104
&& !deployedResource.getMetadata().getOwnerReferences().isEmpty())
105-
.map(ResourceKey::create)
105+
.map(deployedResource -> ResourceKey.create(config, deployedResource))
106106
.filter(deployedResourceKey -> deployedResourcesSnapshot.ownedDeployedResources().stream()
107-
.map(ResourceKey::create)
107+
.map(ownedDeployedResource -> ResourceKey.create(config, ownedDeployedResource))
108108
.noneMatch(deployedResourceKey::equals))
109109
.filter(deployedResourceKey -> requiredResources.stream()
110-
.map(ResourceKey::create)
110+
.map(requiredResource -> ResourceKey.create(config, requiredResource))
111111
.anyMatch(deployedResourceKey::equals))
112112
.toList();
113113
if (!deployedOtherOwnerRequiredResources.isEmpty()) {
@@ -122,7 +122,7 @@ public ReconciliationResult evalReconciliationState(T config) {
122122
}
123123
// Workaround for https://github.com/kubernetes/kubernetes/issues/120960
124124
cleanupNonGarbageCollectedResources(
125-
deployedResourcesSnapshot, deployedOtherOwnerRequiredResources);
125+
config, deployedResourcesSnapshot, deployedOtherOwnerRequiredResources);
126126
return new ReconciliationResult(
127127
List.of(),
128128
List.of(),
@@ -150,13 +150,14 @@ public ReconciliationResult evalReconciliationState(T config) {
150150
}
151151

152152
private void cleanupNonGarbageCollectedResources(
153+
HasMetadata generator,
153154
DeployedResourcesSnapshot deployedResourcesSnapshot,
154155
List<ResourceKey> deployedNonOwnedRequiredResources) {
155156
deployedNonOwnedRequiredResources.stream()
156157
.filter(resourceKey -> resourceKey.kind().equals(
157158
HasMetadata.getKind(ServiceAccount.class)))
158159
.map(resourceKey -> deployedResourcesSnapshot.deployedResources().stream()
159-
.filter(deployedResource -> ResourceKey.create(deployedResource).equals(resourceKey))
160+
.filter(deployedResource -> ResourceKey.create(generator, deployedResource).equals(resourceKey))
160161
.findFirst()
161162
.orElseThrow())
162163
.filter(resource -> resource.getMetadata().getOwnerReferences() != null
@@ -247,7 +248,7 @@ public boolean test(HasMetadata foundDeployedResource) {
247248

248249
private boolean requireDeletion(HasMetadata foundDeployedResource) {
249250
boolean result = requiredResources.stream()
250-
.noneMatch(required -> ResourceKey.same(required, foundDeployedResource));
251+
.noneMatch(required -> ResourceKey.same(config, required, foundDeployedResource));
251252
if (result && LOGGER.isTraceEnabled()) {
252253
LOGGER.trace("Detected deletion for resource {} {}.{}",
253254
foundDeployedResource.getKind(),

‎stackgres-k8s/src/operator/src/main/java/io/stackgres/operator/conciliation/AbstractDeployedResourcesScanner.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -127,10 +127,10 @@ public DeployedResourcesSnapshot getDeployedResources(
127127
.filter(resource -> checkOwnerReference(config, kind, resource))
128128
.toList();
129129

130-
deployedResourcesCache.removeWithLabelsNotIn(genericLabels, deployedResources);
130+
deployedResourcesCache.removeWithLabelsNotIn(config, genericLabels, deployedResources);
131131
final DeployedResourcesSnapshot deployedResourcesSnapshot =
132132
deployedResourcesCache.createDeployedResourcesSnapshot(
133-
ownedDeployedResources, deployedResources);
133+
config, ownedDeployedResources, deployedResources);
134134

135135
return deployedResourcesSnapshot;
136136
}

0 commit comments

Comments
 (0)