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

Include jobId in webhook messages #15704

Merged
merged 1 commit into from
Aug 16, 2022
Merged

Include jobId in webhook messages #15704

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
4 changes: 2 additions & 2 deletions airbyte-notification/src/main/java/io/airbyte/notification/CustomerioNotificationClient.java
View file
Open in desktop
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
Open in desktop
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
Open in desktop
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
Open in desktop
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
Open in desktop
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
Open in desktop
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
Open in desktop
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
Open in desktop
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