This repository has been archived by the owner on Jun 6, 2023. It is now read-only.
forked from airbytehq/airbyte
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
use Kubernetes watch api for retrieving exit codes (airbytehq#11083)
* use kubernetes api for retrieving exit codes * undelete test * clean up more status check interval * fmt * wip * clean up * smarter filtering * reordering * exception handling * better logging for test + speed up acceptance tests temp * re-enable running on branch * fix race condition in test * add log * trigger build * trigger build * re-run tests with everything enabled * run tests * run tests * clean up * respond to comments * fix formatting * fix whitespace * remove comment * 10 -> 5 * log exit code error message
- Loading branch information
Showing
17 changed files
with
237 additions
and
239 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
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
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
71 changes: 71 additions & 0 deletions
71
airbyte-workers/src/main/java/io/airbyte/workers/process/ExitCodeWatcher.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,71 @@ | ||
| /* | ||
| * Copyright (c) 2021 Airbyte, Inc., all rights reserved. | ||
| */ | ||
|
|
||
| package io.airbyte.workers.process; | ||
|
|
||
| import com.google.common.collect.MoreCollectors; | ||
| import io.fabric8.kubernetes.api.model.ContainerStatus; | ||
| import io.fabric8.kubernetes.api.model.Pod; | ||
| import io.fabric8.kubernetes.client.Watcher; | ||
| import io.fabric8.kubernetes.client.WatcherException; | ||
| import java.util.function.Consumer; | ||
| import lombok.extern.slf4j.Slf4j; | ||
|
|
||
| /** | ||
| * The exit code watcher uses the Kubernetes watch API, which provides a subscription to events for | ||
| * a pod. This subscription has better latency than polling at the expense of keeping a connection | ||
| * open with the Kubernetes API server. Since it offers all events, it helps us handle cases like | ||
| * where a pod is swept or deleted immediately after running on a Kubernetes cluster (we will still | ||
| * be able to retrieve the exit code). | ||
| */ | ||
| @Slf4j | ||
| public class ExitCodeWatcher implements Watcher<Pod> { | ||
|
|
||
| private final Consumer<Integer> onExitCode; | ||
| private final Consumer<WatcherException> onWatchFailure; | ||
| private boolean exitCodeRetrieved = false; | ||
|
|
||
| /** | ||
| * | ||
| * @param onExitCode callback used to store the exit code | ||
| * @param onWatchFailure callback that's triggered when the watch fails. should be some failed exit | ||
| * code. | ||
| */ | ||
| public ExitCodeWatcher(final Consumer<Integer> onExitCode, final Consumer<WatcherException> onWatchFailure) { | ||
| this.onExitCode = onExitCode; | ||
| this.onWatchFailure = onWatchFailure; | ||
| } | ||
|
|
||
| @Override | ||
| public void eventReceived(Action action, Pod resource) { | ||
| try { | ||
| if (!exitCodeRetrieved && KubePodResourceHelper.isTerminal(resource)) { | ||
| final ContainerStatus mainContainerStatus = resource.getStatus().getContainerStatuses() | ||
| .stream() | ||
| .filter(containerStatus -> containerStatus.getName().equals(KubePodProcess.MAIN_CONTAINER_NAME)) | ||
| .collect(MoreCollectors.onlyElement()); | ||
|
|
||
| if (mainContainerStatus.getState() != null && mainContainerStatus.getState().getTerminated() != null) { | ||
| final int exitCode = mainContainerStatus.getState().getTerminated().getExitCode(); | ||
| log.info("Processing event with exit code " + exitCode + " for pod: " + resource.getMetadata().getName()); | ||
| onExitCode.accept(exitCode); | ||
| exitCodeRetrieved = true; | ||
| } | ||
| } | ||
| } catch (Exception e) { | ||
| String podName = "<unknown_name>"; | ||
| if (resource.getMetadata() != null) { | ||
| podName = resource.getMetadata().getName(); | ||
| } | ||
|
|
||
| log.error("ExitCodeWatcher event handling failed for pod: " + podName, e); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void onClose(WatcherException cause) { | ||
| onWatchFailure.accept(cause); | ||
| } | ||
|
|
||
| } |
Oops, something went wrong.