-
Notifications
You must be signed in to change notification settings - Fork 4.1k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add JobErrorReporter for sending sync job connector failures to Sentry (
#13899) * skeleton for reporting connector errors to sentry * report on job failures instead of attempt failures * report sync job failures with relevant metadata using JobErrorReporter * send stack traces from python connectors to sentry * test JobCreationAndStatusUpdate and JobErrorReporter * logs * refactor into helper, initial tests * using sentry * run format * load reporting client from env * load sentry dsn from env * send java stack traces to sentry * test sentryclient, refactor to use Hub instance * ErrorReportingClient.report -> .reportJobFailureReason * inject exception helper, test stack trace parse error tagging * rm logs * more stack trace tests * remove logs * fix failing tests * rename ErrorReportingClient to JobErrorReportingClient * rename vars in docker-compose * Return an Optional instead of null when parsing stack traces * dont remove airbyte prefix when setting release name * from_trace_message static * remove failureSummary from jobfailure input, get from Job * send stacktrace string if we weren't able to parse * set deployment mode tag * update .env * just log if something goes wrong
- Loading branch information
1 parent
cc2b82c
commit d6d32b3
Showing
21 changed files
with
1,296 additions
and
7 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
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
102 changes: 102 additions & 0 deletions
102
...e/src/main/java/io/airbyte/scheduler/persistence/job_error_reporter/JobErrorReporter.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,102 @@ | ||
/* | ||
* Copyright (c) 2022 Airbyte, Inc., all rights reserved. | ||
*/ | ||
|
||
package io.airbyte.scheduler.persistence.job_error_reporter; | ||
|
||
import io.airbyte.config.AttemptFailureSummary; | ||
import io.airbyte.config.Configs.DeploymentMode; | ||
import io.airbyte.config.FailureReason; | ||
import io.airbyte.config.FailureReason.FailureOrigin; | ||
import io.airbyte.config.JobSyncConfig; | ||
import io.airbyte.config.StandardDestinationDefinition; | ||
import io.airbyte.config.StandardSourceDefinition; | ||
import io.airbyte.config.StandardWorkspace; | ||
import io.airbyte.config.persistence.ConfigRepository; | ||
import java.util.HashMap; | ||
import java.util.List; | ||
import java.util.UUID; | ||
import org.slf4j.Logger; | ||
import org.slf4j.LoggerFactory; | ||
|
||
public class JobErrorReporter { | ||
|
||
private static final Logger LOGGER = LoggerFactory.getLogger(JobErrorReporter.class); | ||
|
||
private static final String FROM_TRACE_MESSAGE = "from_trace_message"; | ||
private static final String DEPLOYMENT_MODE_META_KEY = "deployment_mode"; | ||
private static final String AIRBYTE_VERSION_META_KEY = "airbyte_version"; | ||
private static final String FAILURE_ORIGIN_META_KEY = "failure_origin"; | ||
private static final String FAILURE_TYPE_META_KEY = "failure_type"; | ||
private static final String CONNECTION_ID_META_KEY = "connection_id"; | ||
private static final String CONNECTOR_NAME_META_KEY = "connector_name"; | ||
private static final String CONNECTOR_DEFINITION_ID_META_KEY = "connector_definition_id"; | ||
private static final String CONNECTOR_RELEASE_STAGE_META_KEY = "connector_release_stage"; | ||
|
||
private final ConfigRepository configRepository; | ||
private final DeploymentMode deploymentMode; | ||
private final String airbyteVersion; | ||
private final JobErrorReportingClient jobErrorReportingClient; | ||
|
||
public JobErrorReporter(final ConfigRepository configRepository, | ||
final DeploymentMode deploymentMode, | ||
final String airbyteVersion, | ||
final JobErrorReportingClient jobErrorReportingClient) { | ||
|
||
this.configRepository = configRepository; | ||
this.deploymentMode = deploymentMode; | ||
this.airbyteVersion = airbyteVersion; | ||
this.jobErrorReportingClient = jobErrorReportingClient; | ||
} | ||
|
||
/** | ||
* Reports a Sync Job's connector-caused FailureReasons to the JobErrorReportingClient | ||
* | ||
* @param connectionId - connection that had the failure | ||
* @param failureSummary - final attempt failure summary | ||
* @param jobSyncConfig - config for the sync job | ||
*/ | ||
public void reportSyncJobFailure(final UUID connectionId, final AttemptFailureSummary failureSummary, final JobSyncConfig jobSyncConfig) { | ||
final List<FailureReason> traceMessageFailures = failureSummary.getFailures().stream() | ||
.filter(failure -> failure.getMetadata() != null && failure.getMetadata().getAdditionalProperties().containsKey(FROM_TRACE_MESSAGE)) | ||
.toList(); | ||
|
||
final StandardWorkspace workspace = configRepository.getStandardWorkspaceFromConnection(connectionId, true); | ||
|
||
for (final FailureReason failureReason : traceMessageFailures) { | ||
final FailureOrigin failureOrigin = failureReason.getFailureOrigin(); | ||
|
||
final HashMap<String, String> metadata = new HashMap<>(); | ||
metadata.put(CONNECTION_ID_META_KEY, connectionId.toString()); | ||
metadata.put(AIRBYTE_VERSION_META_KEY, airbyteVersion); | ||
metadata.put(DEPLOYMENT_MODE_META_KEY, deploymentMode.name()); | ||
metadata.put(FAILURE_ORIGIN_META_KEY, failureOrigin.value()); | ||
metadata.put(FAILURE_TYPE_META_KEY, failureReason.getFailureType().value()); | ||
|
||
try { | ||
if (failureOrigin == FailureOrigin.SOURCE) { | ||
final StandardSourceDefinition sourceDefinition = configRepository.getSourceDefinitionFromConnection(connectionId); | ||
final String dockerImage = jobSyncConfig.getSourceDockerImage(); | ||
|
||
metadata.put(CONNECTOR_DEFINITION_ID_META_KEY, sourceDefinition.getSourceDefinitionId().toString()); | ||
metadata.put(CONNECTOR_NAME_META_KEY, sourceDefinition.getName()); | ||
metadata.put(CONNECTOR_RELEASE_STAGE_META_KEY, sourceDefinition.getReleaseStage().value()); | ||
|
||
jobErrorReportingClient.reportJobFailureReason(workspace, failureReason, dockerImage, metadata); | ||
} else if (failureOrigin == FailureOrigin.DESTINATION) { | ||
final StandardDestinationDefinition destinationDefinition = configRepository.getDestinationDefinitionFromConnection(connectionId); | ||
final String dockerImage = jobSyncConfig.getDestinationDockerImage(); | ||
|
||
metadata.put(CONNECTOR_DEFINITION_ID_META_KEY, destinationDefinition.getDestinationDefinitionId().toString()); | ||
metadata.put(CONNECTOR_NAME_META_KEY, destinationDefinition.getName()); | ||
metadata.put(CONNECTOR_RELEASE_STAGE_META_KEY, destinationDefinition.getReleaseStage().value()); | ||
|
||
jobErrorReportingClient.reportJobFailureReason(workspace, failureReason, dockerImage, metadata); | ||
} | ||
} catch (final Exception e) { | ||
LOGGER.error("Error when reporting job failure reason: {}", failureReason, e); | ||
} | ||
} | ||
} | ||
|
||
} |
21 changes: 21 additions & 0 deletions
21
...ain/java/io/airbyte/scheduler/persistence/job_error_reporter/JobErrorReportingClient.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,21 @@ | ||
/* | ||
* Copyright (c) 2022 Airbyte, Inc., all rights reserved. | ||
*/ | ||
|
||
package io.airbyte.scheduler.persistence.job_error_reporter; | ||
|
||
import io.airbyte.config.FailureReason; | ||
import io.airbyte.config.StandardWorkspace; | ||
import java.util.Map; | ||
|
||
/** | ||
* A generic interface for a client that reports errors | ||
*/ | ||
public interface JobErrorReportingClient { | ||
|
||
/** | ||
* Report a job failure reason | ||
*/ | ||
void reportJobFailureReason(StandardWorkspace workspace, final FailureReason reason, final String dockerImage, Map<String, String> metadata); | ||
|
||
} |
25 changes: 25 additions & 0 deletions
25
...a/io/airbyte/scheduler/persistence/job_error_reporter/JobErrorReportingClientFactory.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,25 @@ | ||
/* | ||
* Copyright (c) 2022 Airbyte, Inc., all rights reserved. | ||
*/ | ||
|
||
package io.airbyte.scheduler.persistence.job_error_reporter; | ||
|
||
import io.airbyte.config.Configs; | ||
import io.airbyte.config.Configs.JobErrorReportingStrategy; | ||
|
||
public class JobErrorReportingClientFactory { | ||
|
||
/** | ||
* Creates an error reporting client based on the desired strategy to use | ||
* | ||
* @param strategy - which type of error reporting client should be created | ||
* @return JobErrorReportingClient | ||
*/ | ||
public static JobErrorReportingClient getClient(final JobErrorReportingStrategy strategy, final Configs configs) { | ||
return switch (strategy) { | ||
case SENTRY -> new SentryJobErrorReportingClient(configs.getJobErrorReportingSentryDSN(), new SentryExceptionHelper()); | ||
case LOGGING -> new LoggingJobErrorReportingClient(); | ||
}; | ||
} | ||
|
||
} |
29 changes: 29 additions & 0 deletions
29
...a/io/airbyte/scheduler/persistence/job_error_reporter/LoggingJobErrorReportingClient.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,29 @@ | ||
/* | ||
* Copyright (c) 2022 Airbyte, Inc., all rights reserved. | ||
*/ | ||
|
||
package io.airbyte.scheduler.persistence.job_error_reporter; | ||
|
||
import io.airbyte.config.FailureReason; | ||
import io.airbyte.config.StandardWorkspace; | ||
import java.util.Map; | ||
import org.slf4j.Logger; | ||
import org.slf4j.LoggerFactory; | ||
|
||
public class LoggingJobErrorReportingClient implements JobErrorReportingClient { | ||
|
||
private static final Logger LOGGER = LoggerFactory.getLogger(LoggingJobErrorReportingClient.class); | ||
|
||
@Override | ||
public void reportJobFailureReason(final StandardWorkspace workspace, | ||
final FailureReason reason, | ||
final String dockerImage, | ||
final Map<String, String> metadata) { | ||
LOGGER.info("Report Job Error -> workspaceId: {}, dockerImage: {}, failureReason: {}, metadata: {}", | ||
workspace.getWorkspaceId(), | ||
dockerImage, | ||
reason, | ||
metadata); | ||
} | ||
|
||
} |
Oops, something went wrong.