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

Trace refresh schema operations #22326

Merged
merged 2 commits into from
Feb 2, 2023
Merged

Trace refresh schema operations #22326

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
0c27c74
Trace refresh schema operations
jdpgrailsdev Feb 2, 2023
93e6124
Merge branch 'master' into jonathan/trace-refresh-schema
jdpgrailsdev Feb 2, 2023
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
28 changes: 19 additions & 9 deletions ...yte-workers/src/main/java/io/airbyte/workers/temporal/sync/RefreshSchemaActivityImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,8 @@
package io.airbyte.workers.temporal.sync;

import static io.airbyte.metrics.lib.ApmTraceConstants.ACTIVITY_TRACE_OPERATION_NAME;
import static io.airbyte.metrics.lib.ApmTraceConstants.Tags.CONNECTION_ID_KEY;
import static io.airbyte.metrics.lib.ApmTraceConstants.Tags.SOURCE_ID_KEY;

import datadog.trace.api.Trace;
import io.airbyte.api.client.generated.SourceApi;
Expand All @@ -13,8 +15,10 @@
import io.airbyte.api.client.model.generated.SourceDiscoverSchemaRequestBody;
import io.airbyte.api.client.model.generated.SourceIdRequestBody;
import io.airbyte.commons.features.EnvVariableFeatureFlags;
import io.airbyte.metrics.lib.ApmTraceUtils;
import jakarta.inject.Singleton;
import java.time.OffsetDateTime;
import java.util.Map;
import java.util.UUID;
import lombok.extern.slf4j.Slf4j;

Expand All @@ -25,48 +29,54 @@ public class RefreshSchemaActivityImpl implements RefreshSchemaActivity {
private final SourceApi sourceApi;
private final EnvVariableFeatureFlags envVariableFeatureFlags;

public RefreshSchemaActivityImpl(SourceApi sourceApi,
EnvVariableFeatureFlags envVariableFeatureFlags) {
public RefreshSchemaActivityImpl(final SourceApi sourceApi,
final EnvVariableFeatureFlags envVariableFeatureFlags) {
this.sourceApi = sourceApi;
this.envVariableFeatureFlags = envVariableFeatureFlags;
}

@Override
@Trace(operationName = ACTIVITY_TRACE_OPERATION_NAME)
public boolean shouldRefreshSchema(UUID sourceCatalogId) {
public boolean shouldRefreshSchema(final UUID sourceCatalogId) {
if (!envVariableFeatureFlags.autoDetectSchema()) {
return false;
}

ApmTraceUtils.addTagsToTrace(Map.of(SOURCE_ID_KEY, sourceCatalogId));
return !schemaRefreshRanRecently(sourceCatalogId);
}

@Override
public void refreshSchema(UUID sourceCatalogId, UUID connectionId) {
@Trace(operationName = ACTIVITY_TRACE_OPERATION_NAME)
public void refreshSchema(final UUID sourceCatalogId, final UUID connectionId) {
if (!envVariableFeatureFlags.autoDetectSchema()) {
return;
}

SourceDiscoverSchemaRequestBody requestBody =
ApmTraceUtils.addTagsToTrace(Map.of(CONNECTION_ID_KEY, connectionId, SOURCE_ID_KEY, sourceCatalogId));

final SourceDiscoverSchemaRequestBody requestBody =
new SourceDiscoverSchemaRequestBody().sourceId(sourceCatalogId).disableCache(true).connectionId(connectionId);

try {
sourceApi.discoverSchemaForSource(requestBody);
} catch (final Exception e) {
ApmTraceUtils.addExceptionToTrace(e);
// catching this exception because we don't want to block replication due to a failed schema refresh
log.error("Attempted schema refresh, but failed with error: ", e);
}
}

private boolean schemaRefreshRanRecently(UUID sourceCatalogId) {
private boolean schemaRefreshRanRecently(final UUID sourceCatalogId) {
try {
SourceIdRequestBody sourceIdRequestBody = new SourceIdRequestBody().sourceId(sourceCatalogId);
ActorCatalogWithUpdatedAt mostRecentFetchEvent = sourceApi.getMostRecentSourceActorCatalog(sourceIdRequestBody);
final SourceIdRequestBody sourceIdRequestBody = new SourceIdRequestBody().sourceId(sourceCatalogId);
final ActorCatalogWithUpdatedAt mostRecentFetchEvent = sourceApi.getMostRecentSourceActorCatalog(sourceIdRequestBody);
if (mostRecentFetchEvent.getUpdatedAt() == null) {
return false;
}
return mostRecentFetchEvent.getUpdatedAt() > OffsetDateTime.now().minusHours(24l).toEpochSecond();
} catch (ApiException e) {
} catch (final ApiException e) {
ApmTraceUtils.addExceptionToTrace(e);
// catching this exception because we don't want to block replication due to a failed schema refresh
log.info("Encountered an error fetching most recent actor catalog fetch event: ", e);
return true;
Expand Down