Skip to content

Commit 67c7bd1

Browse files
teoincontattoteoincontatto
committed
feat: improve performance and responsiveness
1 parent b41f677 commit 67c7bd1

25 files changed

Lines changed: 536 additions & 93 deletions

‎stackgres-k8s/e2e/spec/runbooks/operator-responsiveness‎

Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,73 @@
1+
#!/bin/sh
2+
3+
E2E_RESPOSIVENESS_CARDINALITY="${E2E_RESPOSIVENESS_CARDINALITY:-2000}"
4+
5+
e2e_test_install() {
6+
kubectl create namespace "$CLUSTER_NAMESPACE"
7+
seq 1 "$E2E_RESPOSIVENESS_CARDINALITY" \
8+
| xargs_parallel_shell @ \
9+
-ec "$(create_cluster_script)"
10+
until kubectl get sts -n "$CLUSTER_NAMESPACE" -o name | wc -l | grep -qxF "$E2E_RESPOSIVENESS_CARDINALITY"
11+
do
12+
sleep 5
13+
done
14+
}
15+
16+
create_cluster_script() {
17+
cat << EOF
18+
CLUSTER_NAME="\$("$SHELL" "$E2E_PATH/e2e" get_sgcluster_name "$SPEC_NAME"-"@")"
19+
"$SHELL" "$E2E_PATH/e2e" wait_until create_or_replace_cluster "\$CLUSTER_NAME" "$CLUSTER_NAMESPACE" 1 \
20+
--force \
21+
--set-string cluster.pods.scheduling.nodeSelector.none=none \
22+
--set-string cluster.sgInstanceProfile="\$CLUSTER_NAME"-s \
23+
--set-json instanceProfiles='[{"name":"'"\$CLUSTER_NAME"'-s","cpu":"1","memory":"2Gi"}]' \
24+
--set-string cluster.configurations.sgPostgresConfig="\$CLUSTER_NAME" \
25+
--set-string cluster.configurations.sgPoolingConfig="\$CLUSTER_NAME"
26+
EOF
27+
}
28+
29+
wait_cluster_script() {
30+
cat << EOF
31+
CLUSTER_NAME="\$("$SHELL" "$E2E_PATH/e2e" get_sgcluster_name "$SPEC_NAME"-"@")"
32+
until kubectl get sts -n "$CLUSTER_NAMESPACE" "\$CLUSTER_NAME" >/dev/null 2>&1
33+
do
34+
sleep 1
35+
done
36+
EOF
37+
}
38+
39+
e2e_test_uninstall() {
40+
k8s_async_cleanup_namespace "$CLUSTER_NAMESPACE"
41+
}
42+
43+
e2e_test() {
44+
run_test "Benchmarking operator responsiveness" benchmark_operator
45+
}
46+
47+
benchmark_operator() {
48+
for ROUND in $(seq 1 20)
49+
do
50+
INDEX="$(( ( E2E_RESPOSIVENESS_CARDINALITY * RANDOM / 32767 ) + 1 ))"
51+
CLUSTER_NAME="$(get_sgcluster_name "$SPEC_NAME"-"$INDEX")"
52+
INSTANCES="$(kubectl get sgcluster -n "$CLUSTER_NAMESPACE" "$CLUSTER_NAME" --template '{{ .spec.instances }}')"
53+
START="$(date +%s%N)"
54+
kubectl patch sgcluster -n "$CLUSTER_NAMESPACE" "$CLUSTER_NAME" --type merge -p "spec: { instances: $((INSTANCES + 1)) }"
55+
{
56+
echo "$$" > "$LOG_PATH"/kubectl_get_sts_pid
57+
exec kubectl get sts -n "$CLUSTER_NAMESPACE" "$CLUSTER_NAME" --template '{{ printf "%d\n" .spec.replicas }}' -w
58+
} \
59+
| while read REPLICAS
60+
do
61+
if [ "$((INSTANCES + 1))" = "$REPLICAS" ]
62+
then
63+
kill "$(cat "$LOG_PATH"/kubectl_get_sts_pid)"
64+
break
65+
fi
66+
done
67+
END="$(date +%s%N)"
68+
echo "$(( END - START ))" | tee -a "$LOG_PATH"/reconciliation_response_times
69+
done
70+
echo "Test results"
71+
echo
72+
cat "$LOG_PATH"/reconciliation_response_times
73+
}

‎stackgres-k8s/src/common/src/main/java/io/stackgres/common/OperatorProperty.java‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -44,7 +44,11 @@ public enum OperatorProperty implements StackGresPropertyReader {
4444
ALLOWED_NAMESPACES("stackgres.allowedNamespaces"),
4545
CLUSTER_ROLE_DISABLED("stackgres.clusterRoleDisabled"),
4646
FORCE_UNLOCK_OPERATOR("stackgres.forceUnlockOperator"),
47-
STOP_AFTER_BOOTSTRAP("stackgres.stopAfterBootstrap");
47+
STOP_AFTER_BOOTSTRAP("stackgres.stopAfterBootstrap"),
48+
RECONCILIATION_THREADS("stackgres.reconciliationThreads"),
49+
RECONCILIATION_INITIAL_BACKOFF("stackgres.reconciliationInitialBackoff"),
50+
RECONCILIATION_MAX_BACKOFF("stackgres.reconciliationMaxBackoff"),
51+
RECONCILIATION_BACKOFF_VARIATION("stackgres.reconciliationBackoffVariation");
4852

4953
private static final Properties APPLICATION_PROPERTIES =
5054
StackGresPropertyReader.readApplicationProperties(OperatorProperty.class);

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

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -61,10 +61,17 @@ public CrUpdater(
6161
}
6262

6363
public void updateExistingCustomResources() {
64-
LOGGER.info("Updating existing custom resources");
6564
var config = configFinder.findByNameAndNamespace(operatorName, sgConfigNamespace)
6665
.orElseThrow(() -> new IllegalArgumentException(
6766
"SGConfig " + sgConfigNamespace + "." + operatorName + " was not found"));
67+
if (Optional.of(config)
68+
.map(StackGresConfig::getStatus)
69+
.map(StackGresConfigStatus::getExistingCrUpdatedToVersion)
70+
.filter(StackGresProperty.OPERATOR_VERSION.getString()::equals)
71+
.isPresent()) {
72+
return;
73+
}
74+
LOGGER.info("Updating existing custom resources");
6875
crdLoader.scanCrds()
6976
.stream()
7077
.forEach(installedCrd -> {

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

Lines changed: 68 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,8 @@
1414
import java.util.concurrent.CompletableFuture;
1515
import java.util.concurrent.ExecutorService;
1616
import java.util.concurrent.Executors;
17+
import java.util.concurrent.ScheduledExecutorService;
18+
import java.util.concurrent.TimeUnit;
1719
import java.util.concurrent.atomic.AtomicReference;
1820

1921
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
@@ -23,13 +25,16 @@
2325
import io.fabric8.kubernetes.client.CustomResource;
2426
import io.fabric8.kubernetes.client.KubernetesClient;
2527
import io.stackgres.common.CdiUtil;
28+
import io.stackgres.common.OperatorProperty;
29+
import io.stackgres.common.RetryUtil;
2630
import io.stackgres.common.StackGresContext;
2731
import io.stackgres.common.resource.CustomResourceFinder;
2832
import io.stackgres.common.resource.CustomResourceScanner;
2933
import io.stackgres.operator.app.OperatorLockHolder;
3034
import org.jooq.lambda.Seq;
3135
import org.jooq.lambda.tuple.Tuple;
3236
import org.jooq.lambda.tuple.Tuple2;
37+
import org.jooq.lambda.tuple.Tuple5;
3338
import org.slf4j.Logger;
3439
import org.slf4j.LoggerFactory;
3540

@@ -50,9 +55,15 @@ public abstract class AbstractReconciliator<T extends CustomResource<?, ?>> {
5055
private final OperatorLockHolder operatorLockReconciliator;
5156
private final String reconciliationName;
5257
private final ExecutorService executorService;
53-
private final AtomicReference<List<Optional<T>>> atomicReference =
54-
new AtomicReference<List<Optional<T>>>(List.of());
58+
private final ScheduledExecutorService scheduledExecutorService;
59+
private final AtomicReference<List<Optional<Tuple2<T, Integer>>>> atomicReference =
60+
new AtomicReference<>(List.of());
5561
private final ArrayBlockingQueue<Boolean> arrayBlockingQueue = new ArrayBlockingQueue<>(1);
62+
private final ReconciliatorWorkerThreadPool reconciliatorWorkerThreadPool;
63+
64+
private final int reconciliationInitialBackoff;
65+
private final int reconciliationMaxBackoff;
66+
private final int reconciliationBackoffVariation;
5667

5768
private final CompletableFuture<Void> stopped = new CompletableFuture<>();
5869
private boolean close = false;
@@ -65,6 +76,7 @@ protected AbstractReconciliator(
6576
HandlerDelegator<T> handlerDelegator,
6677
KubernetesClient client,
6778
OperatorLockHolder operatorLockReconciliator,
79+
ReconciliatorWorkerThreadPool reconciliatorWorkerThreadPool,
6880
String reconciliationName) {
6981
this.scanner = scanner;
7082
this.finder = finder;
@@ -76,6 +88,22 @@ protected AbstractReconciliator(
7688
this.operatorLockReconciliator = operatorLockReconciliator;
7789
this.executorService = Executors.newSingleThreadExecutor(
7890
r -> new Thread(r, reconciliationName + "-ReconciliationLoop"));
91+
this.scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(
92+
r -> new Thread(r, reconciliationName + "-ReconciliationScheduler"));
93+
this.reconciliatorWorkerThreadPool = reconciliatorWorkerThreadPool;
94+
this.reconciliationInitialBackoff = OperatorProperty.RECONCILIATION_INITIAL_BACKOFF
95+
.get()
96+
.map(Integer::parseInt)
97+
.filter(initial -> initial > 0)
98+
.orElse(5);
99+
this.reconciliationMaxBackoff = OperatorProperty.RECONCILIATION_MAX_BACKOFF
100+
.get()
101+
.map(Integer::parseInt)
102+
.orElse(300);
103+
this.reconciliationBackoffVariation = OperatorProperty.RECONCILIATION_BACKOFF_VARIATION
104+
.get()
105+
.map(Integer::parseInt)
106+
.orElse(10);
79107
}
80108

81109
public AbstractReconciliator() {
@@ -89,6 +117,11 @@ public AbstractReconciliator() {
89117
this.reconciliationName = null;
90118
this.operatorLockReconciliator = null;
91119
this.executorService = null;
120+
this.scheduledExecutorService = null;
121+
this.reconciliatorWorkerThreadPool = null;
122+
this.reconciliationInitialBackoff = 0;
123+
this.reconciliationMaxBackoff = 0;
124+
this.reconciliationBackoffVariation = 0;
92125
}
93126

94127
protected void start() {
@@ -113,12 +146,16 @@ public void reconcileAll() {
113146
}
114147

115148
public void reconcile(T config) {
116-
reconcile(List.of(Optional.of(config)));
149+
reconcile(List.of(Optional.of(Tuple.tuple(config, 0))));
150+
}
151+
152+
private void reconcile(T config, Integer retry) {
153+
reconcile(List.of(Optional.of(Tuple.tuple(config, retry))));
117154
}
118155

119156
@SuppressFBWarnings(value = "RV_RETURN_VALUE_IGNORED_BAD_PRACTICE",
120157
justification = "We do not care if queue is already filled")
121-
private void reconcile(List<Optional<T>> configs) {
158+
private void reconcile(List<Optional<Tuple2<T, Integer>>> configs) {
122159
atomicReference.updateAndGet(atomicConfigs -> Seq
123160
.seq(atomicConfigs)
124161
.append(configs)
@@ -138,7 +175,7 @@ private void reconciliationLoop() {
138175
continue;
139176
}
140177
arrayBlockingQueue.take();
141-
List<Optional<T>> configs = atomicReference.getAndSet(List.of());
178+
List<Optional<Tuple2<T, Integer>>> configs = atomicReference.getAndSet(List.of());
142179
if (close) {
143180
break;
144181
}
@@ -151,27 +188,32 @@ private void reconciliationLoop() {
151188
stopped.complete(null);
152189
}
153190

154-
protected void reconciliationsCycle(List<Optional<T>> configs) {
191+
protected void reconciliationsCycle(List<Optional<Tuple2<T, Integer>>> configs) {
155192
mergedConfigs(configs).stream()
156193
.filter(t -> Optional.ofNullable(t.v1.getMetadata().getAnnotations())
157194
.map(annotations -> annotations.get(STACKGRES_IO_RECONCILIATION))
158195
.map(Boolean::parseBoolean)
159196
.map(b -> !b)
160197
.orElse(true))
161-
.forEach(t -> reconciliationCycle(t.v1, t.v2));
198+
.forEach(t -> reconciliatorWorkerThreadPool.scheduleReconciliation(
199+
() -> reconciliationCycle(t.v1, t.v2, t.v3),
200+
t.v4,
201+
t.v5));
162202
}
163203

164-
private List<Tuple2<T, Boolean>> mergedConfigs(List<Optional<T>> configs) {
165-
if (configs.stream().anyMatch(Optional::isEmpty)) {
166-
return getExistentSources().stream()
167-
.map(config -> Tuple.tuple(config, false))
168-
.toList();
169-
}
170-
return Seq.seq(configs)
171-
.map(Optional::get)
172-
.grouped(this::configId)
173-
.flatMap(t -> t.v2.limit(1))
174-
.map(config -> Tuple.tuple(config, true))
204+
private List<Tuple5<T, Integer, Boolean, String, Boolean>> mergedConfigs(List<Optional<Tuple2<T, Integer>>> configs) {
205+
var groupedConfigs = Seq.seq(configs)
206+
.flatMap(Optional::stream)
207+
.groupBy(t -> configId(t.v1));
208+
return Seq.seq(groupedConfigs)
209+
.map(config -> Tuple.tuple(
210+
config.v2.getFirst().v1, config.v2.getFirst().v2, true, config.v1, true))
211+
.append(Optional.of(configs.stream().anyMatch(Optional::isEmpty))
212+
.filter(anyMatch -> anyMatch)
213+
.stream()
214+
.flatMap(ignored -> getExistentSources().stream())
215+
.map(config -> Tuple.tuple(config, 0, true, configId(config), false))
216+
.filter(config -> !groupedConfigs.containsKey(config.v4)))
175217
.toList();
176218
}
177219

@@ -185,10 +227,10 @@ private List<T> getExistentSources() {
185227
}
186228

187229
private String configId(T config) {
188-
return config.getMetadata().getNamespace() + "." + config.getMetadata().getName();
230+
return config.getCRDName() + "/" + config.getMetadata().getNamespace() + "/" + config.getMetadata().getName();
189231
}
190232

191-
protected void reconciliationCycle(T configKey, boolean load) {
233+
protected void reconciliationCycle(T configKey, int retry, boolean load) {
192234
final ObjectMeta metadata = configKey.getMetadata();
193235
final String configId = configKey.getKind()
194236
+ " " + metadata.getNamespace() + "." + metadata.getName();
@@ -283,6 +325,12 @@ protected void reconciliationCycle(T configKey, boolean load) {
283325
exceptions.add(ex);
284326
}
285327
if (!exceptions.isEmpty()) {
328+
scheduledExecutorService.schedule(() -> reconcile(configKey, retry + 1),
329+
RetryUtil.calculateExponentialBackoffDelay(
330+
reconciliationInitialBackoff,
331+
reconciliationMaxBackoff,
332+
reconciliationBackoffVariation,
333+
retry + 1), TimeUnit.SECONDS);
286334
var iterator = exceptions.listIterator();
287335
Exception ex = iterator.next();
288336
iterator.forEachRemaining(otherEx -> ex.addSuppressed(otherEx));

0 commit comments

Comments
 (0)