Skip to content

Refactor http-based alarm plugins and extract common logic to HttpAlarmCallback #10401

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

Merged
kezhenxu94 merged 1 commit into from
Feb 17, 2023
Merged

Refactor http-based alarm plugins and extract common logic to HttpAlarmCallback #10401

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
2 changes: 1 addition & 1 deletion dist-material/release-docs/LICENSE
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -316,7 +316,7 @@ The text of each license is the standard Apache 2.0 license.
https://mvnrepository.com/artifact/org.apache.curator/curator-recipes/4.3.0 Apache-2.0
https://mvnrepository.com/artifact/org.apache.curator/curator-x-discovery/4.3.0 Apache-2.0
https://mvnrepository.com/artifact/org.apache.httpcomponents/httpasyncclient/4.1.3 Apache-2.0
https://mvnrepository.com/artifact/org.apache.httpcomponents/httpclient/4.5.13 Apache-2.0
https://mvnrepository.com/artifact/org.apache.httpcomponents/httpclient/4.5.3 Apache-2.0
https://mvnrepository.com/artifact/org.apache.httpcomponents/httpcore/4.4.13 Apache-2.0
https://mvnrepository.com/artifact/org.apache.httpcomponents/httpcore-nio/4.4.13 Apache-2.0
https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients/2.8.1 Apache-2.0
Expand Down
1 change: 1 addition & 0 deletions docs/en/changes/changes.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,6 +94,7 @@
* Refactor `@Column` annotation, swap `Column#name` and `ElasticSearch.Column#columnAlias` and rename `ElasticSearch.Column#columnAlias` to `ElasticSearch.Column#legacyName`.
* Add Python HTTPX module component ID(7019).
* Migrate tests from junit 4 to junit 5.
* Refactor http-based alarm plugins and extract common logic to `HttpAlarmCallback`.

#### UI

Expand Down
4 changes: 0 additions & 4 deletions oap-server/server-alarm-plugin/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,10 +52,6 @@
<groupId>org.mvel</groupId>
<artifactId>mvel2</artifactId>
</dependency>
<dependency>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpclient</artifactId>
</dependency>
<dependency>
<groupId>com.linecorp.armeria</groupId>
<artifactId>armeria-junit5</artifactId>
Expand Down
16 changes: 9 additions & 7 deletions ...-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmCore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,19 +18,19 @@

package org.apache.skywalking.oap.server.core.alarm.provider;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import org.apache.skywalking.oap.server.core.alarm.AlarmCallback;
import org.apache.skywalking.oap.server.core.alarm.AlarmMessage;
import org.joda.time.LocalDateTime;
import org.joda.time.Minutes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
* Alarm core includes metrics values in certain time windows based on alarm settings. By using its internal timer
* trigger and the alarm rules to decide whether send the alarm to database and webhook(s)
Expand Down Expand Up @@ -82,7 +82,9 @@ public void start(List<AlarmCallback> allCallbacks) {
}
List<AlarmMessage> filteredMessages = alarmMessageList.stream().filter(msg -> !msg.isOnlyAsCondition()).collect(Collectors.toList());
if (!filteredMessages.isEmpty()) {
allCallbacks.forEach(callback -> callback.doAlarm(filteredMessages));
for (AlarmCallback callback : allCallbacks) {
callback.doAlarm(filteredMessages);
}
Comment thread
wu-sheng marked this conversation as resolved.
}
}
} catch (Exception e) {
Expand Down
86 changes: 14 additions & 72 deletions ...n/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/WebhookCallback.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,91 +19,33 @@
package org.apache.skywalking.oap.server.core.alarm.provider;

import com.google.gson.Gson;
import io.netty.handler.codec.http.HttpHeaderValues;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.nio.charset.StandardCharsets;
import java.util.List;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.HttpHeaders;
import org.apache.http.HttpStatus;
import org.apache.http.StatusLine;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.skywalking.oap.server.core.alarm.AlarmCallback;
import org.apache.skywalking.oap.server.core.alarm.AlarmMessage;
import org.apache.skywalking.oap.server.core.alarm.HttpAlarmCallback;

import java.io.IOException;
import java.net.URI;
import java.util.List;
import java.util.Map;

/**
* Use SkyWalking alarm webhook API calls a remote endpoints.
*/
@Slf4j
public class WebhookCallback implements AlarmCallback {
private static final int HTTP_CONNECT_TIMEOUT = 1000;
private static final int HTTP_CONNECTION_REQUEST_TIMEOUT = 1000;
private static final int HTTP_SOCKET_TIMEOUT = 10000;

private AlarmRulesWatcher alarmRulesWatcher;
private RequestConfig requestConfig;
private Gson gson = new Gson();

public WebhookCallback(AlarmRulesWatcher alarmRulesWatcher) {
this.alarmRulesWatcher = alarmRulesWatcher;
requestConfig = RequestConfig.custom()
.setConnectTimeout(HTTP_CONNECT_TIMEOUT)
.setConnectionRequestTimeout(HTTP_CONNECTION_REQUEST_TIMEOUT)
.setSocketTimeout(HTTP_SOCKET_TIMEOUT)
.build();
}
@RequiredArgsConstructor
public class WebhookCallback extends HttpAlarmCallback {
private final AlarmRulesWatcher alarmRulesWatcher;
private final Gson gson = new Gson();

@Override
public void doAlarm(List<AlarmMessage> alarmMessage) {
public void doAlarm(List<AlarmMessage> alarmMessage) throws IOException, InterruptedException {
if (alarmRulesWatcher.getWebHooks().isEmpty()) {
return;
}

CloseableHttpClient httpClient = HttpClients.custom().build();
try {
alarmRulesWatcher.getWebHooks().forEach(url -> {
HttpPost post = new HttpPost(url);
post.setConfig(requestConfig);
post.setHeader(HttpHeaders.ACCEPT, HttpHeaderValues.APPLICATION_JSON.toString());
post.setHeader(HttpHeaders.CONTENT_TYPE, HttpHeaderValues.APPLICATION_JSON.toString());

StringEntity entity;
CloseableHttpResponse httpResponse = null;
try {
entity = new StringEntity(gson.toJson(alarmMessage), StandardCharsets.UTF_8);
post.setEntity(entity);
httpResponse = httpClient.execute(post);
StatusLine statusLine = httpResponse.getStatusLine();
if (statusLine != null && statusLine.getStatusCode() != HttpStatus.SC_OK) {
log.error("send alarm to " + url + " failure. Response code: " + statusLine.getStatusCode());
}
} catch (UnsupportedEncodingException e) {
log.error("Alarm to JSON error, " + e.getMessage(), e);
} catch (IOException e) {
log.error("send alarm to " + url + " failure.", e);
} finally {
if (httpResponse != null) {
try {
httpResponse.close();
} catch (IOException e) {
log.error(e.getMessage(), e);
}

}
}
});
} finally {
try {
httpClient.close();
} catch (IOException e) {
log.error(e.getMessage(), e);
}
for (String url : alarmRulesWatcher.getWebHooks()) {
post(URI.create(url), gson.toJson(alarmMessage), Map.of());
}
}
}
106 changes: 24 additions & 82 deletions ...a/org/apache/skywalking/oap/server/core/alarm/provider/dingtalk/DingtalkHookCallback.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,77 +18,50 @@

package org.apache.skywalking.oap.server.core.alarm.provider.dingtalk;

import io.netty.handler.codec.http.HttpHeaderValues;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.HttpHeaders;
import org.apache.http.HttpStatus;
import org.apache.http.StatusLine;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.ContentType;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;
import org.apache.skywalking.oap.server.library.util.StringUtil;
import org.apache.skywalking.oap.server.core.alarm.AlarmCallback;
import org.apache.skywalking.oap.server.core.alarm.AlarmMessage;
import org.apache.skywalking.oap.server.core.alarm.HttpAlarmCallback;
import org.apache.skywalking.oap.server.core.alarm.provider.AlarmRulesWatcher;
import org.apache.skywalking.oap.server.library.util.StringUtil;

import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.URI;
import java.net.URLEncoder;
import java.nio.charset.StandardCharsets;
import java.security.InvalidKeyException;
import java.security.NoSuchAlgorithmException;
import java.util.Base64;
import java.util.List;
import java.util.Map;

/**
* Use SkyWalking alarm dingtalk webhook API.
*/
@Slf4j
public class DingtalkHookCallback implements AlarmCallback {

private static final int HTTP_CONNECT_TIMEOUT = 1000;
private static final int HTTP_CONNECTION_REQUEST_TIMEOUT = 1000;
private static final int HTTP_SOCKET_TIMEOUT = 10000;
private AlarmRulesWatcher alarmRulesWatcher;
private RequestConfig requestConfig;

public DingtalkHookCallback(final AlarmRulesWatcher alarmRulesWatcher) {
this.alarmRulesWatcher = alarmRulesWatcher;
this.requestConfig = RequestConfig.custom()
.setConnectTimeout(HTTP_CONNECT_TIMEOUT)
.setConnectionRequestTimeout(HTTP_CONNECTION_REQUEST_TIMEOUT)
.setSocketTimeout(HTTP_SOCKET_TIMEOUT)
.build();
}
@RequiredArgsConstructor
public class DingtalkHookCallback extends HttpAlarmCallback {
private final AlarmRulesWatcher alarmRulesWatcher;

/**
* Send alarm message if the settings not empty
*/
@Override
public void doAlarm(List<AlarmMessage> alarmMessages) {
if (this.alarmRulesWatcher.getDingtalkSettings() == null || this.alarmRulesWatcher.getDingtalkSettings().getWebhooks().isEmpty()) {
public void doAlarm(List<AlarmMessage> alarmMessages) throws Exception {
if (alarmRulesWatcher.getDingtalkSettings() == null || alarmRulesWatcher.getDingtalkSettings().getWebhooks().isEmpty()) {
return;
}
try (CloseableHttpClient httpClient = HttpClients.custom().build()) {
DingtalkSettings dingtalkSettings = this.alarmRulesWatcher.getDingtalkSettings();
dingtalkSettings.getWebhooks().forEach(webHookUrl -> {
String url = getUrl(webHookUrl);
alarmMessages.forEach(alarmMessage -> {
String requestBody = String.format(
this.alarmRulesWatcher.getDingtalkSettings().getTextTemplate(), alarmMessage.getAlarmMessage()
);
sendAlarmMessage(httpClient, url, requestBody);
});
});
} catch (IOException e) {
log.error(e.getMessage(), e);
final var dingtalkSettings = alarmRulesWatcher.getDingtalkSettings();
for (final var webHookUrl : dingtalkSettings.getWebhooks()) {
final var url = getUrl(webHookUrl);
for (final var alarmMessage : alarmMessages) {
final var requestBody = String.format(
alarmRulesWatcher.getDingtalkSettings().getTextTemplate(), alarmMessage.getAlarmMessage()
);
post(URI.create(url), requestBody, Map.of());
}
}
}

Expand All @@ -107,7 +80,7 @@ private String getUrl(DingtalkSettings.WebHookUrl webHookUrl) {
*/
private String getSignUrl(DingtalkSettings.WebHookUrl webHookUrl) {
try {
Long timestamp = System.currentTimeMillis();
final var timestamp = System.currentTimeMillis();
return String.format("%s&timestamp=%s&sign=%s", webHookUrl.getUrl(), timestamp, sign(timestamp, webHookUrl.getSecret()));
} catch (NoSuchAlgorithmException | UnsupportedEncodingException | InvalidKeyException e) {
throw new RuntimeException(e);
Expand All @@ -118,42 +91,11 @@ private String getSignUrl(DingtalkSettings.WebHookUrl webHookUrl) {
* Sign webhook url using HmacSHA256 algorithm
*/
private String sign(final Long timestamp, String secret) throws NoSuchAlgorithmException, UnsupportedEncodingException, InvalidKeyException {
String stringToSign = timestamp + "\n" + secret;
Mac mac = Mac.getInstance("HmacSHA256");
final var stringToSign = timestamp + "\n" + secret;
final var mac = Mac.getInstance("HmacSHA256");
mac.init(new SecretKeySpec(secret.getBytes(StandardCharsets.UTF_8), "HmacSHA256"));
byte[] signData = mac.doFinal(stringToSign.getBytes(StandardCharsets.UTF_8));
return URLEncoder.encode(new String(Base64.getEncoder().encode(signData)), StandardCharsets.UTF_8.name());
final var signData = mac.doFinal(stringToSign.getBytes(StandardCharsets.UTF_8));
return URLEncoder.encode(new String(Base64.getEncoder().encode(signData)), StandardCharsets.UTF_8);
}

/**
* Send alarm message to remote endpoint
*/
private void sendAlarmMessage(CloseableHttpClient httpClient, String url, String requestBody) {
CloseableHttpResponse httpResponse = null;
try {
HttpPost post = new HttpPost(url);
post.setConfig(requestConfig);
post.setHeader(HttpHeaders.ACCEPT, HttpHeaderValues.APPLICATION_JSON.toString());
post.setHeader(HttpHeaders.CONTENT_TYPE, HttpHeaderValues.APPLICATION_JSON.toString());
StringEntity entity = new StringEntity(requestBody, ContentType.APPLICATION_JSON);
post.setEntity(entity);
httpResponse = httpClient.execute(post);
StatusLine statusLine = httpResponse.getStatusLine();
if (statusLine != null && statusLine.getStatusCode() != HttpStatus.SC_OK) {
log.error("send dingtalk alarm to {} failure. Response code: {}, Response content: {}", url, statusLine.getStatusCode(),
EntityUtils.toString(httpResponse.getEntity()));
}
} catch (Throwable e) {
log.error("send dingtalk alarm to {} failure.", url, e);
} finally {
if (httpResponse != null) {
try {
httpResponse.close();
} catch (IOException e) {
log.error(e.getMessage(), e);
}

}
}
}
}
Loading