Viacheslav Klimov
1 week ago
committed by
GitHub
GitHub
24 changed files with 1064 additions and 27 deletions
@ -0,0 +1,34 @@ |
|||
/** |
|||
* Copyright © 2016-2026 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.monitoring.data.notification; |
|||
|
|||
public record AffectedService(String name, Status status, int failureCount) { |
|||
|
|||
public enum Status { FAILING, RECOVERED, HIGH_LATENCY } |
|||
|
|||
public static AffectedService failing(String name, int failureCount) { |
|||
return new AffectedService(name, Status.FAILING, failureCount); |
|||
} |
|||
|
|||
public static AffectedService recovered(String name) { |
|||
return new AffectedService(name, Status.RECOVERED, 0); |
|||
} |
|||
|
|||
public static AffectedService highLatency(String name) { |
|||
return new AffectedService(name, Status.HIGH_LATENCY, 0); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,22 @@ |
|||
/** |
|||
* Copyright © 2016-2026 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.monitoring.data.notification; |
|||
|
|||
public interface ShortNameProvider { |
|||
|
|||
String getShortName(); |
|||
|
|||
} |
|||
@ -0,0 +1,125 @@ |
|||
/** |
|||
* Copyright © 2016-2026 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.monitoring.notification.channels.impl; |
|||
|
|||
import com.slack.api.Slack; |
|||
import com.slack.api.SlackConfig; |
|||
import com.slack.api.methods.MethodsClient; |
|||
import com.slack.api.methods.SlackApiTextResponse; |
|||
import com.slack.api.methods.request.chat.ChatPostMessageRequest; |
|||
import com.slack.api.methods.request.chat.ChatUpdateRequest; |
|||
import com.slack.api.methods.response.chat.ChatPostMessageResponse; |
|||
import com.slack.api.methods.response.chat.ChatUpdateResponse; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
|
|||
@Slf4j |
|||
public class SlackApiClient { |
|||
|
|||
private static final int DEFAULT_CALL_TIMEOUT_MS = 5000; |
|||
|
|||
private final Slack slack; |
|||
private final String botToken; |
|||
|
|||
public SlackApiClient(String botToken) { |
|||
this(botToken, DEFAULT_CALL_TIMEOUT_MS); |
|||
} |
|||
|
|||
public SlackApiClient(String botToken, int callTimeoutMs) { |
|||
this.botToken = botToken; |
|||
SlackConfig config = new SlackConfig(); |
|||
config.setHttpClientCallTimeoutMillis(callTimeoutMs); |
|||
config.setHttpClientReadTimeoutMillis(callTimeoutMs); |
|||
config.setHttpClientWriteTimeoutMillis(callTimeoutMs); |
|||
this.slack = Slack.getInstance(config); |
|||
} |
|||
|
|||
public String postMessage(String channelId, String text) { |
|||
ChatPostMessageRequest request = ChatPostMessageRequest.builder() |
|||
.channel(channelId) |
|||
.text(text) |
|||
.build(); |
|||
ChatPostMessageResponse response = sendRequest(request); |
|||
return response.getTs(); |
|||
} |
|||
|
|||
public String postThreadReply(String channelId, String threadTs, String text) { |
|||
ChatPostMessageRequest request = ChatPostMessageRequest.builder() |
|||
.channel(channelId) |
|||
.text(text) |
|||
.threadTs(threadTs) |
|||
.build(); |
|||
ChatPostMessageResponse response = sendRequest(request); |
|||
return response.getTs(); |
|||
} |
|||
|
|||
public void close() { |
|||
try { |
|||
slack.close(); |
|||
} catch (Exception e) { |
|||
log.warn("Failed to close Slack client", e); |
|||
} |
|||
} |
|||
|
|||
public void updateMessage(String channelId, String ts, String text) { |
|||
ChatUpdateRequest request = ChatUpdateRequest.builder() |
|||
.channel(channelId) |
|||
.ts(ts) |
|||
.text(text) |
|||
.build(); |
|||
MethodsClient client = slack.methods(botToken); |
|||
ChatUpdateResponse response; |
|||
try { |
|||
response = client.chatUpdate(request); |
|||
} catch (Exception e) { |
|||
throw new RuntimeException("Failed to update Slack message: " + e.getMessage(), e); |
|||
} |
|||
checkResponse(response); |
|||
} |
|||
|
|||
private ChatPostMessageResponse sendRequest(ChatPostMessageRequest request) { |
|||
MethodsClient client = slack.methods(botToken); |
|||
ChatPostMessageResponse response; |
|||
try { |
|||
response = client.chatPostMessage(request); |
|||
} catch (Exception e) { |
|||
throw new RuntimeException("Failed to send Slack message: " + e.getMessage(), e); |
|||
} |
|||
checkResponse(response); |
|||
return response; |
|||
} |
|||
|
|||
private void checkResponse(SlackApiTextResponse response) { |
|||
if (response.isOk()) { |
|||
return; |
|||
} |
|||
String error = response.getError(); |
|||
if (error != null) { |
|||
switch (error) { |
|||
case "missing_scope" -> { |
|||
String neededScope = response.getNeeded(); |
|||
error = "bot token scope '" + neededScope + "' is needed"; |
|||
} |
|||
case "not_in_channel" -> error = "app needs to be added to the channel"; |
|||
} |
|||
} else if (response.getWarning() != null) { |
|||
error = "warning: " + response.getWarning(); |
|||
} else { |
|||
error = "unknown error"; |
|||
} |
|||
throw new RuntimeException("Slack API error: " + error); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,45 @@ |
|||
/** |
|||
* Copyright © 2016-2026 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.monitoring.notification.channels.impl; |
|||
|
|||
import org.thingsboard.monitoring.notification.incident.IncidentTransport; |
|||
|
|||
public class SlackIncidentTransport implements IncidentTransport { |
|||
|
|||
private final SlackApiClient slackApiClient; |
|||
private final String channelId; |
|||
|
|||
public SlackIncidentTransport(SlackApiClient slackApiClient, String channelId) { |
|||
this.slackApiClient = slackApiClient; |
|||
this.channelId = channelId; |
|||
} |
|||
|
|||
@Override |
|||
public String postIncident(String text) { |
|||
return slackApiClient.postMessage(channelId, text); |
|||
} |
|||
|
|||
@Override |
|||
public void postThreadReply(String threadId, String text) { |
|||
slackApiClient.postThreadReply(channelId, threadId, text); |
|||
} |
|||
|
|||
@Override |
|||
public void updateIncident(String threadId, String text) { |
|||
slackApiClient.updateMessage(channelId, threadId, text); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,292 @@ |
|||
/** |
|||
* Copyright © 2016-2026 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.monitoring.notification.incident; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.thingsboard.monitoring.data.notification.AffectedService; |
|||
|
|||
import java.time.Duration; |
|||
import java.time.Instant; |
|||
import java.util.LinkedHashMap; |
|||
import java.util.LinkedHashSet; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.Set; |
|||
import java.util.concurrent.Executors; |
|||
import java.util.concurrent.ScheduledExecutorService; |
|||
import java.util.concurrent.ScheduledFuture; |
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
/** |
|||
* Thread-safety: all public entry points and scheduled callbacks are {@code synchronized} on the |
|||
* manager instance. Transport I/O is performed while the monitor is held. This is safe under the |
|||
* assumptions that (a) the transport enforces a short per-call timeout, and (b) notification |
|||
* producers are single-threaded; see the Slack client (default 5s) for the Slack-based transport. |
|||
*/ |
|||
@Slf4j |
|||
public class IncidentManager { |
|||
|
|||
private final IncidentTransport transport; |
|||
private final long resolutionTimeoutSeconds; |
|||
private final String messagePrefix; |
|||
private final boolean tagChannel; |
|||
private final ScheduledExecutorService scheduler; |
|||
|
|||
private String activeIncidentThreadId; |
|||
private ScheduledFuture<?> resolutionTask; |
|||
private ScheduledFuture<?> durationUpdateTask; |
|||
private Instant incidentStartTime; |
|||
private Instant lastAlertTime; |
|||
private final Map<String, Integer> failingServices = new LinkedHashMap<>(); |
|||
private final Map<String, Integer> recoveredServices = new LinkedHashMap<>(); |
|||
private final Set<String> highLatencyServices = new LinkedHashSet<>(); |
|||
|
|||
public IncidentManager(IncidentTransport transport, long resolutionTimeoutSeconds, |
|||
String messagePrefix, boolean tagChannel) { |
|||
this.transport = transport; |
|||
this.resolutionTimeoutSeconds = resolutionTimeoutSeconds; |
|||
this.messagePrefix = messagePrefix; |
|||
this.tagChannel = tagChannel; |
|||
this.scheduler = Executors.newSingleThreadScheduledExecutor(r -> { |
|||
Thread t = new Thread(r, "incident-manager"); |
|||
t.setDaemon(true); |
|||
return t; |
|||
}); |
|||
} |
|||
|
|||
public synchronized void sendAlert(String message, List<AffectedService> affectedServices) { |
|||
try { |
|||
if (activeIncidentThreadId == null) { |
|||
if (affectedServices.stream().allMatch(s -> s.status() == AffectedService.Status.RECOVERED)) { |
|||
return; |
|||
} |
|||
incidentStartTime = Instant.now(); |
|||
failingServices.clear(); |
|||
recoveredServices.clear(); |
|||
highLatencyServices.clear(); |
|||
applyAffectedServices(affectedServices); |
|||
activeIncidentThreadId = transport.postIncident(buildOngoingMessageText()); |
|||
startDurationUpdater(); |
|||
log.info("New incident created, thread id: {}", activeIncidentThreadId); |
|||
} else if (applyAffectedServices(affectedServices)) { |
|||
safeUpdateHeader(); |
|||
} |
|||
|
|||
try { |
|||
transport.postThreadReply(activeIncidentThreadId, message); |
|||
log.debug("Alert added to incident thread {}", activeIncidentThreadId); |
|||
} catch (Exception e) { |
|||
log.error("Failed to post alert to incident thread {}", activeIncidentThreadId, e); |
|||
} |
|||
} finally { |
|||
if (activeIncidentThreadId != null) { |
|||
lastAlertTime = Instant.now(); |
|||
// High latency is a warning only — it has no explicit recovery signal
|
|||
// (HighLatencyNotification fires only when something is above threshold),
|
|||
// so resolution hinges on failing services alone.
|
|||
if (failingServices.isEmpty()) { |
|||
resetResolutionTimer(); |
|||
} else { |
|||
cancelResolutionTimer(); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
|
|||
private boolean applyAffectedServices(List<AffectedService> affectedServices) { |
|||
boolean changed = false; |
|||
Set<String> latencySnapshot = null; |
|||
for (AffectedService service : affectedServices) { |
|||
String name = service.name(); |
|||
switch (service.status()) { |
|||
case FAILING -> { |
|||
Integer prev = failingServices.put(name, service.failureCount()); |
|||
if (prev == null || prev.intValue() != service.failureCount()) { |
|||
changed = true; |
|||
} |
|||
if (recoveredServices.remove(name) != null) { |
|||
changed = true; |
|||
} |
|||
} |
|||
case RECOVERED -> { |
|||
Integer lastFailureCount = failingServices.remove(name); |
|||
if (lastFailureCount != null) { |
|||
recoveredServices.put(name, lastFailureCount); |
|||
changed = true; |
|||
} |
|||
} |
|||
case HIGH_LATENCY -> { |
|||
if (latencySnapshot == null) { |
|||
latencySnapshot = new LinkedHashSet<>(); |
|||
} |
|||
latencySnapshot.add(name); |
|||
} |
|||
} |
|||
} |
|||
// HighLatencyNotification carries the full current set of high latencies, so treat it as a
|
|||
// snapshot: replace highLatencyServices entirely. Without this, a brief spike would stay
|
|||
// yellow in the header until the incident resolves.
|
|||
if (latencySnapshot != null && !latencySnapshot.equals(highLatencyServices)) { |
|||
highLatencyServices.clear(); |
|||
highLatencyServices.addAll(latencySnapshot); |
|||
changed = true; |
|||
} |
|||
return changed; |
|||
} |
|||
|
|||
private String buildOngoingMessageText() { |
|||
StringBuilder sb = new StringBuilder(); |
|||
if (tagChannel) { |
|||
sb.append("<!channel> "); |
|||
} |
|||
if (messagePrefix != null && !messagePrefix.isEmpty()) { |
|||
sb.append("*").append(messagePrefix).append("*"); |
|||
} |
|||
sb.append(" :rotating_light:"); |
|||
Duration elapsed = Duration.between(incidentStartTime, Instant.now()); |
|||
if (elapsed.toMinutes() >= 1) { |
|||
sb.append(" (").append(formatDuration(elapsed)).append(")"); |
|||
} |
|||
if (hasAffected()) { |
|||
sb.append(" | ").append(formatAffectedServices()); |
|||
} |
|||
return sb.toString(); |
|||
} |
|||
|
|||
private boolean hasAffected() { |
|||
return !failingServices.isEmpty() || !recoveredServices.isEmpty() || !highLatencyServices.isEmpty(); |
|||
} |
|||
|
|||
private void safeUpdateHeader() { |
|||
try { |
|||
transport.updateIncident(activeIncidentThreadId, buildOngoingMessageText()); |
|||
} catch (Exception e) { |
|||
log.error("Failed to update incident message", e); |
|||
} |
|||
} |
|||
|
|||
private void resetResolutionTimer() { |
|||
cancelResolutionTimer(); |
|||
resolutionTask = scheduler.schedule(this::resolveIncident, resolutionTimeoutSeconds, TimeUnit.SECONDS); |
|||
} |
|||
|
|||
private void cancelResolutionTimer() { |
|||
if (resolutionTask != null) { |
|||
resolutionTask.cancel(false); |
|||
resolutionTask = null; |
|||
} |
|||
} |
|||
|
|||
private void startDurationUpdater() { |
|||
if (durationUpdateTask != null) { |
|||
durationUpdateTask.cancel(false); |
|||
} |
|||
durationUpdateTask = scheduler.scheduleAtFixedRate(this::updateDuration, 60, 60, TimeUnit.SECONDS); |
|||
} |
|||
|
|||
private synchronized void updateDuration() { |
|||
if (activeIncidentThreadId == null) { |
|||
return; |
|||
} |
|||
safeUpdateHeader(); |
|||
} |
|||
|
|||
private void stopDurationUpdater() { |
|||
if (durationUpdateTask != null) { |
|||
durationUpdateTask.cancel(false); |
|||
durationUpdateTask = null; |
|||
} |
|||
} |
|||
|
|||
static String formatDuration(Duration duration) { |
|||
long totalMinutes = duration.toMinutes(); |
|||
if (totalMinutes < 60) { |
|||
return totalMinutes + "m"; |
|||
} |
|||
long hours = totalMinutes / 60; |
|||
long minutes = totalMinutes % 60; |
|||
return minutes > 0 ? hours + "h" + minutes + "m" : hours + "h"; |
|||
} |
|||
|
|||
synchronized void resolveIncident() { |
|||
if (activeIncidentThreadId == null) { |
|||
return; |
|||
} |
|||
String threadId = activeIncidentThreadId; |
|||
stopDurationUpdater(); |
|||
String resolutionMessage = buildResolutionMessage(); |
|||
activeIncidentThreadId = null; |
|||
resolutionTask = null; |
|||
failingServices.clear(); |
|||
recoveredServices.clear(); |
|||
highLatencyServices.clear(); |
|||
try { |
|||
transport.updateIncident(threadId, resolutionMessage); |
|||
log.info("Incident resolved (thread was {})", threadId); |
|||
} catch (Exception e) { |
|||
log.error("Failed to send incident resolution message", e); |
|||
} |
|||
} |
|||
|
|||
private String buildResolutionMessage() { |
|||
Duration totalDuration = lastAlertTime != null |
|||
? Duration.between(incidentStartTime, lastAlertTime) |
|||
: Duration.between(incidentStartTime, Instant.now()); |
|||
StringBuilder sb = new StringBuilder(); |
|||
if (messagePrefix != null && !messagePrefix.isEmpty()) { |
|||
sb.append("*").append(messagePrefix).append("*"); |
|||
} |
|||
sb.append(" :white_check_mark:"); |
|||
sb.append(" (").append(formatDuration(totalDuration)).append(")"); |
|||
if (hasAffected()) { |
|||
sb.append(" | ").append(formatAffectedServices()).append("\n"); |
|||
} |
|||
return sb.toString(); |
|||
} |
|||
|
|||
private String formatAffectedServices() { |
|||
StringBuilder sb = new StringBuilder(); |
|||
boolean first = true; |
|||
for (Map.Entry<String, Integer> entry : failingServices.entrySet()) { |
|||
if (!first) sb.append(", "); |
|||
sb.append(":red_circle: ").append(entry.getKey()).append(" (").append(entry.getValue()).append(")"); |
|||
first = false; |
|||
} |
|||
for (String name : highLatencyServices) { |
|||
if (!first) sb.append(", "); |
|||
sb.append(":large_yellow_circle: ").append(name); |
|||
first = false; |
|||
} |
|||
for (Map.Entry<String, Integer> entry : recoveredServices.entrySet()) { |
|||
if (!first) sb.append(", "); |
|||
sb.append(":large_green_circle: ").append(entry.getKey()).append(" (").append(entry.getValue()).append(")"); |
|||
first = false; |
|||
} |
|||
return sb.toString(); |
|||
} |
|||
|
|||
public void shutdown() { |
|||
scheduler.shutdownNow(); |
|||
try { |
|||
if (!scheduler.awaitTermination(5, TimeUnit.SECONDS)) { |
|||
log.warn("Incident scheduler did not terminate in time"); |
|||
} |
|||
} catch (InterruptedException e) { |
|||
Thread.currentThread().interrupt(); |
|||
} |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,26 @@ |
|||
/** |
|||
* Copyright © 2016-2026 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.monitoring.notification.incident; |
|||
|
|||
public interface IncidentTransport { |
|||
|
|||
String postIncident(String text); |
|||
|
|||
void postThreadReply(String threadId, String text); |
|||
|
|||
void updateIncident(String threadId, String text); |
|||
|
|||
} |
|||
@ -0,0 +1,97 @@ |
|||
/** |
|||
* Copyright © 2016-2026 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.monitoring.data.notification; |
|||
|
|||
import org.junit.jupiter.api.Test; |
|||
|
|||
import static org.assertj.core.api.Assertions.assertThat; |
|||
|
|||
class ServiceFailureNotificationTest { |
|||
|
|||
@Test |
|||
void stripResponseBodyRemovesNginxErrorHtml() { |
|||
String msg = "503 Service Temporarily Unavailable on POST request for \"https://domain/api/auth/login\": \"" |
|||
+ "<html><head><title>503 Service Temporarily Unavailable</title></head>" |
|||
+ "<body><center><h1>503 Service Temporarily Unavailable</h1></center><hr><center>nginx</center></body></html>\""; |
|||
|
|||
String sanitized = ServiceFailureNotification.stripResponseBody(msg); |
|||
|
|||
assertThat(sanitized) |
|||
.isEqualTo("503 Service Temporarily Unavailable on POST request for \"https://domain/api/auth/login\""); |
|||
} |
|||
|
|||
@Test |
|||
void stripResponseBodyRemovesDoctypeHtml() { |
|||
String msg = "500 Internal Server Error: \"<!DOCTYPE html><html>...</html>\""; |
|||
|
|||
String sanitized = ServiceFailureNotification.stripResponseBody(msg); |
|||
|
|||
assertThat(sanitized).isEqualTo("500 Internal Server Error"); |
|||
} |
|||
|
|||
@Test |
|||
void stripResponseBodyLeavesPlainMessagesUntouched() { |
|||
String msg = "Connection refused"; |
|||
assertThat(ServiceFailureNotification.stripResponseBody(msg)).isEqualTo(msg); |
|||
} |
|||
|
|||
@Test |
|||
void stripResponseBodyHandlesNull() { |
|||
assertThat(ServiceFailureNotification.stripResponseBody(null)).isNull(); |
|||
} |
|||
|
|||
@Test |
|||
void linkifyReplacesRequestForUrlWithSlackMrkdwnLink() { |
|||
String msg = "503 Service Temporarily Unavailable on POST request for \"https://example.com/api/auth/login\""; |
|||
|
|||
assertThat(ServiceFailureNotification.linkifyRequestUrl(msg)) |
|||
.isEqualTo("503 Service Temporarily Unavailable on POST <https://example.com/api/auth/login|request>"); |
|||
} |
|||
|
|||
@Test |
|||
void linkifyReplacesRequestConnectToUrlFailed() { |
|||
String msg = "I/O error on POST request: Connect to https://example.com:443 failed: Connect timed out"; |
|||
|
|||
assertThat(ServiceFailureNotification.linkifyRequestUrl(msg)) |
|||
.isEqualTo("I/O error on POST <https://example.com:443|request>: Connect timed out"); |
|||
} |
|||
|
|||
@Test |
|||
void linkifyLeavesMessagesWithoutRequestUrlUntouched() { |
|||
String msg = "Connection refused"; |
|||
assertThat(ServiceFailureNotification.linkifyRequestUrl(msg)).isEqualTo(msg); |
|||
} |
|||
|
|||
@Test |
|||
void linkifyHandlesNull() { |
|||
assertThat(ServiceFailureNotification.linkifyRequestUrl(null)).isNull(); |
|||
} |
|||
|
|||
@Test |
|||
void shortNameUsesShortNameProviderWhenAvailable() { |
|||
ShortNameProvider provider = () -> "MQTT"; |
|||
assertThat(ServiceFailureNotification.shortName(provider)).isEqualTo("MQTT"); |
|||
} |
|||
|
|||
@Test |
|||
void shortNameFallsBackToToStringForOtherKeys() { |
|||
Object key = new Object() { |
|||
@Override public String toString() { return "LOGIN"; } |
|||
}; |
|||
assertThat(ServiceFailureNotification.shortName(key)).isEqualTo("LOGIN"); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,183 @@ |
|||
/** |
|||
* Copyright © 2016-2026 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.monitoring.notification.incident; |
|||
|
|||
import org.junit.jupiter.api.AfterEach; |
|||
import org.junit.jupiter.api.BeforeEach; |
|||
import org.junit.jupiter.api.Test; |
|||
import org.thingsboard.monitoring.data.notification.AffectedService; |
|||
|
|||
import java.time.Duration; |
|||
import java.util.List; |
|||
import java.util.concurrent.atomic.AtomicInteger; |
|||
|
|||
import static org.assertj.core.api.Assertions.assertThat; |
|||
|
|||
class IncidentManagerTest { |
|||
|
|||
private RecordingTransport transport; |
|||
private IncidentManager manager; |
|||
|
|||
@BeforeEach |
|||
void setUp() { |
|||
transport = new RecordingTransport(); |
|||
manager = new IncidentManager(transport, 3600L, "tbqa", false); |
|||
} |
|||
|
|||
@AfterEach |
|||
void tearDown() { |
|||
manager.shutdown(); |
|||
} |
|||
|
|||
@Test |
|||
void formatDurationRendersMinutesAndHours() { |
|||
assertThat(IncidentManager.formatDuration(Duration.ofSeconds(30))).isEqualTo("0m"); |
|||
assertThat(IncidentManager.formatDuration(Duration.ofMinutes(5))).isEqualTo("5m"); |
|||
assertThat(IncidentManager.formatDuration(Duration.ofMinutes(59))).isEqualTo("59m"); |
|||
assertThat(IncidentManager.formatDuration(Duration.ofMinutes(60))).isEqualTo("1h"); |
|||
assertThat(IncidentManager.formatDuration(Duration.ofMinutes(75))).isEqualTo("1h15m"); |
|||
assertThat(IncidentManager.formatDuration(Duration.ofMinutes(120))).isEqualTo("2h"); |
|||
} |
|||
|
|||
@Test |
|||
void firstFailureOpensIncidentAndPostsHeaderAndReply() { |
|||
manager.sendAlert("CoAP failure message", |
|||
List.of(AffectedService.failing("CoAP", 1))); |
|||
|
|||
assertThat(transport.incidents).hasSize(1); |
|||
assertThat(transport.incidents.get(0)).contains(":rotating_light:").contains(":red_circle: CoAP (1)"); |
|||
assertThat(transport.replies).hasSize(1); |
|||
assertThat(transport.replies.get(0).text()).isEqualTo("CoAP failure message"); |
|||
} |
|||
|
|||
@Test |
|||
void isolatedRecoveryWithoutActiveIncidentIsIgnored() { |
|||
manager.sendAlert("Login is OK", |
|||
List.of(AffectedService.recovered("Login"))); |
|||
|
|||
assertThat(transport.incidents).isEmpty(); |
|||
assertThat(transport.replies).isEmpty(); |
|||
assertThat(transport.updates).isEmpty(); |
|||
} |
|||
|
|||
@Test |
|||
void subsequentFailureUpdatesHeaderAndPostsReply() { |
|||
manager.sendAlert("CoAP failure", List.of(AffectedService.failing("CoAP", 1))); |
|||
manager.sendAlert("CoAP repeat", List.of(AffectedService.failing("CoAP", 3))); |
|||
|
|||
assertThat(transport.incidents).hasSize(1); |
|||
assertThat(transport.replies).hasSize(2); |
|||
assertThat(transport.updates).hasSize(1); |
|||
assertThat(transport.updates.get(0).text()).contains(":red_circle: CoAP (3)"); |
|||
} |
|||
|
|||
@Test |
|||
void recoveryAfterFailureMovesServiceToGreenAndKeepsFailureCount() { |
|||
manager.sendAlert("CoAP failure", List.of(AffectedService.failing("CoAP", 4))); |
|||
manager.sendAlert("CoAP is OK", List.of(AffectedService.recovered("CoAP"))); |
|||
|
|||
assertThat(transport.updates).hasSize(1); |
|||
String updated = transport.updates.get(0).text(); |
|||
assertThat(updated).contains(":large_green_circle: CoAP (4)").doesNotContain(":red_circle:"); |
|||
} |
|||
|
|||
@Test |
|||
void highLatencyIsTrackedAsYellow() { |
|||
manager.sendAlert("high latency", |
|||
List.of(AffectedService.highLatency("logInLatency"))); |
|||
|
|||
assertThat(transport.incidents.get(0)).contains(":large_yellow_circle: logInLatency"); |
|||
} |
|||
|
|||
@Test |
|||
void repeatingSameFailureCountDoesNotTriggerRedundantUpdate() { |
|||
manager.sendAlert("CoAP failure", List.of(AffectedService.failing("CoAP", 3))); |
|||
manager.sendAlert("CoAP still failing", List.of(AffectedService.failing("CoAP", 3))); |
|||
|
|||
assertThat(transport.updates).isEmpty(); |
|||
assertThat(transport.replies).hasSize(2); |
|||
} |
|||
|
|||
@Test |
|||
void fullLifecycleStartFailRecoverResolve() { |
|||
manager.sendAlert("Login failure", List.of(AffectedService.failing("Login", 1))); |
|||
manager.sendAlert("WS failure", List.of(AffectedService.failing("WS Connect", 1))); |
|||
manager.sendAlert("Login is OK", List.of(AffectedService.recovered("Login"))); |
|||
|
|||
assertThat(transport.incidents).hasSize(1); |
|||
|
|||
manager.resolveIncident(); |
|||
|
|||
assertThat(transport.updates).last() |
|||
.extracting(RecordingTransport.Message::text) |
|||
.asString() |
|||
.contains(":white_check_mark:") |
|||
.contains(":red_circle: WS Connect") |
|||
.contains(":large_green_circle: Login (1)"); |
|||
} |
|||
|
|||
@Test |
|||
void resolveWithoutActiveIncidentIsNoOp() { |
|||
manager.resolveIncident(); |
|||
assertThat(transport.updates).isEmpty(); |
|||
} |
|||
|
|||
@Test |
|||
void doesNotAutoResolveWhileServicesAreStillFailing() throws Exception { |
|||
manager.shutdown(); |
|||
transport = new RecordingTransport(); |
|||
manager = new IncidentManager(transport, 1L, "tbqa", false); |
|||
|
|||
manager.sendAlert("CoAP failure", List.of(AffectedService.failing("CoAP", 1))); |
|||
Thread.sleep(1500); |
|||
|
|||
assertThat(transport.updates) |
|||
.extracting(RecordingTransport.Message::text) |
|||
.noneMatch(t -> t.contains(":white_check_mark:")); |
|||
|
|||
manager.sendAlert("CoAP is OK", List.of(AffectedService.recovered("CoAP"))); |
|||
Thread.sleep(1500); |
|||
|
|||
assertThat(transport.updates) |
|||
.extracting(RecordingTransport.Message::text) |
|||
.anyMatch(t -> t.contains(":white_check_mark:")); |
|||
} |
|||
|
|||
private static class RecordingTransport implements IncidentTransport { |
|||
private final AtomicInteger threadCounter = new AtomicInteger(); |
|||
final java.util.List<String> incidents = new java.util.ArrayList<>(); |
|||
final java.util.List<Message> replies = new java.util.ArrayList<>(); |
|||
final java.util.List<Message> updates = new java.util.ArrayList<>(); |
|||
|
|||
@Override |
|||
public String postIncident(String text) { |
|||
incidents.add(text); |
|||
return "thread-" + threadCounter.incrementAndGet(); |
|||
} |
|||
|
|||
@Override |
|||
public void postThreadReply(String threadId, String text) { |
|||
replies.add(new Message(threadId, text)); |
|||
} |
|||
|
|||
@Override |
|||
public void updateIncident(String threadId, String text) { |
|||
updates.add(new Message(threadId, text)); |
|||
} |
|||
|
|||
record Message(String threadId, String text) {} |
|||
} |
|||
} |
|||
Loading…
Reference in new issue