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

Run container orchestrator acceptance tests #13699

Merged
merged 5 commits into from
Jul 8, 2022
Merged

Run container orchestrator acceptance tests #13699

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
fcd13bc
Add CONTAINER_ORCHESTRATOR=true flag to kube acceptance tests
lmossman Jun 10, 2022
3192a4a
Add CONTAINER_ORCHESTRATOR=true flag to GKE acceptance tests too
lmossman Jun 10, 2022
9407892
fix downtime test
lmossman Jun 30, 2022
1b16b90
fix the other orchestrator tests
lmossman Jun 30, 2022
2309de2
move container orchestrator acceptance tests to their own class
lmossman Jul 8, 2022
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
212 changes: 0 additions & 212 deletions ...te-tests/src/acceptanceTests/java/io/airbyte/test/acceptance/AdvancedAcceptanceTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,6 @@
import io.airbyte.api.client.model.generated.SyncMode;
import io.airbyte.commons.json.Jsons;
import io.airbyte.commons.lang.MoreBooleans;
import io.airbyte.container_orchestrator.ContainerOrchestratorApp;
import io.airbyte.test.utils.AirbyteAcceptanceTestHarness;
import io.fabric8.kubernetes.client.KubernetesClient;
import java.io.IOException;
Expand All @@ -50,11 +49,7 @@
import java.sql.SQLException;
import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
Expand All @@ -63,12 +58,9 @@
import org.junit.jupiter.api.Order;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestMethodOrder;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable;
import org.junitpioneer.jupiter.RetryingTest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;

/**
* The class test for advanced platform functionality that can be affected by the networking
Expand Down Expand Up @@ -306,208 +298,4 @@ public void testBackpressure() throws Exception {
}
}

@RetryingTest(3)
@Order(5)
@EnabledIfEnvironmentVariable(named = "CONTAINER_ORCHESTRATOR",
matches = "true")
public void testDowntimeDuringSync() throws Exception {
final String connectionName = "test-connection";
final UUID sourceId = testHarness.createPostgresSource().getSourceId();
final UUID destinationId = testHarness.createDestination().getDestinationId();
final AirbyteCatalog catalog = testHarness.discoverSourceSchema(sourceId);
final SyncMode syncMode = SyncMode.FULL_REFRESH;
final DestinationSyncMode destinationSyncMode = DestinationSyncMode.OVERWRITE;
catalog.getStreams().forEach(s -> s.getConfig().syncMode(syncMode).destinationSyncMode(destinationSyncMode));

for (final var input : List.of("KILL_BOTH_NON_SYNC_SLIGHTLY_FIRST", "KILL_ONLY_SYNC", "KILL_ONLY_NON_SYNC")) {
LOGGER.info("Checking " + input);

final UUID connectionId =
testHarness.createConnection(connectionName, sourceId, destinationId, List.of(), catalog, null).getConnectionId();

JobInfoRead connectionSyncRead = null;

while (connectionSyncRead == null) {

try {
connectionSyncRead = apiClient.getConnectionApi().syncConnection(new ConnectionIdRequestBody().connectionId(connectionId));
} catch (final Exception e) {
LOGGER.error("retrying after error", e);
}
}

Thread.sleep(10000);

switch (input) {
case "KILL_BOTH_NON_SYNC_SLIGHTLY_FIRST" -> {
LOGGER.info("Scaling down both workers at roughly the same time...");
kubernetesClient.apps().deployments().inNamespace("default").withName("airbyte-worker").scale(0);
kubernetesClient.apps().deployments().inNamespace("default").withName("airbyte-sync-worker").scale(0, true);
Comment on lines -341 to -345
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Removed these inputs and switch statement, because we don't have any deployments called airbyte-sync-worker. I think this was leftover from some other changes or something


LOGGER.info("Scaling up both workers...");
kubernetesClient.apps().deployments().inNamespace("default").withName("airbyte-worker").scale(1);
kubernetesClient.apps().deployments().inNamespace("default").withName("airbyte-sync-worker").scale(1);
}
case "KILL_ONLY_SYNC" -> {
LOGGER.info("Scaling down only sync worker...");
kubernetesClient.apps().deployments().inNamespace("default").withName("airbyte-sync-worker").scale(0, true);

LOGGER.info("Scaling up sync worker...");
kubernetesClient.apps().deployments().inNamespace("default").withName("airbyte-sync-worker").scale(1);
}
case "KILL_ONLY_NON_SYNC" -> {
LOGGER.info("Scaling down only non-sync worker...");
kubernetesClient.apps().deployments().inNamespace("default").withName("airbyte-worker").scale(0, true);

LOGGER.info("Scaling up non-sync worker...");
kubernetesClient.apps().deployments().inNamespace("default").withName("airbyte-worker").scale(1);
}
}

waitForSuccessfulJob(apiClient.getJobsApi(), connectionSyncRead.getJob());

final long numAttempts = apiClient.getJobsApi()
.getJobInfo(new JobIdRequestBody().id(connectionSyncRead.getJob().getId()))
.getAttempts()
.size();

// it should be able to accomplish the resume without an additional attempt!
assertEquals(1, numAttempts);
}
}

@RetryingTest(3)
@Order(6)
@EnabledIfEnvironmentVariable(named = "CONTAINER_ORCHESTRATOR",
matches = "true")
public void testCancelSyncWithInterruption() throws Exception {
final String connectionName = "test-connection";
final UUID sourceId = testHarness.createPostgresSource().getSourceId();
final UUID destinationId = testHarness.createDestination().getDestinationId();
final UUID operationId = testHarness.createOperation().getOperationId();
final AirbyteCatalog catalog = testHarness.discoverSourceSchema(sourceId);
final SyncMode syncMode = SyncMode.FULL_REFRESH;
final DestinationSyncMode destinationSyncMode = DestinationSyncMode.OVERWRITE;
catalog.getStreams().forEach(s -> s.getConfig().syncMode(syncMode).destinationSyncMode(destinationSyncMode));
final UUID connectionId =
testHarness.createConnection(connectionName, sourceId, destinationId, List.of(operationId), catalog, null).getConnectionId();

final JobInfoRead connectionSyncRead = apiClient.getConnectionApi().syncConnection(new ConnectionIdRequestBody().connectionId(connectionId));
waitWhileJobHasStatus(apiClient.getJobsApi(), connectionSyncRead.getJob(), Set.of(JobStatus.RUNNING));

Thread.sleep(5000);

kubernetesClient.apps().deployments().inNamespace("default").withName("airbyte-worker").scale(0);
Thread.sleep(1000);
kubernetesClient.apps().deployments().inNamespace("default").withName("airbyte-worker").scale(1);

final var resp = apiClient.getJobsApi().cancelJob(new JobIdRequestBody().id(connectionSyncRead.getJob().getId()));
assertEquals(JobStatus.CANCELLED, resp.getJob().getStatus());
}

@RetryingTest(3)
@Order(7)
@Timeout(value = 5,
unit = TimeUnit.MINUTES)
@EnabledIfEnvironmentVariable(named = "CONTAINER_ORCHESTRATOR",
matches = "true")
public void testCuttingOffPodBeforeFilesTransfer() throws Exception {
Copy link
Contributor Author

@lmossman lmossman Jun 30, 2022 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I removed this test entirely because it was almost identical to testDowntimeDuringSync and I didn't see the need for it

final String connectionName = "test-connection";
final UUID sourceId = testHarness.createPostgresSource().getSourceId();
final UUID destinationId = testHarness.createDestination().getDestinationId();
final UUID operationId = testHarness.createOperation().getOperationId();
final AirbyteCatalog catalog = testHarness.discoverSourceSchema(sourceId);
final SyncMode syncMode = SyncMode.FULL_REFRESH;
final DestinationSyncMode destinationSyncMode = DestinationSyncMode.OVERWRITE;
catalog.getStreams().forEach(s -> s.getConfig().syncMode(syncMode).destinationSyncMode(destinationSyncMode));

LOGGER.info("Creating connection...");
final UUID connectionId =
testHarness.createConnection(connectionName, sourceId, destinationId, List.of(operationId), catalog, null).getConnectionId();

LOGGER.info("Waiting for connection to be available in Temporal...");

LOGGER.info("Run manual sync...");
final JobInfoRead connectionSyncRead = apiClient.getConnectionApi().syncConnection(new ConnectionIdRequestBody().connectionId(connectionId));

LOGGER.info("Waiting for job to run...");
waitWhileJobHasStatus(apiClient.getJobsApi(), connectionSyncRead.getJob(), Set.of(JobStatus.RUNNING));

LOGGER.info("Scale down workers...");
kubernetesClient.apps().deployments().inNamespace("default").withName("airbyte-worker").scale(0);

LOGGER.info("Wait for worker scale down...");
Thread.sleep(1000);

LOGGER.info("Scale up workers...");
kubernetesClient.apps().deployments().inNamespace("default").withName("airbyte-worker").scale(1);

LOGGER.info("Waiting for worker timeout...");
Thread.sleep(ContainerOrchestratorApp.MAX_SECONDS_TO_WAIT_FOR_FILE_COPY * 1000 + 1000);

LOGGER.info("Waiting for job to retry and succeed...");
waitForSuccessfulJob(apiClient.getJobsApi(), connectionSyncRead.getJob());
}

@RetryingTest(3)
@Order(8)
@Timeout(value = 5,
unit = TimeUnit.MINUTES)
@EnabledIfEnvironmentVariable(named = "CONTAINER_ORCHESTRATOR",
matches = "true")
public void testCancelSyncWhenCancelledWhenWorkerIsNotRunning() throws Exception {
final String connectionName = "test-connection";
final UUID sourceId = testHarness.createPostgresSource().getSourceId();
final UUID destinationId = testHarness.createDestination().getDestinationId();
final UUID operationId = testHarness.createOperation().getOperationId();
final AirbyteCatalog catalog = testHarness.discoverSourceSchema(sourceId);
final SyncMode syncMode = SyncMode.FULL_REFRESH;
final DestinationSyncMode destinationSyncMode = DestinationSyncMode.OVERWRITE;
catalog.getStreams().forEach(s -> s.getConfig().syncMode(syncMode).destinationSyncMode(destinationSyncMode));

LOGGER.info("Creating connection...");
final UUID connectionId =
testHarness.createConnection(connectionName, sourceId, destinationId, List.of(operationId), catalog, null).getConnectionId();

LOGGER.info("Waiting for connection to be available in Temporal...");

LOGGER.info("Run manual sync...");
final JobInfoRead connectionSyncRead = apiClient.getConnectionApi().syncConnection(new ConnectionIdRequestBody().connectionId(connectionId));

LOGGER.info("Waiting for job to run...");
waitWhileJobHasStatus(apiClient.getJobsApi(), connectionSyncRead.getJob(), Set.of(JobStatus.RUNNING));

LOGGER.info("Waiting for job to run a little...");
Thread.sleep(5000);

LOGGER.info("Scale down workers...");
kubernetesClient.apps().deployments().inNamespace("default").withName("airbyte-worker").scale(0);

LOGGER.info("Waiting for worker shutdown...");
Thread.sleep(2000);

LOGGER.info("Starting background cancellation request...");
final var pool = Executors.newSingleThreadExecutor();
final var mdc = MDC.getCopyOfContextMap();
final Future<JobInfoRead> resp =
pool.submit(() -> {
MDC.setContextMap(mdc);
try {
final JobInfoRead jobInfoRead = apiClient.getJobsApi().cancelJob(new JobIdRequestBody().id(connectionSyncRead.getJob().getId()));
LOGGER.info("jobInfoRead = " + jobInfoRead);
return jobInfoRead;
} catch (final ApiException e) {
LOGGER.error("Failed to read from api", e);
throw e;
}
});
Thread.sleep(2000);

LOGGER.info("Scaling up workers...");
kubernetesClient.apps().deployments().inNamespace("default").withName("airbyte-worker").scale(1);

LOGGER.info("Waiting for cancellation to go into effect...");
assertEquals(JobStatus.CANCELLED, resp.get().getJob().getStatus());
}

}
Loading