Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[system test] make Recovery tests run only on KRaft mode only #10637

Open
wants to merge 6 commits into
base: main
Choose a base branch
from
Open

[system test] make Recovery tests run only on KRaft mode only #10637

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
38acdd2
[system test] make Recovery tests run only on ZK mode only
see-quick Sep 24, 2024
2fdb4d0
make tests KRaft compatible
see-quick Oct 11, 2024
3f1e1e8
refactor a bit
see-quick Oct 11, 2024
69fc552
update to fully work with KRAft
see-quick Nov 5, 2024
2aa9a7f
update fabri8 instead of cli
see-quick Nov 6, 2024
049f5eb
update Henrich review
see-quick Nov 8, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
98 changes: 73 additions & 25 deletions systemtest/src/test/java/io/strimzi/systemtest/operators/NamespaceDeletionRecoveryST.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,21 +4,28 @@
*/
package io.strimzi.systemtest.operators;

import io.fabric8.kubernetes.api.model.LabelSelectorBuilder;
import io.fabric8.kubernetes.api.model.PersistentVolume;
import io.fabric8.kubernetes.api.model.PersistentVolumeClaim;
import io.fabric8.kubernetes.api.model.storage.StorageClass;
import io.fabric8.kubernetes.api.model.storage.StorageClassBuilder;
import io.fabric8.kubernetes.client.dsl.base.PatchContext;
import io.fabric8.kubernetes.client.dsl.base.PatchType;
import io.strimzi.api.kafka.model.kafka.KafkaResources;
import io.strimzi.api.kafka.model.kafka.entityoperator.EntityOperatorSpecBuilder;
import io.strimzi.api.kafka.model.nodepool.ProcessRoles;
import io.strimzi.api.kafka.model.topic.KafkaTopic;
import io.strimzi.operator.common.Annotations;
import io.strimzi.systemtest.AbstractST;
import io.strimzi.systemtest.Environment;
import io.strimzi.systemtest.TestConstants;
import io.strimzi.systemtest.annotations.IsolatedTest;
import io.strimzi.systemtest.cli.KafkaCmdClient;
import io.strimzi.systemtest.kafkaclients.internalClients.KafkaClients;
import io.strimzi.systemtest.resources.NamespaceManager;
import io.strimzi.systemtest.resources.NodePoolsConverter;
import io.strimzi.systemtest.resources.ResourceManager;
import io.strimzi.systemtest.resources.crd.KafkaNodePoolResource;
import io.strimzi.systemtest.resources.crd.KafkaResource;
import io.strimzi.systemtest.resources.crd.KafkaTopicResource;
import io.strimzi.systemtest.resources.operator.SetupClusterOperator;
Expand All @@ -27,6 +34,7 @@
import io.strimzi.systemtest.templates.crd.KafkaTemplates;
import io.strimzi.systemtest.templates.crd.KafkaTopicTemplates;
import io.strimzi.systemtest.utils.ClientUtils;
import io.strimzi.systemtest.utils.kafkaUtils.KafkaUtils;
import io.strimzi.systemtest.utils.kubeUtils.controllers.DeploymentUtils;
import io.strimzi.systemtest.utils.kubeUtils.objects.PersistentVolumeClaimUtils;
import org.apache.logging.log4j.LogManager;
Expand All @@ -40,6 +48,7 @@

import static io.strimzi.systemtest.TestTags.RECOVERY;
import static io.strimzi.test.k8s.KubeClusterResource.kubeClient;
import static org.junit.jupiter.api.Assumptions.assumeTrue;

/**
* Suite for testing topic recovery in case of namespace deletion.
Expand Down Expand Up @@ -95,21 +104,16 @@ void testTopicAvailable() {
.withStorageClass(storageClassName)
.endPersistentClaimStorage()
.endSpec()
.build()
)
.build())
);
resourceManager.createResourceWithWait(KafkaTemplates.kafkaPersistent(testStorage.getNamespaceName(), testStorage.getClusterName(), 3, 3)

resourceManager.createResourceWithWait(KafkaTemplates.kafkaPersistentKRaft(testStorage.getNamespaceName(), testStorage.getClusterName(), 3)
.editSpec()
.editKafka()
.withNewPersistentClaimStorage()
.withSize("1Gi")
.endPersistentClaimStorage()
.endKafka()
.editZookeeper()
.withNewPersistentClaimStorage()
.withSize("1Gi")
.endPersistentClaimStorage()
.endZookeeper()
.endSpec()
.build());

Expand All @@ -125,7 +129,7 @@ void testTopicAvailable() {
*
**/
@IsolatedTest("We need for each test case its own Cluster Operator")
void testTopicNotAvailable() {
void testTopicNotAvailable() {
final TestStorage testStorage = new TestStorage(ResourceManager.getTestContext());
prepareEnvironmentForRecovery(testStorage);

Expand All @@ -138,7 +142,17 @@ void testTopicNotAvailable() {
LOGGER.info("Claim: {} has bounded Volume: {}", pvc.getMetadata().getName(), pv.getMetadata().getName());
}

String kafkaPodName = kubeClient().listPodsByPrefixInName(testStorage.getNamespaceName(), testStorage.getBrokerComponentName()).get(0).getMetadata().getName();
// store cluster-id from Kafka CR
final String oldClusterId = KafkaResource.kafkaClient()
.inNamespace(testStorage.getNamespaceName())
.withName(testStorage.getClusterName())
.get()
.getStatus()
.getClusterId();

final Map<String, String> kafkaLabels = KafkaNodePoolResource.getLabelSelector(testStorage.getClusterName(), testStorage.getBrokerPoolName(), ProcessRoles.BROKER).getMatchLabels();
final String kafkaPodName = kubeClient().listPods(testStorage.getNamespaceName(), new LabelSelectorBuilder()
.withMatchLabels(kafkaLabels).build()).get(0).getMetadata().getName();

LOGGER.info("Currently present Topics inside Kafka: {}/{} are: {}", testStorage.getNamespaceName(), kafkaPodName,
KafkaCmdClient.listTopicsUsingPodCli(testStorage.getNamespaceName(), kafkaPodName, KafkaResources.plainBootstrapAddress(testStorage.getClusterName())));
Expand All @@ -152,21 +166,58 @@ void testTopicNotAvailable() {
LOGGER.info("Recreating Cluster Operator");
recreateClusterOperator(testStorage.getNamespaceName());

LOGGER.info("Recreating Kafka cluster without Topic Operator");
resourceManager.createResourceWithWait(
NodePoolsConverter.convertNodePoolsIfNeeded(
KafkaNodePoolTemplates.brokerPoolPersistentStorage(testStorage.getNamespaceName(), testStorage.getBrokerPoolName(), testStorage.getClusterName(), 3).build(),
KafkaNodePoolTemplates.controllerPoolPersistentStorage(testStorage.getNamespaceName(), testStorage.getControllerPoolName(), testStorage.getClusterName(), 3).build()
)
KafkaNodePoolTemplates.brokerPoolPersistentStorage(testStorage.getNamespaceName(), KafkaNodePoolResource.getBrokerPoolName(testStorage.getClusterName()), testStorage.getClusterName(), 3)
.editSpec()
.withNewPersistentClaimStorage()
.withSize("1Gi")
.withStorageClass(storageClassName)
.endPersistentClaimStorage()
.endSpec()
.build(),
KafkaNodePoolTemplates.controllerPoolPersistentStorage(testStorage.getNamespaceName(), KafkaNodePoolResource.getControllerPoolName(testStorage.getClusterName()), testStorage.getClusterName(), 3)
.editSpec()
.withNewPersistentClaimStorage()
.withSize("1Gi")
.withStorageClass(storageClassName)
.endPersistentClaimStorage()
.endSpec()
.build())
);
resourceManager.createResourceWithWait(KafkaTemplates.kafkaPersistent(testStorage.getNamespaceName(), testStorage.getClusterName(), 3, 3)

LOGGER.info("Recreating Kafka cluster without Topic Operator");
resourceManager.createResourceWithoutWait(KafkaTemplates.kafkaPersistentKRaft(testStorage.getNamespaceName(), testStorage.getClusterName(), 3)
.editMetadata()
// Deploy the Kafka cluster using the original configuration for the Kafka resource.
// Add the annotation strimzi.io/pause-reconciliation="true" to the original configuration for the Kafka resource,
// and then deploy the Kafka cluster using the updated configuration.
.addToAnnotations(Annotations.ANNO_STRIMZI_IO_PAUSE_RECONCILIATION, "true")
.endMetadata()
.editSpec()
.withNewEntityOperator()
.endEntityOperator()
.endSpec()
.build()
);

// Edit the Kafka resource to set the .status.clusterId with the recovered value
KafkaResource.kafkaClient()
.inNamespace(testStorage.getNamespaceName())
.withName(testStorage.getClusterName())
.subresource("status")
.patch(PatchContext.of(PatchType.JSON_MERGE), String.format("{\"status\": {\"clusterId\": \"%s\"}}", oldClusterId));

// Unpause the Kafka resource reconciliation
KafkaResource.replaceKafkaResourceInSpecificNamespace(testStorage.getNamespaceName(), testStorage.getClusterName(),
kafka -> {
Map<String, String> annotations = kafka.getMetadata().getAnnotations();
annotations.put(Annotations.ANNO_STRIMZI_IO_PAUSE_RECONCILIATION, "false");
kafka.getMetadata().setAnnotations(annotations);
});

KafkaUtils.waitForKafkaReady(testStorage.getNamespaceName(), testStorage.getClusterName());

LOGGER.info("Currently present Topics inside Kafka: {}/{} are: {}", testStorage.getNamespaceName(), kafkaPodName,
KafkaCmdClient.listTopicsUsingPodCli(testStorage.getNamespaceName(), kafkaPodName, KafkaResources.plainBootstrapAddress(testStorage.getClusterName())));

Expand Down Expand Up @@ -213,23 +264,17 @@ private void prepareEnvironmentForRecovery(TestStorage testStorage) {
.withStorageClass(storageClassName)
.endPersistentClaimStorage()
.endSpec()
.build()
)
.build())
);
resourceManager.createResourceWithWait(KafkaTemplates.kafkaPersistent(testStorage.getNamespaceName(), testStorage.getClusterName(), 3, 3)

resourceManager.createResourceWithWait(KafkaTemplates.kafkaPersistentKRaft(testStorage.getNamespaceName(), testStorage.getClusterName(), 3)
see-quick marked this conversation as resolved.
Show resolved Hide resolved
.editSpec()
.editKafka()
.withNewPersistentClaimStorage()
.withSize("1Gi")
.withStorageClass(storageClassName)
.endPersistentClaimStorage()
.endKafka()
.editZookeeper()
.withNewPersistentClaimStorage()
.withSize("1Gi")
.withStorageClass(storageClassName)
.endPersistentClaimStorage()
.endZookeeper()
.endSpec()
.build());

Expand Down Expand Up @@ -275,13 +320,16 @@ private void deleteAndRecreateNamespace(String namespace) {

@BeforeAll
void createStorageClass() {
assumeTrue(Environment.isKRaftModeEnabled() && Environment.isKafkaNodePoolsEnabled());

// Delete specific StorageClass if present from previous
kubeClient().getClient().storage().v1().storageClasses().withName(storageClassName).delete();

final String storageClassKubernetesIo = "storageclass.kubernetes.io/is-default-class";
// Get default StorageClass and change reclaim policy
StorageClass defaultStorageClass = kubeClient().getClient().storage().v1().storageClasses().list().getItems().stream().filter(sg -> {
Map<String, String> annotations = sg.getMetadata().getAnnotations();
return annotations.get("storageclass.kubernetes.io/is-default-class").equals("true");
return annotations != null && annotations.containsKey(storageClassKubernetesIo) && annotations.get(storageClassKubernetesIo).equals("true");
}).findFirst().get();

StorageClass retainStorageClass = new StorageClassBuilder(defaultStorageClass)
Expand Down
Loading
Loading