Skip to content

configure temporal workflow execution ttl #9838

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

Merged
merged 2 commits into from
Jan 28, 2022
Merged

configure temporal workflow execution ttl #9838

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a5e0818
configure temporal workflow execution ttl
jrhizor Jan 27, 2022
18415f6
Merge branch 'master' into jrhizor/increase-temporal-workflow-executi…
jrhizor Jan 28, 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
2 changes: 2 additions & 0 deletions airbyte-workers/src/main/java/io/airbyte/workers/WorkerApp.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -323,6 +323,8 @@ public static void main(final String[] args) throws IOException, InterruptedExce

final WorkflowServiceStubs temporalService = TemporalUtils.createTemporalService(temporalHost);

TemporalUtils.configureTemporalNamespace(temporalService);

final Database configDatabase = new ConfigsDatabaseInstance(
configs.getConfigDatabaseUser(),
configs.getConfigDatabasePassword(),
Expand Down
28 changes: 28 additions & 0 deletions airbyte-workers/src/main/java/io/airbyte/workers/temporal/TemporalUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,9 +10,12 @@
import io.airbyte.scheduler.models.JobRunConfig;
import io.temporal.activity.Activity;
import io.temporal.api.common.v1.WorkflowExecution;
import io.temporal.api.namespace.v1.NamespaceConfig;
import io.temporal.api.namespace.v1.NamespaceInfo;
import io.temporal.api.workflowservice.v1.DescribeNamespaceRequest;
import io.temporal.api.workflowservice.v1.DescribeNamespaceResponse;
import io.temporal.api.workflowservice.v1.ListNamespacesRequest;
import io.temporal.api.workflowservice.v1.UpdateNamespaceRequest;
import io.temporal.client.ActivityCompletionException;
import io.temporal.client.WorkflowClient;
import io.temporal.client.WorkflowOptions;
Expand All @@ -30,6 +33,7 @@
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang3.time.DurationFormatUtils;
import org.apache.commons.lang3.tuple.ImmutablePair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand All @@ -56,6 +60,30 @@ public static WorkflowServiceStubs createTemporalService(final String temporalHo

public static final String DEFAULT_NAMESPACE = "default";

private static final Duration WORKFLOW_EXECUTION_TTL = Duration.ofDays(7);
private static final String HUMAN_READABLE_WORKFLOW_EXECUTION_TTL =
DurationFormatUtils.formatDurationWords(WORKFLOW_EXECUTION_TTL.toMillis(), true, true);

public static void configureTemporalNamespace(WorkflowServiceStubs temporalService) {
final var client = temporalService.blockingStub();
final var describeNamespaceRequest = DescribeNamespaceRequest.newBuilder().setNamespace(DEFAULT_NAMESPACE).build();
final var currentRetentionGrpcDuration = client.describeNamespace(describeNamespaceRequest).getConfig().getWorkflowExecutionRetentionTtl();
final var currentRetention = Duration.ofSeconds(currentRetentionGrpcDuration.getSeconds());

if (currentRetention.equals(WORKFLOW_EXECUTION_TTL)) {
LOGGER.info("Workflow execution TTL already set for namespace " + DEFAULT_NAMESPACE + ". Remains unchanged as: "
+ HUMAN_READABLE_WORKFLOW_EXECUTION_TTL);
} else {
final var newGrpcDuration = com.google.protobuf.Duration.newBuilder().setSeconds(WORKFLOW_EXECUTION_TTL.getSeconds()).build();
final var humanReadableCurrentRetention = DurationFormatUtils.formatDurationWords(currentRetention.toMillis(), true, true);
final var namespaceConfig = NamespaceConfig.newBuilder().setWorkflowExecutionRetentionTtl(newGrpcDuration).build();
final var updateNamespaceRequest = UpdateNamespaceRequest.newBuilder().setNamespace(DEFAULT_NAMESPACE).setConfig(namespaceConfig).build();
LOGGER.info("Workflow execution TTL differs for namespace " + DEFAULT_NAMESPACE + ". Changing from (" + humanReadableCurrentRetention + ") to ("
+ HUMAN_READABLE_WORKFLOW_EXECUTION_TTL + "). ");
client.updateNamespace(updateNamespaceRequest);
}
}

@FunctionalInterface
public interface TemporalJobCreator<T extends Serializable> {

Expand Down
2 changes: 1 addition & 1 deletion ...rs/src/main/java/io/airbyte/workers/temporal/scheduling/shared/ActivityConfiguration.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ public class ActivityConfiguration {
.setScheduleToStartTimeout(Duration.ofDays(MAX_SYNC_TIMEOUT_DAYS))
.setCancellationType(ActivityCancellationType.WAIT_CANCELLATION_COMPLETED)
.setRetryOptions(TemporalUtils.NO_RETRY)
.setHeartbeatTimeout(Duration.ofSeconds(30))
.setHeartbeatTimeout(TemporalUtils.HEARTBEAT_TIMEOUT)
.build();

}