Skip to content

Commit

Permalink
Include jobId in webhook messages (#15704)
Browse files Browse the repository at this point in the history
  • Loading branch information
@akawalsky
akawalsky authored Aug 16, 2022
1 parent 83dd2c3 commit f0a69c4
Show file tree
Hide file tree
Showing 8 changed files with 27 additions and 18 deletions.
4 changes: 2 additions & 2 deletions airbyte-notification/src/main/java/io/airbyte/notification/CustomerioNotificationClient.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -63,13 +63,13 @@ public CustomerioNotificationClient(final Notification notification,
}

@Override
public boolean notifyJobFailure(final String sourceConnector, final String destinationConnector, final String jobDescription, final String logUrl)
public boolean notifyJobFailure(final String sourceConnector, final String destinationConnector, final String jobDescription, final String logUrl, final Long jobId)
throws IOException, InterruptedException {
throw new NotImplementedException();
}

@Override
public boolean notifyJobSuccess(final String sourceConnector, final String destinationConnector, final String jobDescription, final String logUrl)
public boolean notifyJobSuccess(final String sourceConnector, final String destinationConnector, final String jobDescription, final String logUrl, final Long jobId)
throws IOException, InterruptedException {
throw new NotImplementedException();
}
Expand Down
6 changes: 4 additions & 2 deletions airbyte-notification/src/main/java/io/airbyte/notification/NotificationClient.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,14 +23,16 @@ public abstract boolean notifyJobFailure(
String sourceConnector,
String destinationConnector,
String jobDescription,
String logUrl)
String logUrl,
Long jobId)
throws IOException, InterruptedException;

public abstract boolean notifyJobSuccess(
String sourceConnector,
String destinationConnector,
String jobDescription,
String logUrl)
String logUrl,
Long jobId)
throws IOException, InterruptedException;

public abstract boolean notifyConnectionDisabled(String receiverEmail,
Expand Down
10 changes: 6 additions & 4 deletions airbyte-notification/src/main/java/io/airbyte/notification/SlackNotificationClient.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -44,25 +44,27 @@ public SlackNotificationClient(final Notification notification) {
}

@Override
public boolean notifyJobFailure(final String sourceConnector, final String destinationConnector, final String jobDescription, final String logUrl)
public boolean notifyJobFailure(final String sourceConnector, final String destinationConnector, final String jobDescription, final String logUrl, final Long jobId)
throws IOException, InterruptedException {
return notifyFailure(renderTemplate(
"slack/failure_slack_notification_template.txt",
sourceConnector,
destinationConnector,
jobDescription,
logUrl));
logUrl,
String.valueOf(jobId)));
}

@Override
public boolean notifyJobSuccess(final String sourceConnector, final String destinationConnector, final String jobDescription, final String logUrl)
public boolean notifyJobSuccess(final String sourceConnector, final String destinationConnector, final String jobDescription, final String logUrl, final Long jobId)
throws IOException, InterruptedException {
return notifySuccess(renderTemplate(
"slack/success_slack_notification_template.txt",
sourceConnector,
destinationConnector,
jobDescription,
logUrl));
logUrl,
String.valueOf(jobId)));
}

@Override
Expand Down
2 changes: 2 additions & 0 deletions airbyte-notification/src/main/resources/slack/failure_slack_notification_template.txt
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,3 +2,5 @@ Your connection from %s to %s just failed...
This happened with %s

You can access its logs here: %s

Job ID: %s
2 changes: 2 additions & 0 deletions airbyte-notification/src/main/resources/slack/success_slack_notification_template.txt
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,3 +2,5 @@ Your connection from %s to %s succeeded
This was for %s

You can access its logs here: %s

Job ID: %s
9 changes: 5 additions & 4 deletions airbyte-notification/src/test/java/io/airbyte/notification/SlackNotificationClientTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@ class SlackNotificationClientTest {
private static final String JOB_DESCRIPTION = "job description";
private static final String LOG_URL = "logUrl";
private static final String SOURCE_TEST = "source-test";
private static final Long JOB_ID = 1L;

public static final String WEBHOOK_URL = "http://localhost:";
private static final String EXPECTED_FAIL_MESSAGE = "Your connection from source-test to destination-test just failed...\n"
Expand Down Expand Up @@ -81,15 +82,15 @@ void testBadWebhookUrl() {
new SlackNotificationClient(new Notification()
.withNotificationType(NotificationType.SLACK)
.withSlackConfiguration(new SlackNotificationConfiguration().withWebhook(WEBHOOK_URL + server.getAddress().getPort() + "/bad")));
assertThrows(IOException.class, () -> client.notifyJobFailure(SOURCE_TEST, DESTINATION_TEST, JOB_DESCRIPTION, LOG_URL));
assertThrows(IOException.class, () -> client.notifyJobFailure(SOURCE_TEST, DESTINATION_TEST, JOB_DESCRIPTION, LOG_URL, JOB_ID));
}

@Test
void testEmptyWebhookUrl() throws IOException, InterruptedException {
final SlackNotificationClient client =
new SlackNotificationClient(
new Notification().withNotificationType(NotificationType.SLACK).withSlackConfiguration(new SlackNotificationConfiguration()));
assertFalse(client.notifyJobFailure(SOURCE_TEST, DESTINATION_TEST, JOB_DESCRIPTION, LOG_URL));
assertFalse(client.notifyJobFailure(SOURCE_TEST, DESTINATION_TEST, JOB_DESCRIPTION, LOG_URL, JOB_ID));
}

@Test
Expand All @@ -111,7 +112,7 @@ void testNotifyJobFailure() throws IOException, InterruptedException {
new SlackNotificationClient(new Notification()
.withNotificationType(NotificationType.SLACK)
.withSlackConfiguration(new SlackNotificationConfiguration().withWebhook(WEBHOOK_URL + server.getAddress().getPort() + TEST_PATH)));
assertTrue(client.notifyJobFailure(SOURCE_TEST, DESTINATION_TEST, JOB_DESCRIPTION, LOG_URL));
assertTrue(client.notifyJobFailure(SOURCE_TEST, DESTINATION_TEST, JOB_DESCRIPTION, LOG_URL, JOB_ID));
}

@Test
Expand All @@ -122,7 +123,7 @@ void testNotifyJobSuccess() throws IOException, InterruptedException {
.withNotificationType(NotificationType.SLACK)
.withSendOnSuccess(true)
.withSlackConfiguration(new SlackNotificationConfiguration().withWebhook(WEBHOOK_URL + server.getAddress().getPort() + TEST_PATH)));
assertTrue(client.notifyJobSuccess(SOURCE_TEST, DESTINATION_TEST, JOB_DESCRIPTION, LOG_URL));
assertTrue(client.notifyJobSuccess(SOURCE_TEST, DESTINATION_TEST, JOB_DESCRIPTION, LOG_URL, JOB_ID));
}

@Test
Expand Down
4 changes: 2 additions & 2 deletions ...ler/scheduler-persistence/src/main/java/io/airbyte/scheduler/persistence/JobNotifier.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -103,12 +103,12 @@ private void notifyJob(final String reason,
MoreMaps.merge(jobMetadata, sourceMetadata, destinationMetadata, notificationMetadata.build()));

if (FAILURE_NOTIFICATION == action) {
if (!notificationClient.notifyJobFailure(sourceConnector, destinationConnector, jobDescription, logUrl)) {
if (!notificationClient.notifyJobFailure(sourceConnector, destinationConnector, jobDescription, logUrl, job.getId())) {
LOGGER.warn("Failed to successfully notify failure: {}", notification);
}
break;
} else if (SUCCESS_NOTIFICATION == action) {
if (!notificationClient.notifyJobSuccess(sourceConnector, destinationConnector, jobDescription, logUrl)) {
if (!notificationClient.notifyJobSuccess(sourceConnector, destinationConnector, jobDescription, logUrl, job.getId())) {
LOGGER.warn("Failed to successfully notify success: {}", notification);
}
break;
Expand Down
8 changes: 4 additions & 4 deletions ...scheduler-persistence/src/test/java/io/airbyte/scheduler/persistence/JobNotifierTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,8 +4,7 @@

package io.airbyte.scheduler.persistence;

import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.*;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
Expand Down Expand Up @@ -85,7 +84,7 @@ void testFailJob() throws IOException, InterruptedException, JsonValidationExcep
when(configRepository.getStandardDestinationDefinition(any())).thenReturn(destinationDefinition);
when(configRepository.getStandardWorkspace(WORKSPACE_ID, true)).thenReturn(getWorkspace());
when(workspaceHelper.getWorkspaceForJobIdIgnoreExceptions(job.getId())).thenReturn(WORKSPACE_ID);
when(notificationClient.notifyJobFailure(anyString(), anyString(), anyString(), anyString())).thenReturn(true);
when(notificationClient.notifyJobFailure(anyString(), anyString(), anyString(), anyString(), anyLong())).thenReturn(true);

jobNotifier.failJob("JobNotifierTest was running", job);
final DateTimeFormatter formatter = DateTimeFormatter.ofLocalizedDateTime(FormatStyle.FULL).withZone(ZoneId.systemDefault());
Expand All @@ -94,7 +93,8 @@ void testFailJob() throws IOException, InterruptedException, JsonValidationExcep
"destination-test",
String.format("sync started on %s, running for 1 day 10 hours 17 minutes 36 seconds, as the JobNotifierTest was running.",
formatter.format(Instant.ofEpochSecond(job.getStartedAtInSecond().get()))),
String.format("http://localhost:8000/workspaces/%s/connections/%s", WORKSPACE_ID, job.getScope()));
String.format("http://localhost:8000/workspaces/%s/connections/%s", WORKSPACE_ID, job.getScope()),
job.getId());

final Builder<String, Object> metadata = ImmutableMap.builder();
metadata.put("connection_id", UUID.fromString(job.getScope()));
Expand Down

0 comments on commit f0a69c4

Please sign in to comment.