Skip to content

Commit 191f37c

Browse files
teoincontattoteoincontatto
committed
fix: SGStream uses wrong SGCluster service name
1 parent 0656964 commit 191f37c

6 files changed

Lines changed: 58 additions & 19 deletions

File tree

‎stackgres-k8s/src/common/src/main/resources/crds/SGStream.yaml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -57,7 +57,7 @@ spec:
5757
5858
* `tombstone`: allow to stop completely Debezium streaming and the SGStream. This signal is useful to give an end to the streaming in a graceful way allowing for the removal of the logical slot created by Debezium.
5959
* `command`: allow to execute any SQL command on the target database. Only available then the target type is `SGCluster`.
60-
required: ["source", "target"]
60+
required: ["source", "target", "pods"]
6161
properties:
6262
source:
6363
type: object

‎stackgres-k8s/src/operator/src/main/java/io/stackgres/operator/conciliation/factory/stream/StreamRole.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
import io.fabric8.kubernetes.api.model.rbac.RoleRefBuilder;
2020
import io.fabric8.kubernetes.api.model.rbac.SubjectBuilder;
2121
import io.stackgres.common.crd.CommonDefinition;
22+
import io.stackgres.common.crd.sgcluster.StackGresCluster;
2223
import io.stackgres.common.crd.sgstream.StackGresStream;
2324
import io.stackgres.common.labels.LabelFactoryForStream;
2425
import io.stackgres.operator.conciliation.OperatorVersionBinder;
@@ -107,6 +108,11 @@ private Role createRole(StackGresStreamContext context) {
107108
.withResources(HasMetadata.getPlural(StackGresStream.class))
108109
.withVerbs("get", "list", "watch", "patch", "update")
109110
.build())
111+
.addToRules(new PolicyRuleBuilder()
112+
.withApiGroups(CommonDefinition.GROUP)
113+
.withResources(HasMetadata.getPlural(StackGresCluster.class))
114+
.withVerbs("get")
115+
.build())
110116
.build();
111117
}
112118

‎stackgres-k8s/src/stream/src/main/java/io/stackgres/stream/jobs/source/AbstractPostgresDebeziumEngineHandler.java‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,9 +15,11 @@
1515

1616
import io.debezium.engine.DebeziumEngine;
1717
import io.debezium.engine.DebeziumEngine.CompletionCallback;
18+
import io.debezium.engine.DebeziumEngine.ConnectorCallback;
1819
import io.debezium.engine.format.SerializationFormat;
1920
import io.fabric8.kubernetes.api.model.Secret;
2021
import io.stackgres.common.StreamPath;
22+
import io.stackgres.common.crd.sgcluster.StackGresCluster;
2123
import io.stackgres.common.crd.sgstream.StackGresStream;
2224
import io.stackgres.common.crd.sgstream.StackGresStreamDebeziumEngineProperties;
2325
import io.stackgres.common.crd.sgstream.StackGresStreamSourcePostgres;
@@ -44,6 +46,9 @@ public abstract class AbstractPostgresDebeziumEngineHandler implements SourceEve
4446

4547
private static final Logger LOGGER = LoggerFactory.getLogger(AbstractPostgresDebeziumEngineHandler.class);
4648

49+
@Inject
50+
CustomResourceFinder<StackGresCluster> clusterFinder;
51+
4752
@Inject
4853
ResourceFinder<Secret> secretFinder;
4954

@@ -117,6 +122,8 @@ public <T> CompletableFuture<Void> streamChangeEvents(
117122
try {
118123
engine = DebeziumEngine.create(format)
119124
.using(props)
125+
.using(new ConnectorCallback() {
126+
})
120127
.using(new CompletionCallback() {
121128
@Override
122129
public void handle(boolean success, String message, Throwable error) {
@@ -141,7 +148,7 @@ public void handle(boolean success, String message, Throwable error) {
141148
}
142149
try {
143150
TombstoneDebeziumSignalAction tombstoneSignalAction = new TombstoneDebeziumSignalAction(
144-
stream, streamScheduler, secretFinder, engine, streamCompleted);
151+
stream, streamScheduler, clusterFinder, secretFinder, engine, streamCompleted);
145152
StreamDebeziumSignalActionProvider.registerSignalAction(
146153
StreamDebeziumSignalActionProvider.TOMBSTONE_SIGNAL_TYPE, tombstoneSignalAction);
147154
executor.execute(engine);

‎stackgres-k8s/src/stream/src/main/java/io/stackgres/stream/jobs/source/SgClusterDebeziumEngineHandler.java‎

Lines changed: 14 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@
1010

1111
import io.stackgres.common.PatroniUtil;
1212
import io.stackgres.common.crd.SecretKeySelector;
13+
import io.stackgres.common.crd.sgcluster.StackGresCluster;
1314
import io.stackgres.common.crd.sgstream.StackGresStream;
1415
import io.stackgres.common.crd.sgstream.StackGresStreamSourcePostgresDebeziumProperties;
1516
import io.stackgres.common.crd.sgstream.StackGresStreamSourceSgCluster;
@@ -25,8 +26,8 @@ public class SgClusterDebeziumEngineHandler extends AbstractPostgresDebeziumEngi
2526

2627
@Override
2728
protected void setSourceProperties(StackGresStream stream, Properties props) {
28-
String namespace = stream.getMetadata().getNamespace();
29-
var sgCluster = Optional.of(stream.getSpec().getSource().getSgCluster());
29+
final String namespace = stream.getMetadata().getNamespace();
30+
final var sgCluster = Optional.of(stream.getSpec().getSource().getSgCluster());
3031
DebeziumUtil.configureDebeziumSectionProperties(
3132
props,
3233
sgCluster
@@ -37,29 +38,32 @@ protected void setSourceProperties(StackGresStream stream, Properties props) {
3738
props.setProperty("publication.name", publicationName(stream));
3839

3940
DebeziumUtil.configureDebeziumIncludesAndExcludes(props, sgCluster.orElseThrow());
40-
String clusterName = sgCluster.map(StackGresStreamSourceSgCluster::getName)
41+
final String clusterName = sgCluster.map(StackGresStreamSourceSgCluster::getName)
4142
.orElseThrow(() -> new IllegalArgumentException("The name of SGCluster is not specified"));
42-
String usernameSecretName = sgCluster
43+
final StackGresCluster cluster = clusterFinder.findByNameAndNamespace(clusterName, namespace)
44+
.orElseThrow(() -> new IllegalArgumentException(StackGresCluster.KIND + " " + clusterName + " not found"));
45+
final String clusterServiceName = PatroniUtil.readWriteName(cluster);
46+
final String usernameSecretName = sgCluster
4347
.map(StackGresStreamSourceSgCluster::getUsername)
4448
.map(SecretKeySelector::getName)
4549
.orElseGet(() -> PatroniUtil.secretName(clusterName));
46-
String usernameSecretKey = sgCluster
50+
final String usernameSecretKey = sgCluster
4751
.map(StackGresStreamSourceSgCluster::getUsername)
4852
.map(SecretKeySelector::getKey)
4953
.orElseGet(() -> StackGresPasswordKeys.SUPERUSER_USERNAME_KEY);
50-
var username = getSecretKeyValue(namespace, usernameSecretName, usernameSecretKey);
51-
String passwordSecretName = sgCluster
54+
final var username = getSecretKeyValue(namespace, usernameSecretName, usernameSecretKey);
55+
final String passwordSecretName = sgCluster
5256
.map(StackGresStreamSourceSgCluster::getPassword)
5357
.map(SecretKeySelector::getName)
5458
.orElseGet(() -> PatroniUtil.secretName(clusterName));
55-
String passwordSecretKey = sgCluster
59+
final String passwordSecretKey = sgCluster
5660
.map(StackGresStreamSourceSgCluster::getPassword)
5761
.map(SecretKeySelector::getKey)
5862
.orElseGet(() -> StackGresPasswordKeys.SUPERUSER_PASSWORD_KEY);
59-
var password = getSecretKeyValue(namespace, passwordSecretName, passwordSecretKey);
63+
final var password = getSecretKeyValue(namespace, passwordSecretName, passwordSecretKey);
6064

6165
props.setProperty("connector.class", "io.debezium.connector.postgresql.PostgresConnector");
62-
props.setProperty("database.hostname", clusterName);
66+
props.setProperty("database.hostname", clusterServiceName);
6367
props.setProperty("database.port", String.valueOf(PatroniUtil.REPLICATION_SERVICE_PORT));
6468
props.setProperty("database.dbname", Optional
6569
.ofNullable(stream.getSpec().getSource().getSgCluster())

‎stackgres-k8s/src/stream/src/main/java/io/stackgres/stream/jobs/source/TombstoneDebeziumSignalAction.java‎

Lines changed: 13 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import io.stackgres.common.PatroniUtil;
2424
import io.stackgres.common.crd.Condition;
2525
import io.stackgres.common.crd.SecretKeySelector;
26+
import io.stackgres.common.crd.sgcluster.StackGresCluster;
2627
import io.stackgres.common.crd.sgstream.StackGresStream;
2728
import io.stackgres.common.crd.sgstream.StackGresStreamSourcePostgres;
2829
import io.stackgres.common.crd.sgstream.StackGresStreamSourceSgCluster;
@@ -33,6 +34,7 @@
3334
import io.stackgres.common.crd.sgstream.StreamTargetType;
3435
import io.stackgres.common.kubernetesclient.KubernetesClientUtil;
3536
import io.stackgres.common.patroni.StackGresPasswordKeys;
37+
import io.stackgres.common.resource.CustomResourceFinder;
3638
import io.stackgres.common.resource.CustomResourceScheduler;
3739
import io.stackgres.common.resource.ResourceFinder;
3840
import io.stackgres.operatorframework.resource.ResourceUtil;
@@ -51,6 +53,7 @@ public class TombstoneDebeziumSignalAction implements SignalAction<Partition> {
5153

5254
final StackGresStream stream;
5355
final CustomResourceScheduler<StackGresStream> streamScheduler;
56+
final CustomResourceFinder<StackGresCluster> clusterFinder;
5457
final ResourceFinder<Secret> secretFinder;
5558
final DebeziumEngine<?> engine;
5659
final CompletableFuture<Void> streamCompleted;
@@ -59,11 +62,13 @@ public class TombstoneDebeziumSignalAction implements SignalAction<Partition> {
5962
public TombstoneDebeziumSignalAction(
6063
StackGresStream stream,
6164
CustomResourceScheduler<StackGresStream> streamScheduler,
65+
CustomResourceFinder<StackGresCluster> clusterFinder,
6266
ResourceFinder<Secret> secretFinder,
6367
DebeziumEngine<?> engine,
6468
CompletableFuture<Void> streamCompleted) {
6569
this.stream = stream;
6670
this.streamScheduler = streamScheduler;
71+
this.clusterFinder = clusterFinder;
6772
this.secretFinder = secretFinder;
6873
this.engine = engine;
6974
this.streamCompleted = streamCompleted;
@@ -133,6 +138,9 @@ private void restoreTargetConstraints() {
133138
final String namespace = stream.getMetadata().getNamespace();
134139
final String clusterName = sgCluster.map(StackGresStreamTargetSgCluster::getName)
135140
.orElseThrow(() -> new IllegalArgumentException("The name of SGCluster is not specified"));
141+
final StackGresCluster cluster = clusterFinder.findByNameAndNamespace(clusterName, namespace)
142+
.orElseThrow(() -> new IllegalArgumentException(StackGresCluster.KIND + " " + clusterName + " not found"));
143+
final String clusterServiceName = PatroniUtil.readWriteName(cluster);
136144
final String clusterPort = String.valueOf(PatroniUtil.REPLICATION_SERVICE_PORT);
137145
final String clusterDatabase = Optional.ofNullable(stream.getSpec().getTarget().getSgCluster())
138146
.map(StackGresStreamTargetSgCluster::getDatabase)
@@ -160,7 +168,7 @@ private void restoreTargetConstraints() {
160168
props.setProperty("connection.password", password);
161169
props.setProperty("connection.url", "jdbc:postgresql://%s:%s/%s"
162170
.formatted(
163-
clusterName,
171+
clusterServiceName,
164172
clusterPort,
165173
clusterDatabase));
166174
final JdbcSinkConnectorConfig config = new JdbcSinkConnectorConfig(props
@@ -248,6 +256,9 @@ private Properties getDebeziumJdbcSinkConnectorConfigSgCluster() {
248256
final String namespace = stream.getMetadata().getNamespace();
249257
final String clusterName = sgCluster.map(StackGresStreamSourceSgCluster::getName)
250258
.orElseThrow(() -> new IllegalArgumentException("The name of SGCluster is not specified"));
259+
final StackGresCluster cluster = clusterFinder.findByNameAndNamespace(clusterName, namespace)
260+
.orElseThrow(() -> new IllegalArgumentException(StackGresCluster.KIND + " " + clusterName + " not found"));
261+
final String clusterServiceName = PatroniUtil.readWriteName(cluster);
251262
final String clusterPort = String.valueOf(PatroniUtil.REPLICATION_SERVICE_PORT);
252263
final String clusterDatabase = Optional.ofNullable(stream.getSpec().getSource().getSgCluster())
253264
.map(StackGresStreamSourceSgCluster::getDatabase)
@@ -275,7 +286,7 @@ private Properties getDebeziumJdbcSinkConnectorConfigSgCluster() {
275286
props.setProperty("connection.password", password);
276287
props.setProperty("connection.url", "jdbc:postgresql://%s:%s/%s"
277288
.formatted(
278-
clusterName,
289+
clusterServiceName,
279290
clusterPort,
280291
clusterDatabase));
281292
return props;

‎stackgres-k8s/src/stream/src/main/java/io/stackgres/stream/jobs/target/migration/SgClusterStreamMigrationHandler.java‎

Lines changed: 16 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -31,13 +31,15 @@
3131
import io.fabric8.kubernetes.api.model.Secret;
3232
import io.stackgres.common.PatroniUtil;
3333
import io.stackgres.common.crd.SecretKeySelector;
34+
import io.stackgres.common.crd.sgcluster.StackGresCluster;
3435
import io.stackgres.common.crd.sgstream.StackGresStream;
3536
import io.stackgres.common.crd.sgstream.StackGresStreamSourceSgCluster;
3637
import io.stackgres.common.crd.sgstream.StackGresStreamTargetJdbcSinkDebeziumProperties;
3738
import io.stackgres.common.crd.sgstream.StackGresStreamTargetSgCluster;
3839
import io.stackgres.common.crd.sgstream.StreamSourceType;
3940
import io.stackgres.common.crd.sgstream.StreamTargetType;
4041
import io.stackgres.common.patroni.StackGresPasswordKeys;
42+
import io.stackgres.common.resource.CustomResourceFinder;
4143
import io.stackgres.common.resource.ResourceFinder;
4244
import io.stackgres.operatorframework.resource.ResourceUtil;
4345
import io.stackgres.stream.jobs.DebeziumUtil;
@@ -67,6 +69,9 @@ public class SgClusterStreamMigrationHandler implements TargetEventHandler {
6769

6870
private static final Logger LOGGER = LoggerFactory.getLogger(SgClusterStreamMigrationHandler.class);
6971

72+
@Inject
73+
CustomResourceFinder<StackGresCluster> clusterFinder;
74+
7075
@Inject
7176
ResourceFinder<Secret> secretFinder;
7277

@@ -143,6 +148,9 @@ public void start() {
143148
StackGresStreamTargetJdbcSinkDebeziumProperties.class);
144149
final String clusterName = sgCluster.map(StackGresStreamTargetSgCluster::getName)
145150
.orElseThrow(() -> new IllegalArgumentException("The name of SGCluster is not specified"));
151+
final StackGresCluster cluster = clusterFinder.findByNameAndNamespace(clusterName, namespace)
152+
.orElseThrow(() -> new IllegalArgumentException(StackGresCluster.KIND + " " + clusterName + " not found"));
153+
final String clusterServiceName = PatroniUtil.readWriteName(cluster);
146154
final String clusterPort = String.valueOf(PatroniUtil.REPLICATION_SERVICE_PORT);
147155
final String clusterDatabase = Optional.ofNullable(stream.getSpec().getSource().getSgCluster())
148156
.map(StackGresStreamSourceSgCluster::getDatabase)
@@ -170,7 +178,7 @@ public void start() {
170178
props.setProperty("connection.password", password);
171179
props.setProperty("connection.url", "jdbc:postgresql://%s:%s/%s"
172180
.formatted(
173-
clusterName,
181+
clusterServiceName,
174182
clusterPort,
175183
clusterDatabase));
176184
final JdbcSinkConnectorConfig config = new JdbcSinkConnectorConfig(props
@@ -192,7 +200,7 @@ public void start() {
192200

193201
if (!Optional.ofNullable(stream.getSpec().getTarget()
194202
.getSgCluster().getSkipDdlImport()).orElse(false)) {
195-
importDdl(props, namespace, clusterName, clusterPort, clusterDatabase);
203+
importDdl(props, namespace, clusterServiceName, clusterPort, clusterDatabase);
196204
} else {
197205
LOGGER.info("Import of DDL has been skipped as required by configuration");
198206
}
@@ -336,14 +344,14 @@ private void executeCommand(StatelessSession session, String commandSql) {
336344
}
337345
}
338346

339-
private void importDdl(final Properties props, final String namespace, final String clusterName,
347+
private void importDdl(final Properties props, final String namespace, final String clusterServiceName,
340348
final String clusterPort, final String clusterDatabase) {
341349
final String sourceType = stream.getSpec().getSource().getType();
342350
switch(StreamSourceType.fromString(sourceType)) {
343351
case SGCLUSTER:
344352
props.setProperty("connection.url", "jdbc:postgresql://%s:%s/%s"
345353
.formatted(
346-
clusterName,
354+
clusterServiceName,
347355
clusterPort,
348356
"postgres"));
349357
final JdbcSinkConnectorConfig importConfig = new JdbcSinkConnectorConfig(props
@@ -370,6 +378,9 @@ private void importDdlSgCluster(StatelessSession session, String namespace, Stri
370378
.orElse("postgres");
371379
String sourceClusterName = sourceSgCluster.map(StackGresStreamSourceSgCluster::getName)
372380
.orElseThrow(() -> new IllegalArgumentException("The name of SGCluster is not specified"));
381+
final StackGresCluster sourceCluster = clusterFinder.findByNameAndNamespace(sourceClusterName, namespace)
382+
.orElseThrow(() -> new IllegalArgumentException(StackGresCluster.KIND + " " + sourceClusterName + " not found"));
383+
final String sourceClusterServiceName = PatroniUtil.readWriteName(sourceCluster);
373384
String sourceUsernameSecretName = sourceSgCluster
374385
.map(StackGresStreamSourceSgCluster::getUsername)
375386
.map(SecretKeySelector::getName)
@@ -394,7 +405,7 @@ private void importDdlSgCluster(StatelessSession session, String namespace, Stri
394405

395406
LOGGER.info("Importing DDL from source SGCluster {} for database {}", sourceClusterName, sourceClusterDatabase);
396407
executeCommand(session, SnapshotHelperQueries.IMPORT_DDL.readSql().formatted(
397-
" host=" + sourceClusterName
408+
" host=" + sourceClusterServiceName
398409
+ " port=" + sourceClusterPort
399410
+ " dbname=" + sourceClusterDatabase
400411
+ " user=" + sourceUsername

0 commit comments

Comments
 (0)