-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add test that uses s3 (Minio) as a state backend
Signed-off-by: Jakub Stejskal <[email protected]>
- Loading branch information
Showing
5 changed files
with
812 additions
and
146 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
142 changes: 142 additions & 0 deletions
142
src/main/java/io/streams/operands/minio/SetupMinio.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,142 @@ | ||
| /* | ||
| * Copyright streamshub authors. | ||
| * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). | ||
| */ | ||
| package io.streams.operands.minio; | ||
|
|
||
| import io.fabric8.kubernetes.api.model.EnvVar; | ||
| import io.fabric8.kubernetes.api.model.IntOrString; | ||
| import io.fabric8.kubernetes.api.model.LabelSelector; | ||
| import io.fabric8.kubernetes.api.model.LabelSelectorBuilder; | ||
| import io.fabric8.kubernetes.api.model.Service; | ||
| import io.fabric8.kubernetes.api.model.ServiceBuilder; | ||
| import io.fabric8.kubernetes.api.model.VolumeBuilder; | ||
| import io.fabric8.kubernetes.api.model.VolumeMountBuilder; | ||
| import io.fabric8.kubernetes.api.model.apps.Deployment; | ||
| import io.fabric8.kubernetes.api.model.apps.DeploymentBuilder; | ||
| import io.skodjob.testframe.resources.KubeResourceManager; | ||
| import io.streams.constants.TestConstants; | ||
| import org.apache.logging.log4j.LogManager; | ||
| import org.apache.logging.log4j.Logger; | ||
|
|
||
| import java.util.Map; | ||
|
|
||
| public class SetupMinio { | ||
| private static final Logger LOGGER = LogManager.getLogger(SetupMinio.class); | ||
|
|
||
| public static final String MINIO = "minio"; | ||
| public static final String ADMIN_CREDS = "minioadminLongerThan16BytesForFIPS"; | ||
| public static final String MINIO_STORAGE_ALIAS = "local"; | ||
| public static final int MINIO_PORT = 9000; | ||
| public static final int MINIO_CONSOLE_PORT = 9090; | ||
| private static final String MINIO_IMAGE = "quay.io/minio/minio:latest"; | ||
|
|
||
| /** | ||
| * Deploy minio to a specific namespace, creates service for it and init client inside the Minio pod | ||
| * @param namespace where Minio will be installed to | ||
| */ | ||
| public static void deployMinio(String namespace) { | ||
| // Create a Minio deployment | ||
| Deployment minioDeployment = new DeploymentBuilder() | ||
| .withNewMetadata() | ||
| .withName(MINIO) | ||
| .withNamespace(namespace) | ||
| .withLabels(Map.of(TestConstants.DEPLOYMENT_TYPE, MINIO)) | ||
| .endMetadata() | ||
| .withNewSpec() | ||
| .withReplicas(1) | ||
| .withNewSelector() | ||
| .withMatchLabels(Map.of(TestConstants.APP_POD_LABEL, MINIO)) | ||
| .endSelector() | ||
| .withNewTemplate() | ||
| .withNewMetadata() | ||
| .withLabels(Map.of(TestConstants.APP_POD_LABEL, MINIO)) | ||
| .endMetadata() | ||
| .withNewSpec() | ||
| .addNewContainer() | ||
| .withName(MINIO) | ||
| .withImage(MINIO_IMAGE) | ||
| .withArgs("server", "/data", "--console-address", ":" + MINIO_CONSOLE_PORT) | ||
| .addToEnv(new EnvVar("MINIO_ROOT_USER", ADMIN_CREDS, null)) | ||
| .addToEnv(new EnvVar("MINIO_ROOT_PASSWORD", ADMIN_CREDS, null)) | ||
| .addNewPort() | ||
| .withContainerPort(MINIO_PORT) | ||
| .endPort() | ||
| .withVolumeMounts(new VolumeMountBuilder() | ||
| .withName("minio-storage") | ||
| .withMountPath("/data") | ||
| .build()) | ||
| .endContainer() | ||
| .withVolumes(new VolumeBuilder() | ||
| .withName("minio-storage") | ||
| .withNewEmptyDir() | ||
| .endEmptyDir() | ||
| .build()) | ||
| .endSpec() | ||
| .endTemplate() | ||
| .endSpec() | ||
| .build(); | ||
|
|
||
| // Create the deployment | ||
| KubeResourceManager.getInstance().createResourceWithWait(minioDeployment); | ||
|
|
||
| // Create a service to expose Minio | ||
| Service minioService = new ServiceBuilder() | ||
| .withNewMetadata() | ||
| .withName(MINIO) | ||
| .withNamespace(namespace) | ||
| .endMetadata() | ||
| .withNewSpec() | ||
| .withSelector(Map.of(TestConstants.APP_POD_LABEL, MINIO)) | ||
| .addNewPort() | ||
| .withName("api") | ||
| .withPort(MINIO_PORT) | ||
| .withTargetPort(new IntOrString(MINIO_PORT)) | ||
| .endPort() | ||
| .addNewPort() | ||
| .withName("console") | ||
| .withPort(MINIO_CONSOLE_PORT) | ||
| .withTargetPort(new IntOrString(MINIO_CONSOLE_PORT)) | ||
| .endPort() | ||
| .endSpec() | ||
| .build(); | ||
|
|
||
| KubeResourceManager.getInstance().createResourceWithoutWait(minioService); | ||
| // NetworkPolicyResource.allowNetworkPolicyAllIngressForMatchingLabel(namespace, MINIO, Map.of(TestConstants.APP_POD_LABEL, MINIO)); | ||
|
|
||
| initMinioClient(namespace); | ||
| } | ||
|
|
||
| /** | ||
| * Init client inside the Minio pod. This allows other commands to be executed during the tests. | ||
| * @param namespace where Minio is installed | ||
| */ | ||
| private static void initMinioClient(String namespace) { | ||
| final LabelSelector labelSelector = new LabelSelectorBuilder().withMatchLabels(Map.of(TestConstants.APP_POD_LABEL, MINIO)).build(); | ||
| final String minioPod = KubeResourceManager.getKubeClient().listPods(namespace, labelSelector).get(0).getMetadata().getName(); | ||
|
|
||
| KubeResourceManager.getKubeCmdClient().inNamespace(namespace).execInPod(minioPod, | ||
| "mc", | ||
| "config", | ||
| "host", | ||
| "add", | ||
| MINIO_STORAGE_ALIAS, | ||
| "http://localhost:" + MINIO_PORT, | ||
| ADMIN_CREDS, ADMIN_CREDS); | ||
| } | ||
|
|
||
| /** | ||
| * Create bucket in Minio instance in specific namespace. | ||
| * @param namespace Minio location | ||
| * @param bucketName name of the bucket that will be created and used within the tests | ||
| */ | ||
| public static void createBucket(String namespace, String bucketName) { | ||
| final LabelSelector labelSelector = new LabelSelectorBuilder().withMatchLabels(Map.of(TestConstants.APP_POD_LABEL, MINIO)).build(); | ||
| final String minioPod = KubeResourceManager.getKubeClient().listPods(namespace, labelSelector).get(0).getMetadata().getName(); | ||
|
|
||
| KubeResourceManager.getKubeCmdClient().inNamespace(namespace).execInPod(minioPod, | ||
| "mc", | ||
| "mb", | ||
| MINIO_STORAGE_ALIAS + "/" + bucketName); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,110 @@ | ||
| /* | ||
| * Copyright streamshub authors. | ||
| * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). | ||
| */ | ||
| package io.streams.utils; | ||
|
|
||
| import io.fabric8.kubernetes.api.model.LabelSelector; | ||
| import io.fabric8.kubernetes.api.model.LabelSelectorBuilder; | ||
| import io.skodjob.testframe.TestFrameConstants; | ||
| import io.skodjob.testframe.resources.KubeResourceManager; | ||
| import io.skodjob.testframe.wait.Wait; | ||
| import io.streams.constants.TestConstants; | ||
| import io.streams.operands.minio.SetupMinio; | ||
| import org.apache.logging.log4j.LogManager; | ||
| import org.apache.logging.log4j.Logger; | ||
|
|
||
| import java.util.Map; | ||
| import java.util.regex.Matcher; | ||
| import java.util.regex.Pattern; | ||
|
|
||
| public class MinioUtils { | ||
| private static final Logger LOGGER = LogManager.getLogger(MinioUtils.class); | ||
|
|
||
| private MinioUtils() { | ||
|
|
||
| } | ||
|
|
||
| /** | ||
| * Collect data from Minio about usage of a specific bucket | ||
| * | ||
| * @param namespace Name of the Namespace where the Minio Pod is running | ||
| * @param bucketName Name of the bucket for which we want to get info about its size | ||
| * @return Overall statistics about the bucket in String format | ||
| */ | ||
| public static String getBucketSizeInfo(String namespace, String bucketName) { | ||
| final LabelSelector labelSelector = new LabelSelectorBuilder() | ||
| .withMatchLabels(Map.of(TestConstants.APP_POD_LABEL, SetupMinio.MINIO)) | ||
| .build(); | ||
| final String minioPod = KubeResourceManager.getKubeClient() | ||
| .listPods(namespace, labelSelector) | ||
| .get(0) | ||
| .getMetadata() | ||
| .getName(); | ||
|
|
||
| return KubeResourceManager.getKubeCmdClient() | ||
| .inNamespace(namespace) | ||
| .execInPod(minioPod, | ||
| "mc", | ||
| "stat", | ||
| "local/" + bucketName) | ||
| .out(); | ||
|
|
||
| } | ||
|
|
||
| /** | ||
| * Parse out total size of bucket from the information about usage. | ||
| * | ||
| * @param bucketInfo String containing all stat info about bucket | ||
| * @return Map consists of parsed size and it's unit | ||
| */ | ||
| private static Map<String, Object> parseTotalSize(String bucketInfo) { | ||
| Pattern pattern = Pattern.compile("Total size:\\s*(?<size>[\\d.]+)\\s*(?<unit>.*)"); | ||
| Matcher matcher = pattern.matcher(bucketInfo); | ||
|
|
||
| if (matcher.find()) { | ||
| return Map.of("size", Double.parseDouble(matcher.group("size")), "unit", matcher.group("unit")); | ||
| } else { | ||
| throw new IllegalArgumentException("Total size not found in the provided string"); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Wait until size of the bucket is not 0 B. | ||
| * | ||
| * @param namespace Minio location | ||
| * @param bucketName bucket name | ||
| */ | ||
| public static void waitForDataInMinio(String namespace, String bucketName) { | ||
| Wait.until("data sync from Kafka to Minio", | ||
| TestFrameConstants.GLOBAL_POLL_INTERVAL_MEDIUM, | ||
| TestFrameConstants.GLOBAL_TIMEOUT, | ||
| () -> { | ||
| String bucketSizeInfo = getBucketSizeInfo(namespace, bucketName); | ||
| Map<String, Object> parsedSize = parseTotalSize(bucketSizeInfo); | ||
| double bucketSize = (Double) parsedSize.get("size"); | ||
| LOGGER.info("Collected bucket size: {} {}", bucketSize, parsedSize.get("unit")); | ||
| LOGGER.debug("Collected bucket info:\n{}", bucketSizeInfo); | ||
|
|
||
| return bucketSize > 0; | ||
| }); | ||
| } | ||
|
|
||
| /** | ||
| * Wait until size of the bucket is 0 B. | ||
| * | ||
| * @param namespace Minio location | ||
| * @param bucketName bucket name | ||
| */ | ||
| public static void waitForNoDataInMinio(String namespace, String bucketName) { | ||
| Wait.until("data deletion in Minio", TestFrameConstants.GLOBAL_POLL_INTERVAL_MEDIUM, TestFrameConstants.GLOBAL_TIMEOUT, () -> { | ||
| String bucketSizeInfo = getBucketSizeInfo(namespace, bucketName); | ||
| Map<String, Object> parsedSize = parseTotalSize(bucketSizeInfo); | ||
| double bucketSize = (Double) parsedSize.get("size"); | ||
| LOGGER.info("Collected bucket size: {} {}", bucketSize, parsedSize.get("unit")); | ||
| LOGGER.debug("Collected bucket info:\n{}", bucketSizeInfo); | ||
|
|
||
| return bucketSize == 0; | ||
| }); | ||
| } | ||
| } |
Oops, something went wrong.